Class: Mongo::Collection::View::ChangeStream

# File 'lib/mongo/collection/view/change_stream.rb', line 67

attr_reader :cursor

# File 'lib/mongo/collection/view/change_stream.rb', line 328

def for_cluster?
  @changes_for == CLUSTER
end

# File 'lib/mongo/collection/view/change_stream.rb', line 336

def for_collection?
  !for_cluster? && !for_database?
end

# File 'lib/mongo/collection/view/change_stream.rb', line 332

def for_database?
  @changes_for == DATABASE
end

# File 'lib/mongo/collection/view/change_stream.rb', line 63

attr_reader :options

# File 'lib/mongo/collection/view/change_stream.rb', line 456

def resuming?
  !!@resuming
end

# File 'lib/mongo/collection/view/change_stream.rb', line 385

def change_doc
  {}.tap do |doc|
    if @options[:full_document]
      doc[:fullDocument] = @options[:full_document]
    end

    if @options[:full_document_before_change]
      doc[:fullDocumentBeforeChange] = @options[:full_document_before_change]
    end

    if @options.key?(:show_expanded_events)
      doc[:showExpandedEvents] = @options[:show_expanded_events]
    end

    if resuming?
      # We have a resume token once we retrieved any documents.
      # However, if the first getMore fails and the user didn't pass
      # a resume token we won't have a resume token to use.
      # Use start_at_operation time in this case
      if resume_token
        # Spec says we need to remove both startAtOperationTime and startAfter if
        # either was passed in by user, thus we won't forward them
        doc[:resumeAfter] = resume_token
      elsif @start_at_operation_time_supported && @start_at_operation_time
        # It is crucial to check @start_at_operation_time_supported
        # here - we may have switched to an older server that
        # does not support operation times and therefore shouldn't
        # try to send one to it!
        #
        # @start_at_operation_time is already a BSON::Timestamp
        doc[:startAtOperationTime] = @start_at_operation_time
      else
        # Can't resume if we don't have either
        raise Mongo::Error::MissingResumeToken
      end
    else
      if @start_after
        doc[:startAfter] = @start_after
      elsif resume_token
        doc[:resumeAfter] = resume_token
      end

      if options[:start_at_operation_time]
        doc[:startAtOperationTime] = time_to_bson_timestamp(
          options[:start_at_operation_time])
      end
    end

    doc[:allChangesForCluster] = true if for_cluster?
  end
end

# File 'lib/mongo/collection/view/change_stream.rb', line 254

def close(opts = {})
  unless closed?
    begin
      @cursor.close(opts)
    rescue Error::OperationFailure::Family, Error::SocketError, Error::SocketTimeoutError, Error::MissingConnection
      # ignore
    end
    @cursor = nil
  end
end

# File 'lib/mongo/collection/view/change_stream.rb', line 340

def create_cursor!(timeout_ms = nil)
  # clear the cache because we may get a newer or an older server
  # (rolling upgrades)
  @start_at_operation_time_supported = nil

  session = client.get_session(@options)
  context = Operation::Context.new(client: client, session: session, view: self, operation_timeouts: timeout_ms ? { operation_timeout_ms: timeout_ms } : operation_timeouts)

  start_at_operation_time = nil
  start_at_operation_time_supported = nil

  @cursor = read_with_retry_cursor(session, server_selector, self, context: context) do |server|
    server.with_connection do |connection|
      start_at_operation_time_supported = connection.description.server_version_gte?('4.0')

      result = send_initial_query(connection, context)

      if doc = result.replies.first && result.replies.first.documents.first
        start_at_operation_time = doc['operationTime']
      else
        # The above may set @start_at_operation_time to nil
        # if it was not in the document for some reason,
        # for consistency set it to nil here as well.
        # NB: since this block may be executed more than once, each
        # execution must write to start_at_operation_time either way.
        start_at_operation_time = nil
      end
      result
    end
  end

  @start_at_operation_time = start_at_operation_time
  @start_at_operation_time_supported = start_at_operation_time_supported
end

# File 'lib/mongo/collection/view/change_stream.rb', line 307

def cursor_type
  :tailable_await
end

# File 'lib/mongo/collection/view/change_stream.rb', line 169

def each
  raise StopIteration.new if closed?
  loop do
    document = try_next
    yield document if document
  end
rescue StopIteration
  return self
end

# File 'lib/mongo/collection/view/change_stream.rb', line 285

def inspect
  "#<Mongo::Collection::View:ChangeStream:0x#{object_id} filters=#{@change_stream_filters} " +
    "options=#{@options} resume_token=#{resume_token}>"
end

# File 'lib/mongo/collection/view/change_stream.rb', line 322

def max_await_time_ms
  options[:max_await_time_ms]
end

# File 'lib/mongo/collection/view/change_stream.rb', line 375

def pipeline
  [{ '$changeStream' => change_doc }] + @change_stream_filters
end

# File 'lib/mongo/collection/view/change_stream.rb', line 462

def recreate_cursor!(context = nil)
  @timed_out = false

  close
  create_cursor!(context&.remaining_timeout_ms)
end

# File 'lib/mongo/collection/view/change_stream.rb', line 299

def resume_token
  cursor_resume_token = @cursor.resume_token if @cursor
  cursor_resume_token || @resume_token
end

# File 'lib/mongo/collection/view/change_stream.rb', line 437

def send_initial_query(connection, context)
  initial_query_op(context.session, view.read_preference)
    .execute_with_connection(
      connection,
      context: context,
    )
end

# File 'lib/mongo/collection/view/change_stream.rb', line 445

def time_to_bson_timestamp(time)
  if time.is_a?(Time)
    seconds = time.to_f
    BSON::Timestamp.new(seconds.to_i, ((seconds - seconds.to_i) * 1000000).to_i)
  elsif time.is_a?(BSON::Timestamp)
    time
  else
    raise ArgumentError, 'Time must be a Time or a BSON::Timestamp instance'
  end
end

# File 'lib/mongo/collection/view/change_stream.rb', line 314

def timeout_mode
  :iteration
end

# File 'lib/mongo/collection/view/change_stream.rb', line 227

def to_enum
  enum = super
  enum.send(:instance_variable_set, '@obj', self)
  class << enum
    def try_next
      @obj.try_next
    end
  end
  enum
end

# File 'lib/mongo/collection/view/change_stream.rb', line 191

def try_next
  recreate_cursor! if @timed_out

  raise StopIteration.new if closed?

  begin
    doc = @cursor.try_next
  rescue Mongo::Error => e
    # "If a next call fails with a timeout error, drivers MUST NOT
    # invalidate the change stream. The subsequent next call MUST
    # perform a resume attempt to establish a new change stream on the
    # server..."
    #
    # However, SocketTimeoutErrors are TimeoutErrors, but are also
    # change-stream-resumable. To preserve existing (specified) behavior,
    # We only count timeouts when the error is not also
    # change-stream-resumable.
    @timed_out = e.is_a?(Mongo::Error::TimeoutError) && !e.change_stream_resumable?

    raise unless @timed_out || e.change_stream_resumable?

    @resume_token = @cursor.resume_token
    raise e if @timed_out

    recreate_cursor!(@cursor.context)
    retry
  end

  # We need to verify each doc has an _id, so we
  # have a resume token to work with
  if doc && doc['_id'].nil?
    raise Error::MissingResumeToken
  end
  doc
end