class Mongo::Collection::View::ChangeStream

Provides behavior around a `$changeStream` pipeline stage in the aggregation framework. Specifying this stage allows users to request that notifications are sent for all changes to a particular collection or database.

@note Only available in server versions 3.6 and higher. @note ChangeStreams do not work properly with JRuby because of the

issue documented here: https://github.com/jruby/jruby/issues/4212.
Namely, JRuby eagerly evaluates #next on an Enumerator in a background
green thread, therefore calling #next on the change stream will cause
getMores to be called in a loop in the background.

@since 2.5.0

Constants

CLUSTER

@return [ Symbol ] Used to indicate that the change stream should listen for changes on

the entire cluster rather than just the collection.

@since 2.6.0

DATABASE

@return [ Symbol ] Used to indicate that the change stream should listen for changes on

the entire database rather than just the collection.

@since 2.6.0

FULL_DOCUMENT_DEFAULT

@return [ String ] The fullDocument option default value.

@since 2.5.0

Attributes

options[R]

@return [ BSON::Document ] The change stream options.

@since 2.5.0

Public Class Methods

new(view, pipeline, changes_for, options = {}) click to toggle source

Initialize the change stream for the provided collection view, pipeline and options.

@example Create the new change stream view.

ChangeStream.new(view, pipeline, options)

@param [ Collection::View ] view The collection view. @param [ Array<Hash> ] pipeline The pipeline of operators to filter the change notifications. @param [ Hash ] options The change stream options.

@option options [ String ] :full_document Allowed values: 'default', 'updateLookup'. Defaults to 'default'.

When set to 'updateLookup', the change notification for partial updates will include both a delta
describing the changes to the document, as well as a copy of the entire document that was changed
from some time after the change occurred.

@option options [ BSON::Document, Hash ] :resume_after Specifies the logical starting point for the

new change stream.

@option options [ Integer ] :max_await_time_ms The maximum amount of time for the server to wait

on new documents to satisfy a change stream query.

@option options [ Integer ] :batch_size The number of documents to return per batch. @option options [ BSON::Document, Hash ] :collation The collation to use. @option options [ BSON::Timestamp ] :start_at_operation_time Only

return changes that occurred at or after the specified timestamp. Any
command run against the server will return a cluster time that can
be used here. Only recognized by server versions 4.0+.

@option options [ Bson::Document, Hash ] :start_after Similar to :resume_after, this

option takes a resume token and starts a new change stream returning the first
notification after the token. This will allow users to watch collections that have been
dropped and recreated or newly renamed collections without missing any notifications.

The server will report an error if `startAfter` and `resumeAfter` are both specified.

@since 2.5.0

# File lib/mongo/collection/view/change_stream.rb, line 92
def initialize(view, pipeline, changes_for, options = {})
  @view = view
  @changes_for = changes_for
  @change_stream_filters = pipeline && pipeline.dup
  @options = options && options.dup.freeze
  @start_after = @options[:start_after]

  # The resume token tracked by the change stream, used only
  # when there is no cursor, or no cursor resume token
  @resume_token = @start_after || @options[:resume_after]

  create_cursor!

  # We send different parameters when we resume a change stream
  # compared to when we send the first query
  @resuming = true
end

Public Instance Methods

close() click to toggle source

Close the change stream.

@example Close the change stream.

stream.close

@return [ nil ] nil.

@since 2.5.0

# File lib/mongo/collection/view/change_stream.rb, line 200
def close
  unless closed?
    begin; @cursor.send(:kill_cursors); rescue; end
    @cursor = nil
  end
end
closed?() click to toggle source

Is the change stream closed?

@example Determine whether the change stream is closed.

stream.closed?

@return [ true, false ] If the change stream is closed.

@since 2.5.0

# File lib/mongo/collection/view/change_stream.rb, line 215
def closed?
  @cursor.nil?
end
each() { |document| ... } click to toggle source

Iterate through documents returned by the change stream.

This method retries once per error on resumable errors (two consecutive errors result in the second error being raised, an error which is recovered from resets the error count to zero).

@example Iterate through the stream of documents.

stream.each do |document|
  p document
end

@return [ Enumerator ] The enumerator.

@since 2.5.0

@yieldparam [ BSON::Document ] Each change stream document.

# File lib/mongo/collection/view/change_stream.rb, line 126
def each
  raise StopIteration.new if closed?
  loop do
    document = try_next
    yield document if document
  end
rescue StopIteration => e
  return self
end
inspect() click to toggle source

Get a formatted string for use in inspection.

@example Inspect the change stream object.

stream.inspect

@return [ String ] The change stream inspection.

@since 2.5.0

# File lib/mongo/collection/view/change_stream.rb, line 227
def inspect
  "#<Mongo::Collection::View:ChangeStream:0x#{object_id} filters=#{@change_stream_filters} " +
    "options=#{@options} resume_token=#{resume_token}>"
end
resume_token() click to toggle source

Returns the resume token that the stream will use to automatically resume, if one exists.

@example Get the change stream resume token.

stream.resume_token

@return [ BSON::Document | nil ] The change stream resume token.

@since 2.10.0

# File lib/mongo/collection/view/change_stream.rb, line 241
def resume_token
  cursor_resume_token = @cursor.resume_token if @cursor
  cursor_resume_token || @resume_token
end
to_enum() click to toggle source
Calls superclass method
# File lib/mongo/collection/view/change_stream.rb, line 181
def to_enum
  enum = super
  enum.send(:instance_variable_set, '@obj', self)
  class << enum
    def try_next
      @obj.try_next
    end
  end
  enum
end
try_next() click to toggle source

Return one document from the change stream, if one is available.

Retries once on a resumable error.

Raises StopIteration if the change stream is closed.

This method will wait up to max_await_time_ms milliseconds for changes from the server, and if no changes are received it will return nil.

@return [ BSON::Document | nil ] A change stream document. @since 2.6.0

# File lib/mongo/collection/view/change_stream.rb, line 148
def try_next
  raise StopIteration.new if closed?
  retried = false

  begin
    doc = @cursor.try_next
  rescue Mongo::Error => e
    if retried || !e.change_stream_resumable?
      raise
    end

    retried = true
    # 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

Private Instance Methods

aggregate_spec(session) click to toggle source
# File lib/mongo/collection/view/change_stream.rb, line 292
def aggregate_spec(session)
  super(session).tap do |spec|
    spec[:selector][:aggregate] = 1 unless for_collection?
  end
end
change_doc() click to toggle source
# File lib/mongo/collection/view/change_stream.rb, line 298
def change_doc
  {}.tap do |doc|
    if @options[:full_document]
      doc[:fullDocument] = @options[:full_document]
    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
create_cursor!() click to toggle source
# File lib/mongo/collection/view/change_stream.rb, line 260
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|
    start_at_operation_time_supported = server.description.server_version_gte?('4.0')

    result = send_initial_query(server, 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
  @start_at_operation_time = start_at_operation_time
  @start_at_operation_time_supported = start_at_operation_time_supported
end
for_cluster?() click to toggle source
# File lib/mongo/collection/view/change_stream.rb, line 248
def for_cluster?
  @changes_for == CLUSTER
end
for_collection?() click to toggle source
# File lib/mongo/collection/view/change_stream.rb, line 256
def for_collection?
  !for_cluster? && !for_database?
end
for_database?() click to toggle source
# File lib/mongo/collection/view/change_stream.rb, line 252
def for_database?
  @changes_for == DATABASE
end
pipeline() click to toggle source
# File lib/mongo/collection/view/change_stream.rb, line 288
def pipeline
  [{ '$changeStream' => change_doc }] + @change_stream_filters
end
resuming?() click to toggle source
# File lib/mongo/collection/view/change_stream.rb, line 357
def resuming?
  !!@resuming
end
send_initial_query(server, session) click to toggle source
# File lib/mongo/collection/view/change_stream.rb, line 342
def send_initial_query(server, session)
  initial_query_op(session).execute(server)
end
time_to_bson_timestamp(time) click to toggle source
# File lib/mongo/collection/view/change_stream.rb, line 346
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