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 onthe 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 onthe 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
@return [ BSON::Document ] The change stream options.
@since 2.5.0
Public Class Methods
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 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
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
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
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
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
# 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
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
Mongo::Collection::View::Aggregation#aggregate_spec
# 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
# 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
# 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
# File lib/mongo/collection/view/change_stream.rb, line 248 def for_cluster? @changes_for == CLUSTER end
# File lib/mongo/collection/view/change_stream.rb, line 256 def for_collection? !for_cluster? && !for_database? end
# File lib/mongo/collection/view/change_stream.rb, line 252 def for_database? @changes_for == DATABASE end
# File lib/mongo/collection/view/change_stream.rb, line 288 def pipeline [{ '$changeStream' => change_doc }] + @change_stream_filters end
# File lib/mongo/collection/view/change_stream.rb, line 357 def resuming? !!@resuming end
# File lib/mongo/collection/view/change_stream.rb, line 342 def send_initial_query(server, session) initial_query_op(session).execute(server) end
# 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