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

Inherits:
Aggregation
  • Object
show all
Includes:
Aggregation::Behavior, Retryable
Defined in:
lib/mongo/collection/view/change_stream.rb,
lib/mongo/collection/view/change_stream/retryable.rb

Overview

Note:

Only available in server versions 3.6 and higher.

Note:

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.

Since:

  • 2.5.0

Defined Under Namespace

Modules: Retryable

Constant Summary collapse

FULL_DOCUMENT_DEFAULT =

Returns The fullDocument option default value.

Returns:

  • (String) —

    The fullDocument option default value.

Since:

  • 2.5.0

'default'.freeze
DATABASE =

Returns Used to indicate that the change stream should listen for changes on the entire database rather than just the collection.

Returns:

  • (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

:database
CLUSTER =

Returns Used to indicate that the change stream should listen for changes on the entire cluster rather than just the collection.

Returns:

  • (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

:cluster

Constants included from Loggable

Loggable::PREFIX

Constants included from Explainable

Explainable::ALL_PLANS_EXECUTION, Explainable::EXECUTION_STATS, Explainable::QUERY_PLANNER

Instance Attribute Summary collapse

Attributes included from Aggregation::Behavior

#view

Instance Method Summary collapse

Methods included from Retryable

#read_worker, #select_server, #write_worker

Methods included from Aggregation::Behavior

#allow_disk_use, #explain, #timeout_ms, #write?

Methods included from Loggable

#log_debug, #log_error, #log_fatal, #log_info, #log_warn, #logger

Methods included from Explainable

#explain

Methods included from Iterable

#close_query

Methods included from Mongo::CursorHost

#validate_timeout_mode!

Constructor Details

#initialize(view, pipeline, changes_for, options = {}) ⇒ ChangeStream

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

Examples:

Create the new change stream view.

ChangeStream.new(view, pipeline, options)

Parameters:

  • view (Collection::View) —

    The collection view.

  • pipeline (Array<Hash>) —

    The pipeline of operators to filter the change notifications.

  • options (Hash) (defaults to: {}) —

    The change stream options.

Options Hash (options):

  • :full_document (String) —

    Allowed values: nil, ‘default’, ‘updateLookup’, ‘whenAvailable’, ‘required’.

    The default is to not send a value (i.e. nil), which is equivalent to ‘default’. By default, the change notification for partial updates will include a delta describing the changes to the document.

    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.

    When set to ‘whenAvailable’, configures the change stream to return the post-image of the modified document for replace and update change events if the post-image for this event is available.

    When set to ‘required’, the same behavior as ‘whenAvailable’ except that an error is raised if the post-image is not available.

  • :full_document_before_change (String) —

    Allowed values: nil, ‘whenAvailable’, ‘required’, ‘off’.

    The default is to not send a value (i.e. nil), which is equivalent to ‘off’.

    When set to ‘whenAvailable’, configures the change stream to return the pre-image of the modified document for replace, update, and delete change events if it is available.

    When set to ‘required’, the same behavior as ‘whenAvailable’ except that an error is raised if the pre-image is not available.

  • :resume_after (BSON::Document, Hash) —

    Specifies the logical starting point for the new change stream.

  • :max_await_time_ms (Integer) —

    The maximum amount of time for the server to wait on new documents to satisfy a change stream query.

  • :batch_size (Integer) —

    The number of documents to return per batch.

  • :collation (BSON::Document, Hash) —

    The collation to use.

  • :start_at_operation_time (BSON::Timestamp) —

    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+.

  • :start_after (Bson::Document, Hash) —

    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.

  • :comment (Object) —

    A user-provided comment to attach to this command.

  • :show_expanded_events (Boolean) —

    Enables the server to send the ‘expanded’ list of change stream events. The list of additional events included with this flag set are: createIndexes, dropIndexes, modify, create, shardCollection, reshardCollection, refineCollectionShardKey.

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

Since:

  • 2.5.0



133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
 
# File 'lib/mongo/collection/view/change_stream.rb', line 133

def initialize(view, pipeline, changes_for, options = {})
  # change stream cursors can only be :iterable, so we don't allow
  # timeout_mode to be specified.
  perform_setup(view, options, forbid: %i[ timeout_mode ]) do
    @changes_for = changes_for
    @change_stream_filters = pipeline && pipeline.dup
    @start_after = @options[:start_after]
  end

  # 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

#cursor ⇒ Cursor (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns the underlying cursor for this operation.

Returns:

  • (Cursor) —

    the underlying cursor for this operation

Since:

  • 2.5.0



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

def cursor
  @cursor
end

#options ⇒ BSON::Document (readonly)

Returns The change stream options.

Returns:

  • (BSON::Document) —

    The change stream options.

Since:

  • 2.5.0



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

def options
  @options
end

Instance Method Details

#close(opts = {}) ⇒ nil

Note:

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.

Examples:

Close the change stream.

stream.close

Returns:

  • (nil) —

    Always nil.

Since:

  • 2.5.0



254
255
256
257
258
259
260
261
262
263
 
# 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

#closed? ⇒ true, false

Is the change stream closed?

Examples:

Determine whether the change stream is closed.

stream.closed?

Returns:

  • (true, false) —

    If the change stream is closed.

Since:

  • 2.5.0



273
274
275
 
# File 'lib/mongo/collection/view/change_stream.rb', line 273

def closed?
  @cursor.nil?
end

#cursor_type ⇒ Object

“change streams are an abstraction around tailable-awaitData cursors…”

Returns:

  • :tailable_await

Since:

  • 2.5.0



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

def cursor_type
  :tailable_await
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).

Examples:

Iterate through the stream of documents.

stream.each do |document|
  p document
end

Yield Parameters:

  • Each (BSON::Document) —

    change stream document.

Returns:

  • (Enumerator) —

    The enumerator.

Since:

  • 2.5.0



169
170
171
172
173
174
175
176
177
 
# 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

#inspect ⇒ String

Get a formatted string for use in inspection.

Examples:

Inspect the change stream object.

stream.inspect

Returns:

  • (String) —

    The change stream inspection.

Since:

  • 2.5.0



285
286
287
288
 
# 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

#max_await_time_ms ⇒ Integer | nil

Returns the value of the max_await_time_ms option that was passed to this change stream.

Returns:

  • (Integer | nil) —

    the max_await_time_ms value

Since:

  • 2.5.0



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

def max_await_time_ms
  options[:max_await_time_ms]
end

#resume_token ⇒ BSON::Document | nil

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

Examples:

Get the change stream resume token.

stream.resume_token

Returns:

  • (BSON::Document | nil) —

    The change stream resume token.

Since:

  • 2.10.0



299
300
301
302
 
# 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

#timeout_mode ⇒ Object

“change streams…implicitly use ITERATION mode”

Returns:

  • :iteration

Since:

  • 2.5.0



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

def timeout_mode
  :iteration
end

#to_enum ⇒ Object

Since:

  • 2.5.0



227
228
229
230
231
232
233
234
235
236
 
# 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

#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.

Returns:

  • (BSON::Document | nil) —

    A change stream document.

Raises:

  • (StopIteration)

Since:

  • 2.6.0



191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
 
# 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