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

Inherits:
Aggregation
  • Object
show all
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

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

#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

#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’ (behaves same as nil), ‘updateLookup’. 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. The default is to not send a value.

  • :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.

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

Since:

  • 2.5.0



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, 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-master/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



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?

Examples:

Determine whether the change stream is closed.

stream.closed?

Returns:

  • (true, false) —

    If the change stream is closed.

Since:

  • 2.5.0



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

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



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.

Examples:

Inspect the change stream object.

stream.inspect

Returns:

  • (String) —

    The change stream inspection.

Since:

  • 2.5.0



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.

Examples:

Get the change stream resume token.

stream.resume_token

Returns:

  • (BSON::Document | nil) —

    The change stream resume token.

Since:

  • 2.10.0



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

Since:

  • 2.5.0



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.

Returns:

  • (BSON::Document | nil) —

    A change stream document.

Raises:

  • (StopIteration)

Since:

  • 2.6.0



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