Class: Mongo::Collection::View::ChangeStream
- Inherits:
-
Aggregation
- Object
- Aggregation
- Mongo::Collection::View::ChangeStream
- Includes:
- Retryable
- Defined in:
- build/ruby-driver-master/lib/mongo/collection/view/change_stream.rb,
build/ruby-driver-master/lib/mongo/collection/view/change_stream/retryable.rb
Overview
Only available in server versions 3.6 and higher.
ChangeStreams do not work properly with JRuby because of the issue documented here: 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.
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.
Defined Under Namespace
Modules: Retryable
Constant Summary collapse
- FULL_DOCUMENT_DEFAULT =
Returns The fullDocument option default value.
'default'.freeze
- DATABASE =
Returns Used to indicate that the change stream should listen for changes on the entire database rather than just the collection.
:database
- CLUSTER =
Returns Used to indicate that the change stream should listen for changes on the entire cluster rather than just the collection.
:cluster
Constants inherited from Aggregation
Constants included from Loggable
Constants included from Explainable
Explainable::ALL_PLANS_EXECUTION, Explainable::EXECUTION_STATS, Explainable::QUERY_PLANNER
Instance Attribute Summary collapse
-
#options ⇒ BSON::Document
readonly
The change stream options.
Attributes inherited from Aggregation
Attributes included from Iterable
Instance Method Summary collapse
-
#close ⇒ nil
Close the change stream.
-
#closed? ⇒ true, false
Is the change stream closed?.
-
#each {|Each| ... } ⇒ Enumerator
Iterate through documents returned by the change stream.
-
#initialize(view, pipeline, changes_for, options = {}) ⇒ ChangeStream
constructor
Initialize the change stream for the provided collection view, pipeline and options.
-
#inspect ⇒ String
Get a formatted string for use in inspection.
-
#resume_token ⇒ BSON::Document | nil
Returns the resume token that the stream will use to automatically resume, if one exists.
- #to_enum ⇒ Object
-
#try_next ⇒ BSON::Document | nil
Return one document from the change stream, if one is available.
Methods inherited from Aggregation
#allow_disk_use, #explain, #write?
Methods included from Retryable
#legacy_write_with_retry, #nro_write_with_retry, #read_with_one_retry, #read_with_retry, #read_with_retry_cursor, #write_with_retry
Methods included from Loggable
#log_debug, #log_error, #log_fatal, #log_info, #log_warn, #logger
Methods included from Explainable
Methods included from Iterable
Constructor Details
#initialize(view, pipeline, changes_for, options = {}) ⇒ ChangeStream
Initialize the change stream for the provided collection view, pipeline and options.
99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 |
# File 'build/ruby-driver-master/lib/mongo/collection/view/change_stream.rb', line 99 def initialize(view, pipeline, changes_for, = {}) @view = view @changes_for = changes_for @change_stream_filters = pipeline && pipeline.dup @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 |
Instance Attribute Details
#options ⇒ BSON::Document (readonly)
Returns The change stream options.
61 62 63 |
# File 'build/ruby-driver-master/lib/mongo/collection/view/change_stream.rb', line 61 def @options end |
Instance Method Details
#close ⇒ nil
This method attempts to close the cursor used by the change stream, which in turn closes the server-side change stream cursor. This method ignores any errors that occur when closing the server-side cursor.
Close the change stream.
209 210 211 212 213 214 215 216 217 218 |
# File 'build/ruby-driver-master/lib/mongo/collection/view/change_stream.rb', line 209 def close unless closed? begin @cursor.close rescue Error::OperationFailure, Error::SocketError, Error::SocketTimeoutError # ignore end @cursor = nil end end |
#closed? ⇒ true, false
Is the change stream closed?
228 229 230 |
# File 'build/ruby-driver-master/lib/mongo/collection/view/change_stream.rb', line 228 def closed? @cursor.nil? end |
#each {|Each| ... } ⇒ Enumerator
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).
133 134 135 136 137 138 139 140 141 |
# File 'build/ruby-driver-master/lib/mongo/collection/view/change_stream.rb', line 133 def each raise StopIteration.new if closed? loop do document = try_next yield document if document end rescue StopIteration => e return self end |
#inspect ⇒ String
Get a formatted string for use in inspection.
240 241 242 243 |
# File 'build/ruby-driver-master/lib/mongo/collection/view/change_stream.rb', line 240 def inspect "#<Mongo::Collection::View:ChangeStream:0x#{object_id} filters=#{@change_stream_filters} " + "options=#{@options} resume_token=#{resume_token}>" end |
#resume_token ⇒ BSON::Document | nil
Returns the resume token that the stream will use to automatically resume, if one exists.
254 255 256 257 |
# File 'build/ruby-driver-master/lib/mongo/collection/view/change_stream.rb', line 254 def resume_token cursor_resume_token = @cursor.resume_token if @cursor cursor_resume_token || @resume_token end |
#to_enum ⇒ Object
185 186 187 188 189 190 191 192 193 194 |
# File 'build/ruby-driver-master/lib/mongo/collection/view/change_stream.rb', line 185 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 ⇒ BSON::Document | nil
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.
155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 |
# File 'build/ruby-driver-master/lib/mongo/collection/view/change_stream.rb', line 155 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 |