Class: Mongo::Cluster

Inherits:
Object
  • Object
show all
Extended by:
Forwardable
Includes:
Mongo::ClusterTime::Consumer, Event::Subscriber, Loggable, Monitoring::Publishable
Defined in:
build/ruby-driver-v2.19/lib/mongo/cluster.rb,
build/ruby-driver-v2.19/lib/mongo/cluster/topology.rb,
build/ruby-driver-v2.19/lib/mongo/cluster/topology.rb,
build/ruby-driver-v2.19/lib/mongo/cluster/sdam_flow.rb,
build/ruby-driver-v2.19/lib/mongo/cluster/topology/base.rb,
build/ruby-driver-v2.19/lib/mongo/cluster/topology/single.rb,
build/ruby-driver-v2.19/lib/mongo/cluster/topology/sharded.rb,
build/ruby-driver-v2.19/lib/mongo/cluster/topology/unknown.rb,
build/ruby-driver-v2.19/lib/mongo/cluster/periodic_executor.rb,
build/ruby-driver-v2.19/lib/mongo/cluster/reapers/cursor_reaper.rb,
build/ruby-driver-v2.19/lib/mongo/cluster/reapers/socket_reaper.rb,
build/ruby-driver-v2.19/lib/mongo/cluster/topology/load_balanced.rb,
build/ruby-driver-v2.19/lib/mongo/cluster/topology/no_replica_set_options.rb,
build/ruby-driver-v2.19/lib/mongo/cluster/topology/replica_set_no_primary.rb,
build/ruby-driver-v2.19/lib/mongo/cluster/topology/replica_set_with_primary.rb

Overview

Copyright © 2018-2020 MongoDB Inc.

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Defined Under Namespace

Modules: Topology Classes: CursorReaper, PeriodicExecutor, SdamFlow, SocketReaper

Constant Summary collapse

MAX_READ_RETRIES =

The default number of legacy read retries.

Since:

  • 2.1.1

1
MAX_WRITE_RETRIES =

The default number of legacy write retries.

Since:

  • 2.4.2

1
READ_RETRY_INTERVAL =

The default read retry interval, in seconds, when using legacy read retries.

Since:

  • 2.1.1

5
IDLE_WRITE_PERIOD_SECONDS =

How often an idle primary writes a no-op to the oplog.

Since:

  • 2.4.0

10
CLUSTER_TIME =
Deprecated.

The cluster time key in responses from mongos servers.

Since:

  • 2.5.0

'clusterTime'.freeze

Constants included from Loggable

Loggable::PREFIX

Instance Attribute Summary collapse

Attributes included from Mongo::ClusterTime::Consumer

#cluster_time

Attributes included from Event::Subscriber

#event_listeners

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Mongo::ClusterTime::Consumer

#advance_cluster_time

Methods included from Loggable

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

Methods included from Event::Subscriber

#subscribe_to

Methods included from Monitoring::Publishable

#publish_cmap_event, #publish_event, #publish_sdam_event

Constructor Details

#initialize(seeds, monitoring, options = Options::Redacted.new) ⇒ Cluster

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.

Note:

Cluster should never be directly instantiated outside of a Client.

Note:

When connecting to a mongodb+srv:// URI, the client expands such a URI into a list of servers and passes that list to the Cluster constructor. When connecting to a standalone mongod, the Cluster constructor receives the corresponding address as an array of one string.

Instantiate the new cluster.

Examples:

Instantiate the cluster.

Mongo::Cluster.new(["127.0.0.1:27017"], monitoring)

Parameters:

  • seeds (Array<String>) —

    The addresses of the configured servers

  • monitoring (Monitoring) —

    The monitoring.

  • options (Hash) (defaults to: Options::Redacted.new) —

    Options. Client constructor forwards its options to Cluster constructor, although Cluster recognizes only a subset of the options recognized by Client.

Options Hash (options):

  • :direct_connection (true | false) —

    Whether to connect directly to the specified seed, bypassing topology discovery. Exactly one seed must be provided.

  • :connect (Symbol) —

    Deprecated - use :direct_connection option instead of this option. The connection method to use. This forces the cluster to behave in the specified way instead of auto-discovering. One of :direct, :replica_set, :sharded

  • :replica_set (Symbol) —

    The name of the replica set to connect to. Servers not in this replica set will be ignored.

  • :scan (true | false) —

    Whether to scan all seeds in constructor. The default in driver version 2.x is to do so; driver version 3.x will not scan seeds in constructor. Opt in to the new behavior by setting this option to false. Note: setting this option to nil enables scanning seeds in constructor in driver version 2.x. Driver version 3.x will recognize this option but will ignore it and will never scan seeds in the constructor.

  • :monitoring_io (true | false) —

    For internal driver use only. Set to false to prevent SDAM-related I/O from being done by this cluster or servers under it. Note: setting this option to false will make the cluster non-functional. It is intended for use in tests which manually invoke SDAM state transitions.

  • :cleanup (true | false) —

    For internal driver use only. Set to false to prevent endSessions command being sent to the server to clean up server sessions when the cluster is disconnected, and to to not start the periodic executor. If :monitoring_io is false, :cleanup automatically defaults to false as well.

  • :heartbeat_frequency (Float) —

    The interval, in seconds, for the server monitor to refresh its description via hello.

  • :resolv_options (Hash) —

    For internal driver use only. Options to pass through to Resolv::DNS constructor for SRV lookups.

  • :server_api (Hash) —

    The requested server API version. This hash can have the following items:

    • :version – string

    • :strict – boolean

    • :deprecation_errors – boolean

Since:

  • 2.0.0



120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
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
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
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 120

def initialize(seeds, monitoring, options = Options::Redacted.new)
  if seeds.nil?
    raise ArgumentError, 'Seeds cannot be nil'
  end

  options = options.dup
  if options[:monitoring_io] == false && !options.key?(:cleanup)
    options[:cleanup] = false
  end
  @options = options.freeze

  # @update_lock covers @servers, @connecting, @connected, @topology and
  # @sessions_supported. Generally instance variables that do not have a
  # designated for them lock should only be modified under the update lock.
  # Note that topology change is locked by @update_lock and not by
  # @sdam_flow_lock.
  @update_lock = Mutex.new
  @servers = []
  @monitoring = monitoring
  @event_listeners = Event::Listeners.new
  @app_metadata = Server::AppMetadata.new(@options.merge(purpose: :application))
  @monitor_app_metadata = Server::Monitor::AppMetadata.new(@options.merge(purpose: :monitor))
  @push_monitor_app_metadata = Server::Monitor::AppMetadata.new(@options.merge(purpose: :push_monitor))
  @cluster_time_lock = Mutex.new
  @cluster_time = nil
  @srv_monitor_lock = Mutex.new
  @srv_monitor = nil
  @server_selection_semaphore = Semaphore.new
  @topology = Topology.initial(self, monitoring, options)
  # State change lock is similar to the sdam flow lock, but is designed
  # to serialize state changes initated by consumers of Cluster
  # (e.g. application connecting or disconnecting the cluster), so that
  # e.g. an application calling disconnect-connect-disconnect rapidly
  # does not put the cluster into an inconsistent state.
  # Monitoring updates performed internally by the driver do not take
  # the state change lock.
  @state_change_lock = Mutex.new
  # @sdam_flow_lock covers just the sdam flow. Note it does not apply
  # to @topology replacements which are done under @update_lock.
  @sdam_flow_lock = Mutex.new
  Session::SessionPool.create(self)

  if seeds.empty? && load_balanced?
    raise ArgumentError, 'Load-balanced clusters with no seeds are prohibited'
  end

  # The opening topology is always unknown with no servers.
  # https://github.com/mongodb/specifications/pull/388
  opening_topology = Topology::Unknown.new(options, monitoring, self)

  publish_sdam_event(
    Monitoring::TOPOLOGY_OPENING,
    Monitoring::Event::TopologyOpening.new(opening_topology)
  )

  @seeds = seeds = seeds.uniq
  servers = seeds.map do |seed|
    # Server opening events must be sent after topology change events.
    # Therefore separate server addition, done here before topology change
    # event is published, from starting to monitor the server which is
    # done later.
    add(seed, monitor: false)
  end

  if seeds.size >= 1
    # Recreate the topology to get the current server list into it
    recreate_topology(topology, opening_topology)
  end

  if load_balanced?
    # We are required by the specifications to produce certain SDAM events
    # when in load-balanced topology.
    # These events don't make a lot of sense from the standpoint of the
    # driver's SDAM implementation, nor from the standpoint of the
    # driver's load balancer implementation.
    # They are just required boilerplate.
    #
    # Note that this call must be done above the monitoring_io check
    # because that short-circuits the rest of the constructor.
    fabricate_lb_sdam_events_and_set_server_type
  end

  if options[:monitoring_io] == false
    # Omit periodic executor construction, because without servers
    # no commands can be sent to the cluster and there shouldn't ever
    # be anything that needs to be cleaned up.
    #
    # Omit monitoring individual servers and the legacy single round of
    # of SDAM on the main thread, as it would race with tests that mock
    # SDAM responses.
    @connecting = @connected = false
    return
  end

  # Update instance variables prior to starting monitoring threads.
  @connecting = false
  @connected = true

  if options[:cleanup] != false
    @cursor_reaper = CursorReaper.new(self)
    @socket_reaper = SocketReaper.new(self)
    @periodic_executor = PeriodicExecutor.new([
      @cursor_reaper, @socket_reaper,
    ], options)

    @periodic_executor.run!
  end

  unless load_balanced?
    # Need to record start time prior to starting monitoring
    start_monotime = Utils.monotonic_time

    servers.each do |server|
      server.start_monitoring
    end

    if options[:scan] != false
      server_selection_timeout = options[:server_selection_timeout] || ServerSelector::SERVER_SELECTION_TIMEOUT
      # The server selection timeout can be very short especially in
      # tests, when the client waits for a synchronous scan before
      # starting server selection. Limiting the scan to server selection time
      # then aborts the scan before it can process even local servers.
      # Therefore, allow at least 3 seconds for the scan here.
      if server_selection_timeout < 3
        server_selection_timeout = 3
      end
      deadline = start_monotime + server_selection_timeout
      # Wait for the first scan of each server to complete, for
      # backwards compatibility.
      # If any servers are discovered during this SDAM round we are going to
      # wait for these servers to also be queried, and so on, up to the
      # server selection timeout or the 3 second minimum.
      loop do
        # Ensure we do not try to read the servers list while SDAM is running
        servers = @sdam_flow_lock.synchronize do
          servers_list.dup
        end
        if servers.all? { |server| server.last_scan_monotime && server.last_scan_monotime >= start_monotime }
          break
        end
        if (time_remaining = deadline - Utils.monotonic_time) <= 0
          break
        end
        log_debug("Waiting for up to #{'%.2f' % time_remaining} seconds for servers to be scanned: #{summary}")
        # Since the semaphore may have been signaled between us checking
        # the servers list above and the wait call below, we should not
        # wait for the full remaining time - wait for up to 0.5 second, then
        # recheck the state.
        begin
          server_selection_semaphore.wait([time_remaining, 0.5].min)
        rescue ::Timeout::Error
          # nothing
        end
      end
    end

    start_stop_srv_monitor
  end
end

Instance Attribute Details

#app_metadata ⇒ Mongo::Server::AppMetadata (readonly)

Returns The application metadata, used for connection handshakes.

Returns:

Since:

  • 2.4.0



317
318
319
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 317

def app_metadata
  @app_metadata
end

#monitor_app_metadata ⇒ Object (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.

Since:

  • 2.0.0



320
321
322
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 320

def monitor_app_metadata
  @monitor_app_metadata
end

#monitoring ⇒ Monitoring (readonly)

Returns monitoring The monitoring.

Returns:

Since:

  • 2.0.0



308
309
310
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 308

def monitoring
  @monitoring
end

#options ⇒ Hash (readonly)

Returns The options hash.

Returns:

  • (Hash) —

    The options hash.

Since:

  • 2.0.0



305
306
307
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 305

def options
  @options
end

#push_monitor_app_metadata ⇒ Object (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.

Since:

  • 2.0.0



323
324
325
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 323

def push_monitor_app_metadata
  @push_monitor_app_metadata
end

#seeds ⇒ Array<String> (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 addresses of seed servers. Contains addresses that were given to Cluster when it was instantiated, not current addresses that the cluster is using as a result of SDAM.

Returns:

  • (Array<String>) —

    The addresses of seed servers. Contains addresses that were given to Cluster when it was instantiated, not current addresses that the cluster is using as a result of SDAM.

Since:

  • 2.7.0



331
332
333
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 331

def seeds
  @seeds
end

#server_selection_semaphore ⇒ Object (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.

Since:

  • 2.0.0



488
489
490
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 488

def server_selection_semaphore
  @server_selection_semaphore
end

#session_pool ⇒ Object (readonly)

Since:

  • 2.5.1



336
337
338
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 336

def session_pool
  @session_pool
end

#srv_monitor ⇒ Object (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.

Since:

  • 2.0.0



358
359
360
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 358

def srv_monitor
  @srv_monitor
end

#topology ⇒ Integer? (readonly)

The logical session timeout value in minutes.

Examples:

Get the logical session timeout in minutes.

cluster.logical_session_timeout

Returns:

  • (Integer, nil) —

    The logical session timeout.

Since:

  • 2.5.0



311
312
313
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 311

def topology
  @topology
end

Class Method Details

.create(client, monitoring: nil) ⇒ Cluster

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.

Create a cluster for the provided client, for use when we don’t want the client’s original cluster instance to be the same.

Examples:

Create a cluster for the client.

Cluster.create(client)

Parameters:

  • client (Client) —

    The client to create on.

  • monitoring. (Monitoring | nil) —

    The monitoring instance to use with the new cluster. If nil, a new instance of Monitoring will be created.

Returns:

Since:

  • 2.0.0



295
296
297
298
299
300
301
302
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 295

def self.create(client, monitoring: nil)
  cluster = Cluster.new(
    client.cluster.addresses.map(&:to_s),
    monitoring || Monitoring.new,
    client.cluster_options,
  )
  client.instance_variable_set(:@cluster, cluster)
end

Instance Method Details

#==(other) ⇒ true, false

Determine if this cluster of servers is equal to another object. Checks the servers currently in the cluster, not what was configured.

Examples:

Is the cluster equal to the object?

cluster == other

Parameters:

  • other (Object) —

    The object to compare to.

Returns:

  • (true, false) —

    If the objects are equal.

Since:

  • 2.0.0



738
739
740
741
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 738

def ==(other)
  return false unless other.is_a?(Cluster)
  addresses == other.addresses && options == other.options
end

#add(host, add_options = nil) ⇒ Server

Add a server to the cluster with the provided address. Useful in auto-discovery of new servers when an existing server executes a hello and potentially non-configured servers were included.

Examples:

Add the server for the address to the cluster.

cluster.add('127.0.0.1:27018')

Parameters:

  • host (String) —

    The address of the server to add.

  • options (Hash) —

    a customizable set of options

Returns:

  • (Server) —

    The newly added server, if not present already.

Since:

  • 2.0.0



836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 836

def add(host, add_options=nil)
  address = Address.new(host, options)
  if !addresses.include?(address)
    opts = options.merge(monitor: false)
    # If we aren't starting the montoring threads, we also don't want to
    # start the pool's populator thread.
    opts.merge!(populator_io: false) unless options.fetch(:monitoring_io, true)
    # Note that in a load-balanced topology, every server must be a
    # load balancer (load_balancer: true is specified in the options)
    # but this option isn't set here because we are required by the
    # specifications to pretent the server started out as an unknown one
    # and publish server description change event into the load balancer
    # one. The actual correct description for this server will be set
    # by the fabricate_lb_sdam_events_and_set_server_type method.
    server = Server.new(address, self, @monitoring, event_listeners, opts)
    @update_lock.synchronize do
      # Need to recheck whether server is present in @servers, because
      # the previous check was not under a lock.
      # Since we are under the update lock here, we cannot call servers_list.
      return if @servers.map(&:address).include?(address)

      @servers.push(server)
    end
    if add_options.nil? || add_options[:monitor] != false
      server.start_monitoring
    end
    server
  end
end

#addresses ⇒ Array<Mongo::Address>

The addresses in the cluster.

Examples:

Get the addresses in the cluster.

cluster.addresses

Returns:

Since:

  • 2.0.6



451
452
453
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 451

def addresses
  servers_list.map(&:address)
end

#close ⇒ nil

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.

Note:

Applications should call Client#close to disconnect from

Closes the cluster.

the cluster rather than calling this method. This method is for internal driver use only.

Disconnects all servers in the cluster, publishing appropriate SDAM events in the process. Stops SRV monitoring if it is active. Marks the cluster disconnected.

A closed cluster is no longer usable. If the client is reconnected, it will create a new cluster instance.

Returns:

  • (nil) —

    Always nil.

Since:

  • 2.0.0



506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 506

def close
  @state_change_lock.synchronize do
    unless connecting? || connected?
      return nil
    end
    if options[:cleanup] != false
      session_pool.end_sessions
      @periodic_executor.stop!
    end
    @srv_monitor_lock.synchronize do
      if @srv_monitor
        @srv_monitor.stop!
      end
    end
    @servers.each do |server|
      if server.connected?
        server.close
        publish_sdam_event(
          Monitoring::SERVER_CLOSED,
          Monitoring::Event::ServerClosed.new(server.address, topology)
        )
      end
    end
    publish_sdam_event(
      Monitoring::TOPOLOGY_CLOSED,
      Monitoring::Event::TopologyClosed.new(topology)
    )
    @update_lock.synchronize do
      @connecting = @connected = false
    end
  end
  nil
end

#connected? ⇒ true|false

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.

Whether the cluster object is connected to its cluster.

Returns:

  • (true|false) —

    Whether the cluster is connected.

Since:

  • 2.7.0



424
425
426
427
428
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 424

def connected?
  @update_lock.synchronize do
    !!@connected
  end
end

#connecting? ⇒ true|false

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.

Whether the cluster object is in the process of connecting to its cluster.

Returns:

  • (true|false) —

    Whether the cluster is connecting.

Since:

  • 2.0.0



412
413
414
415
416
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 412

def connecting?
  @update_lock.synchronize do
    !!@connecting
  end
end

#disconnect_server_if_connected(server) ⇒ Object

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.

Since:

  • 2.0.0



944
945
946
947
948
949
950
951
952
953
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 944

def disconnect_server_if_connected(server)
  if server.connected?
    server.clear_description
    server.disconnect!
    publish_sdam_event(
      Monitoring::SERVER_CLOSED,
      Monitoring::Event::ServerClosed.new(server.address, topology)
    )
  end
end

#has_readable_server?(server_selector = nil) ⇒ true, false

Determine if the cluster would select a readable server for the provided read preference.

Examples:

Is a readable server present?

topology.has_readable_server?(server_selector)

Parameters:

  • server_selector (ServerSelector) (defaults to: nil) —

    The server selector.

Returns:

  • (true, false) —

    If a readable server is present.

Since:

  • 2.4.0



755
756
757
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 755

def has_readable_server?(server_selector = nil)
  topology.has_readable_server?(self, server_selector)
end

#has_writable_server? ⇒ true, false

Determine if the cluster would select a writable server.

Examples:

Is a writable server present?

topology.has_writable_server?

Returns:

  • (true, false) —

    If a writable server is present.

Since:

  • 2.4.0



767
768
769
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 767

def has_writable_server?
  topology.has_writable_server?(self)
end

#heartbeat_interval ⇒ Float

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.

Get the refresh interval for the server. This will be defined via an option or will default to 10.

Returns:

  • (Float) —

    The heartbeat interval, in seconds.

Since:

  • 2.10.0



403
404
405
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 403

def heartbeat_interval
  options[:heartbeat_frequency] || Server::Monitor::DEFAULT_HEARTBEAT_INTERVAL
end

#inspect ⇒ String

Get the nicer formatted string for use in inspection.

Examples:

Inspect the cluster.

cluster.inspect

Returns:

  • (String) —

    The cluster inspection.

Since:

  • 2.0.0



473
474
475
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 473

def inspect
  "#<Mongo::Cluster:0x#{object_id} servers=#{servers} topology=#{topology.summary}>"
end

#load_balanced? ⇒ true | false

Returns whether the cluster is configured to be in the load-balanced topology.

Returns:

  • (true | false) —

    Whether the topology is load-balanced.

Since:

  • 2.0.0



345
346
347
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 345

def load_balanced?
  topology.is_a?(Topology::LoadBalanced)
end

#max_read_retries ⇒ Integer

Deprecated.
Note:

max_read_retries should be retrieved from the Client instance, not from a Cluster instance, because clusters may be shared between clients with different values for max read retries.

Get the maximum number of times the client can retry a read operation when using legacy read retries.

Examples:

Get the max read retries.

cluster.max_read_retries

Returns:

  • (Integer) —

    The maximum number of retries.

Since:

  • 2.1.1



374
375
376
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 374

def max_read_retries
  options[:max_read_retries] || MAX_READ_RETRIES
end

#next_primary(ping = nil, session = nil) ⇒ Mongo::Server

Get the next primary server we can send an operation to.

Examples:

Get the next primary server.

cluster.next_primary

Parameters:

  • ping (true, false) (defaults to: nil) —

    Whether to ping the server before selection. Deprecated and ignored.

  • session (Session | nil) (defaults to: nil) —

    Optional session to take into account for mongos pinning.

Returns:

Since:

  • 2.0.0



784
785
786
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 784

def next_primary(ping = nil, session = nil)
  ServerSelector.primary.select_server(self, nil, session)
end

#pool(server) ⇒ Server::ConnectionPool

Deprecated.

Get the connection pool for the server.

Examples:

Get the connection pool.

cluster.pool(server)

Parameters:

  • server (Server) —

    The server.

Returns:

Since:

  • 2.2.0



799
800
801
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 799

def pool(server)
  server.pool
end

#read_retry_interval ⇒ Float

Deprecated.
Note:

read_retry_interval should be retrieved from the Client instance, not from a Cluster instance, because clusters may be shared between clients with different values for the read retry interval.

Get the interval, in seconds, in which read retries when using legacy read retries.

Examples:

Get the read retry interval.

cluster.read_retry_interval

Returns:

  • (Float) —

    The interval.

Since:

  • 2.1.1



392
393
394
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 392

def read_retry_interval
  options[:read_retry_interval] || READ_RETRY_INTERVAL
end

#reconnect! ⇒ true

Deprecated.

Use Client#reconnect to reconnect to the cluster instead of calling this method. This method does not send SDAM events.

Reconnect all servers.

Examples:

Reconnect the cluster’s servers.

cluster.reconnect!

Returns:

  • (true) —

    Always true.

Since:

  • 2.1.0



550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 550

def reconnect!
  @state_change_lock.synchronize do
    @update_lock.synchronize do
      @connecting = true
    end
    scan!
    servers.each do |server|
      server.reconnect!
    end
    @periodic_executor.restart!
    @srv_monitor_lock.synchronize do
      if @srv_monitor
        @srv_monitor.run!
      end
    end
    @update_lock.synchronize do
      @connecting = false
      @connected = true
    end
  end
end

#remove(host, disconnect: true) ⇒ Array<Server> | true | false

Note:

The return value of this method is not part of the driver’s public API.

Remove the server from the cluster for the provided address, if it exists.

Examples:

Remove the server from the cluster.

server.remove('127.0.0.1:27017')

Parameters:

  • host (String) —

    The host/port or socket address.

  • disconnect (true | false) (defaults to: true) —

    Whether to disconnect the servers being removed. For internal driver use only.

Returns:

  • (Array<Server> | true | false) —

    If disconnect is any value other than false, including nil, returns whether any servers were removed. If disconnect is false, returns an array of servers that were removed (and should be disconnected by the caller).

Since:

  • 2.0.0



885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 885

def remove(host, disconnect: true)
  address = Address.new(host)
  removed_servers = []
  @update_lock.synchronize do
    @servers.delete_if do |server|
      (server.address == address).tap do |delete|
        if delete
          removed_servers << server
        end
      end
    end
  end
  if disconnect != false
    removed_servers.each do |server|
      disconnect_server_if_connected(server)
    end
  end
  if disconnect != false
    removed_servers.any?
  else
    removed_servers
  end
end

#run_sdam_flow(previous_desc, updated_desc, options = {}) ⇒ Object

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.

Runs SDAM flow on the cluster.

This method can be invoked to process a new server description returned by the server on a monitoring or non-monitoring connection, and also by the driver when it marks a server unknown as a result of a (network) error.

Parameters:

Options Hash (options):

  • :keep_connection_pool (true | false) —

    Usually when the new server description is unknown, the connection pool on the respective server is cleared. Set this option to true to keep the existing connection pool (required when handling not master errors on 4.2+ servers).

  • :awaited (true | false) —

    Whether the updated description was a result of processing an awaited hello.

  • :service_id (Object) —

    Change state for the specified service id only.

  • :scan_error (Mongo::Error | nil) —

    The error encountered while scanning, or nil if no error was raised.

Since:

  • 2.0.0



636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 636

def run_sdam_flow(previous_desc, updated_desc, options = {})
  if load_balanced?
    if updated_desc.config.empty?
      unless options[:keep_connection_pool]
        servers_list.each do |server|
          # TODO should service id be taken out of updated_desc?
          # We could also assert that
          # options[:service_id] == updated_desc.service_id
          err = options[:scan_error]
          interrupt = err && (err.is_a?(Error::SocketError) || err.is_a?(Error::SocketTimeoutError))
          server.clear_connection_pool(service_id: options[:service_id], interrupt_in_use_connections: interrupt)
        end
      end
    end
    return
  end

  @sdam_flow_lock.synchronize do
    flow = SdamFlow.new(self, previous_desc, updated_desc,
      awaited: options[:awaited])
    flow.server_description_changed

    # SDAM flow may alter the updated description - grab the final
    # version for the purposes of broadcasting if a server is available
    updated_desc = flow.updated_desc

    unless options[:keep_connection_pool]
      if flow.became_unknown?
        servers_list.each do |server|
          if server.address == updated_desc.address
            err = options[:scan_error]
            interrupt = err && (err.is_a?(Error::SocketError) || err.is_a?(Error::SocketTimeoutError))
            server.clear_connection_pool(interrupt_in_use_connections: interrupt)
          end
        end
      end
    end

    start_stop_srv_monitor
  end

  # Some updated descriptions, e.g. a mismatched me one, result in the
  # server whose description we are processing being removed from
  # the topology. When this happens, the server's monitoring thread gets
  # killed. As a result, any code after the flow invocation may not run
  # a particular monitor instance, hence there should generally not be
  # any code in this method past the flow invocation.
  #
  # However, this broadcast call can be here because if the monitoring
  # thread got killed the server should have been closed and no client
  # should be currently waiting for it, thus not signaling the semaphore
  # shouldn't cause any problems.
  unless updated_desc.unknown?
    server_selection_semaphore.broadcast
  end
end

#scan!(sync = true) ⇒ true

Note:

In both synchronous and asynchronous scans, each monitor thread maintains a minimum interval between scans, meaning calling this method may not initiate a scan on a particular server the very next instant.

Force a scan of all known servers in the cluster.

If the sync parameter is true which is the default, the scan is performed synchronously in the thread which called this method. Each server in the cluster is checked sequentially. If there are many servers in the cluster or they are slow to respond, this can be a long running operation.

If the sync parameter is false, this method instructs all server monitor threads to perform an immediate scan and returns without waiting for scan results.

Examples:

Force a full cluster scan.

cluster.scan!

Returns:

  • (true) —

    Always true.

Since:

  • 2.0.0



595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 595

def scan!(sync=true)
  if sync
    servers_list.each do |server|
      if server.monitor
        server.monitor.scan!
      else
        log_warn("Synchronous scan requested on cluster #{summary} but server #{server} has no monitor")
      end
    end
  else
    servers_list.each do |server|
      server.scan_semaphore.signal
    end
  end
  true
end

#servers ⇒ Array<Server>

Get a list of server candidates from the cluster that can have operations executed on them.

Examples:

Get the server candidates for an operation.

cluster.servers

Returns:

  • (Array<Server>) —

    The candidate servers.

Since:

  • 2.0.0



439
440
441
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 439

def servers
  topology.servers(servers_list)
end

#servers_list ⇒ Object

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.

Since:

  • 2.0.0



937
938
939
940
941
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 937

def servers_list
  @update_lock.synchronize do
    @servers.dup
  end
end

#set_server_list(server_address_strs) ⇒ Object

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.

Sets the list of servers to the addresses in the provided list of address strings.

This method is called by the SRV monitor after receiving new DNS records for the monitored hostname.

Removes servers in the cluster whose addresses are not in the passed list of server addresses, and adds servers for any addresses in the argument which are not already in the cluster.

Parameters:

  • server_address_strs (Array<String>) —

    List of server addresses to sync the cluster servers to.

Since:

  • 2.0.0



707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 707

def set_server_list(server_address_strs)
  @sdam_flow_lock.synchronize do
    # If one of the new addresses is not in the current servers list,
    # add it to the servers list.
    server_address_strs.each do |address_str|
      unless servers_list.any? { |server| server.address.seed == address_str }
        add(address_str)
      end
    end

    # If one of the servers' addresses are not in the new address list,
    # remove that server from the servers list.
    servers_list.each do |server|
      unless server_address_strs.any? { |address_str| server.address.seed == address_str }
        remove(server.address.seed)
      end
    end
  end
end

#summary ⇒ Object

Note:

This method is experimental and subject to change.

Since:

  • 2.7.0



481
482
483
484
485
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 481

def summary
  "#<Cluster " +
  "topology=#{topology.summary} "+
  "servers=[#{servers_list.map(&:summary).join(',')}]>"
end

#update_cluster_time(result) ⇒ Object

Update the max cluster time seen in a response.

Examples:

Update the cluster time.

cluster.update_cluster_time(result)

Parameters:

Returns:

  • (Object) —

    The cluster time.

Since:

  • 2.5.0



813
814
815
816
817
818
819
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 813

def update_cluster_time(result)
  if cluster_time_doc = result.cluster_time
    @cluster_time_lock.synchronize do
      advance_cluster_time(cluster_time_doc)
    end
  end
end

#update_topology(new_topology) ⇒ Object

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.

Since:

  • 2.0.0



910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 910

def update_topology(new_topology)
  old_topology = nil
  @update_lock.synchronize do
    old_topology = topology
    @topology = new_topology
  end

  # If new topology has data bearing servers, we know for sure whether
  # sessions are supported - update our cached value.
  # If new topology has no data bearing servers, leave the old value
  # as it is and sessions_supported? method will perform server selection
  # to try to determine session support accurately, falling back to the
  # last known value.
  if topology.data_bearing_servers?
    sessions_supported = !!topology.logical_session_timeout
    @update_lock.synchronize do
      @sessions_supported = sessions_supported
    end
  end

  publish_sdam_event(
    Monitoring::TOPOLOGY_CHANGED,
    Monitoring::Event::TopologyChanged.new(old_topology, topology)
  )
end

#validate_session_support! ⇒ Object

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.

Raises Error::SessionsNotAvailable if the deployment that the driver is connected to does not support sessions.

Session support may change over time, for example due to servers in the deployment being upgraded or downgraded. If the client isn’t connected to any servers and doesn’t find any servers for the duration of server selection timeout, this method will raise NoServerAvailable. This method is called from the operation execution flow, and if it raises NoServerAvailable the entire operation will fail with that exception, since the operation execution has waited for the server selection timeout for any server to become available (which would be a superset of the servers suitable for the operation being attempted) and none materialized.

Raises:

  • (Error::SessionsNotAvailable) —

    If the deployment that the driver is connected to does not support sessions.

  • (Error::NoServerAvailable) —

    If the client isn’t connected to any servers and doesn’t find any servers for the duration of server selection timeout.

Since:

  • 2.0.0



976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
 
# File 'build/ruby-driver-v2.19/lib/mongo/cluster.rb', line 976

def validate_session_support!
  if topology.is_a?(Topology::LoadBalanced)
    return
  end

  @state_change_lock.synchronize do
    @sdam_flow_lock.synchronize do
      if topology.data_bearing_servers?
        unless topology.logical_session_timeout
          raise_sessions_not_supported
        end
      end
    end
  end

  # No data bearing servers known - perform server selection to try to
  # get a response from at least one of them, to return an accurate
  # assessment of whether sessions are currently supported.
  ServerSelector.get(mode: :primary_preferred).select_server(self)
  @state_change_lock.synchronize do
    @sdam_flow_lock.synchronize do
      unless topology.logical_session_timeout
        raise_sessions_not_supported
      end
    end
  end
end