Class: Mongo::Cluster
- Inherits:
-
Object
- Object
- Mongo::Cluster
- Extended by:
- Forwardable
- Defined in:
- build/ruby-driver-v2.17/lib/mongo/cluster.rb,
build/ruby-driver-v2.17/lib/mongo/cluster/topology.rb,
build/ruby-driver-v2.17/lib/mongo/cluster/topology.rb,
build/ruby-driver-v2.17/lib/mongo/cluster/sdam_flow.rb,
build/ruby-driver-v2.17/lib/mongo/cluster/topology/base.rb,
build/ruby-driver-v2.17/lib/mongo/cluster/topology/single.rb,
build/ruby-driver-v2.17/lib/mongo/cluster/topology/sharded.rb,
build/ruby-driver-v2.17/lib/mongo/cluster/topology/unknown.rb,
build/ruby-driver-v2.17/lib/mongo/cluster/periodic_executor.rb,
build/ruby-driver-v2.17/lib/mongo/cluster/reapers/cursor_reaper.rb,
build/ruby-driver-v2.17/lib/mongo/cluster/reapers/socket_reaper.rb,
build/ruby-driver-v2.17/lib/mongo/cluster/topology/load_balanced.rb,
build/ruby-driver-v2.17/lib/mongo/cluster/topology/no_replica_set_options.rb,
build/ruby-driver-v2.17/lib/mongo/cluster/topology/replica_set_no_primary.rb,
build/ruby-driver-v2.17/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.
1
- MAX_WRITE_RETRIES =
The default number of legacy write retries.
1
- READ_RETRY_INTERVAL =
The default read retry interval, in seconds, when using legacy read retries.
5
- IDLE_WRITE_PERIOD_SECONDS =
How often an idle primary writes a no-op to the oplog.
10
- CLUSTER_TIME =
Deprecated.
The cluster time key in responses from mongos servers.
'clusterTime'.freeze
Constants included from Loggable
Instance Attribute Summary collapse
-
#app_metadata ⇒ Mongo::Server::AppMetadata
readonly
The application metadata, used for connection handshakes.
- #monitor_app_metadata ⇒ Object readonly private
-
#monitoring ⇒ Monitoring
readonly
Monitoring The monitoring.
-
#options ⇒ Hash
readonly
The options hash.
- #push_monitor_app_metadata ⇒ Object readonly private
-
#seeds ⇒ Array<String>
readonly
private
The addresses of seed servers.
- #server_selection_semaphore ⇒ Object readonly private
- #session_pool ⇒ Object readonly
- #srv_monitor ⇒ Object readonly private
-
#topology ⇒ Integer?
readonly
The logical session timeout value in minutes.
Attributes included from Mongo::ClusterTime::Consumer
Attributes included from Event::Subscriber
Class Method Summary collapse
-
.create(client) ⇒ Cluster
private
Create a cluster for the provided client, for use when we don’t want the client’s original cluster instance to be the same.
Instance Method Summary collapse
-
#==(other) ⇒ true, false
Determine if this cluster of servers is equal to another object.
-
#add(host, add_options = nil) ⇒ Server
Add a server to the cluster with the provided address.
-
#addresses ⇒ Array<Mongo::Address>
The addresses in the cluster.
-
#connected? ⇒ true|false
private
Whether the cluster object is connected to its cluster.
-
#connecting? ⇒ true|false
private
Whether the cluster object is in the process of connecting to its cluster.
-
#disconnect! ⇒ true
Closes the cluster.
- #disconnect_server_if_connected(server) ⇒ Object private
-
#has_readable_server?(server_selector = nil) ⇒ true, false
Determine if the cluster would select a readable server for the provided read preference.
-
#has_writable_server? ⇒ true, false
Determine if the cluster would select a writable server.
-
#heartbeat_interval ⇒ Float
private
Get the refresh interval for the server.
-
#initialize(seeds, monitoring, options = Options::Redacted.new) ⇒ Cluster
constructor
private
Instantiate the new cluster.
-
#inspect ⇒ String
Get the nicer formatted string for use in inspection.
-
#load_balanced? ⇒ true | false
Returns whether the cluster is configured to be in the load-balanced topology.
- #max_read_retries ⇒ Integer deprecated Deprecated.
-
#next_primary(ping = nil, session = nil) ⇒ Mongo::Server
Get the next primary server we can send an operation to.
- #pool(server) ⇒ Server::ConnectionPool deprecated Deprecated.
- #read_retry_interval ⇒ Float deprecated Deprecated.
-
#reconnect! ⇒ true
deprecated
Deprecated.
Use Client#reconnect to reconnect to the cluster instead of calling this method. This method does not send SDAM events.
-
#remove(host, disconnect: true) ⇒ Array<Server> | true | false
Remove the server from the cluster for the provided address, if it exists.
-
#run_sdam_flow(previous_desc, updated_desc, options = {}) ⇒ Object
private
Runs SDAM flow on the cluster.
-
#scan!(sync = true) ⇒ true
Force a scan of all known servers in the cluster.
-
#servers ⇒ Array<Server>
Get a list of server candidates from the cluster that can have operations executed on them.
- #servers_list ⇒ Object private
-
#set_server_list(server_address_strs) ⇒ Object
private
Sets the list of servers to the addresses in the provided list of address strings.
- #summary ⇒ Object
-
#update_cluster_time(result) ⇒ Object
Update the max cluster time seen in a response.
- #update_topology(new_topology) ⇒ Object private
-
#validate_session_support! ⇒ Object
private
Raises Error::SessionsNotAvailable if the deployment that the driver is connected to does not support sessions.
Methods included from Mongo::ClusterTime::Consumer
Methods included from Loggable
#log_debug, #log_error, #log_fatal, #log_info, #log_warn, #logger
Methods included from Event::Subscriber
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.
Cluster should never be directly instantiated outside of a Client.
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.
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.17/lib/mongo/cluster.rb', line 120 def initialize(seeds, monitoring, = Options::Redacted.new) if seeds.nil? raise ArgumentError, 'Seeds cannot be nil' end = .dup if [:monitoring_io] == false && !.key?(:cleanup) [:cleanup] = false end @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, ) # 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(, 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 [: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 [:cleanup] != false @cursor_reaper = CursorReaper.new(self) @socket_reaper = SocketReaper.new(self) @periodic_executor = PeriodicExecutor.new([ @cursor_reaper, @socket_reaper, ], ) @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 [:scan] != false server_selection_timeout = [: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 1 second, then # recheck the state. begin server_selection_semaphore.wait([time_remaining, 1].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.
315 316 317 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 315 def @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.
318 319 320 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 318 def @monitor_app_metadata end |
#monitoring ⇒ Monitoring (readonly)
Returns monitoring The monitoring.
306 307 308 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 306 def monitoring @monitoring end |
#options ⇒ Hash (readonly)
Returns The options hash.
303 304 305 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 303 def @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.
321 322 323 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 321 def @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.
329 330 331 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 329 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.
486 487 488 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 486 def server_selection_semaphore @server_selection_semaphore end |
#session_pool ⇒ Object (readonly)
334 335 336 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 334 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.
356 357 358 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 356 def srv_monitor @srv_monitor end |
#topology ⇒ Integer? (readonly)
The logical session timeout value in minutes.
309 310 311 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 309 def topology @topology end |
Class Method Details
.create(client) ⇒ 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.
293 294 295 296 297 298 299 300 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 293 def self.create(client) cluster = Cluster.new( client.cluster.addresses.map(&:to_s), Monitoring.new, client., ) 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.
723 724 725 726 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 723 def ==(other) return false unless other.is_a?(Cluster) addresses == other.addresses && == other. 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.
821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 821 def add(host, =nil) address = Address.new(host, ) if !addresses.include?(address) opts = .merge(monitor: false) # 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 .nil? || [:monitor] != false server.start_monitoring end server end end |
#addresses ⇒ Array<Mongo::Address>
The addresses in the cluster.
449 450 451 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 449 def addresses servers_list.map(&:address) 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.
422 423 424 425 426 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 422 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.
410 411 412 413 414 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 410 def connecting? @update_lock.synchronize do !!@connecting end end |
#disconnect! ⇒ true
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.
501 502 503 504 505 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 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 501 def disconnect! @state_change_lock.synchronize do unless connecting? || connected? return true end if [: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.disconnect! 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 true 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.
926 927 928 929 930 931 932 933 934 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 926 def disconnect_server_if_connected(server) if server.connected? 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.
740 741 742 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 740 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.
752 753 754 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 752 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.
401 402 403 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 401 def heartbeat_interval [:heartbeat_frequency] || Server::Monitor::DEFAULT_HEARTBEAT_INTERVAL end |
#inspect ⇒ String
Get the nicer formatted string for use in inspection.
471 472 473 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 471 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.
343 344 345 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 343 def load_balanced? topology.is_a?(Topology::LoadBalanced) end |
#max_read_retries ⇒ Integer
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.
372 373 374 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 372 def max_read_retries [: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.
769 770 771 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 769 def next_primary(ping = nil, session = nil) ServerSelector.primary.select_server(self, nil, session) end |
#pool(server) ⇒ Server::ConnectionPool
Get the connection pool for the server.
784 785 786 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 784 def pool(server) server.pool end |
#read_retry_interval ⇒ Float
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.
390 391 392 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 390 def read_retry_interval [:read_retry_interval] || READ_RETRY_INTERVAL end |
#reconnect! ⇒ true
Use Client#reconnect to reconnect to the cluster instead of calling this method. This method does not send SDAM events.
Reconnect all servers.
545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 545 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
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.
867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 867 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.
629 630 631 632 633 634 635 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 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 629 def run_sdam_flow(previous_desc, updated_desc, = {}) if load_balanced? if updated_desc.config.empty? unless [: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 server.clear_connection_pool(service_id: [:service_id]) end end end return end @sdam_flow_lock.synchronize do flow = SdamFlow.new(self, previous_desc, updated_desc, awaited: [: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 [:keep_connection_pool] if flow.became_unknown? servers_list.each do |server| if server.address == updated_desc.address server.clear_connection_pool 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
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.
590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 590 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.
437 438 439 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 437 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.
919 920 921 922 923 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 919 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.
696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 696 def set_server_list(server_address_strs) @sdam_flow_lock.synchronize do server_address_strs.each do |address_str| unless servers_list.any? { |server| server.address.seed == address_str } add(address_str) end end 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
This method is experimental and subject to change.
479 480 481 482 483 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 479 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.
798 799 800 801 802 803 804 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 798 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.
892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 892 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 is currently not connected to any data bearing servers, this method considers the state of session support as of when the client was last connected to at least one server. If the client has never connected to any servers, the deployment is considered to not support sessions.
947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 |
# File 'build/ruby-driver-v2.17/lib/mongo/cluster.rb', line 947 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. begin 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 rescue Error::NoServerAvailable # We haven't been able to contact any servers - use last known # value for session support. @state_change_lock.synchronize do @sdam_flow_lock.synchronize do @update_lock.synchronize do unless @sessions_supported raise_sessions_not_supported end end end end end end |