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

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

def for_cluster?
  @changes_for == CLUSTER
end

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

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

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

def for_database?
  @changes_for == DATABASE
end

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

attr_reader :options

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

def resuming?
  !!@resuming
end

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

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 237

def close
  unless closed?
    begin
      @cursor.close
    rescue Error::OperationFailure, Error::SocketError, Error::SocketTimeoutError, Error::MissingConnection
      # ignore
    end
    @cursor = nil
  end
end

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

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

  session = client.send(:get_session, @options)
  start_at_operation_time = nil
  start_at_operation_time_supported = nil
  @cursor = read_with_retry_cursor(session, server_selector, view) do |server|
    server.with_connection do |connection|
      start_at_operation_time_supported = connection.description.server_version_gte?('4.0')

      result = send_initial_query(connection, session)
      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 161

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 268

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 331

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

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

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 393

def send_initial_query(connection, session)
  initial_query_op(session, view.read_preference)
    .execute_with_connection(
      connection,
      context: Operation::Context.new(client: client, session: session),
    )
end

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

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 213

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 183

def try_next
  raise StopIteration.new if closed?
  begin
    doc = @cursor.try_next
  rescue Mongo::Error => e
    if !e.change_stream_resumable?
      raise
    end

    # Rerun initial aggregation.
    # Any errors here will stop iteration and break out of this
    # method.

    # Save cursor's resume token so we can use it
    # to create a new cursor
    @resume_token = @cursor.resume_token

    close
    create_cursor!
    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