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

Inherits:
Aggregation
  • Object
show all
Includes:
Retryable
Defined in:
build/ruby-driver-v2.19/lib/mongo/collection/view/change_stream.rb,
build/ruby-driver-v2.19/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 inherited from Aggregation

Aggregation::REROUTE

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 inherited from Aggregation

#view

Attributes included from Iterable

#cursor

Instance Method Summary collapse

Methods inherited from Aggregation

#allow_disk_use, #explain, #write?

Methods included from Retryable

#read_worker, #select_server, #write_worker

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

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



127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
 
# File 'build/ruby-driver-v2.19/lib/mongo/collection/view/change_stream.rb', line 127

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

Instance Attribute Details

#options ⇒ BSON::Document (readonly)

Returns The change stream options.

Returns:

  • (BSON::Document) —

    The change stream options.

Since:

  • 2.5.0



61
62
63
 
# File 'build/ruby-driver-v2.19/lib/mongo/collection/view/change_stream.rb', line 61

def options
  @options
end

Instance Method Details

#close ⇒ 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



237
238
239
240
241
242
243
244
245
246
 
# File 'build/ruby-driver-v2.19/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

#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



256
257
258
 
# File 'build/ruby-driver-v2.19/lib/mongo/collection/view/change_stream.rb', line 256

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

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



161
162
163
164
165
166
167
168
169
 
# File 'build/ruby-driver-v2.19/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

#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



268
269
270
271
 
# File 'build/ruby-driver-v2.19/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

#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



282
283
284
285
 
# File 'build/ruby-driver-v2.19/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

#to_enum ⇒ Object

Since:

  • 2.5.0



213
214
215
216
217
218
219
220
221
222
 
# File 'build/ruby-driver-v2.19/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

#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



183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
 
# File 'build/ruby-driver-v2.19/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