pub(crate) fn new(options: ClientOptions) -> Result<Self> {
        let description = TopologyDescription::new(options.clone())?;
        let is_load_balanced = description.topology_type() == TopologyType::LoadBalanced;

        let id = ObjectId::new();
        if let Some(ref handler) = options.sdam_event_handler {
            let event = TopologyOpeningEvent { topology_id: id };
            handler.handle_topology_opening_event(event);
        }

        let common = Common {
            is_alive: Arc::new(AtomicBool::new(true)),
            message_manager: TopologyMessageManager::new(),
            options: options.clone(),
            id,
        };

        let http_client = HttpClient::default();

        let topology_state = TopologyState {
            description,
            servers: Default::default(),
            http_client,
            options: options.clone(),
            id,
        };

        let state = Arc::new(RwLock::new(topology_state));
        let topology = Topology { state, common };

        // we can block in place here because we're the only ones with access to the lock, so it
        // should be acquired immediately.
        let mut topology_state = RUNTIME.block_in_place(topology.state.write());
        for address in &options.hosts {
            topology_state.add_new_server(address.clone(), options.clone(), &topology.downgrade());
        }

        if let Some(ref handler) = options.sdam_event_handler {
            let event = TopologyDescriptionChangedEvent {
                topology_id: id,
                previous_description: TopologyDescription::new_empty().into(),
                new_description: topology_state.description.clone().into(),
            };
            handler.handle_topology_description_changed_event(event);

            for server_address in &options.hosts {
                let event = ServerOpeningEvent {
                    topology_id: id,
                    address: server_address.clone(),
                };
                handler.handle_server_opening_event(event);
            }
        }

        if is_load_balanced {
            for server_address in &options.hosts {
                // Load-balanced clients don't have a heartbeat monitor, so we synthesize
                // updating the server to `ServerType::LoadBalancer` with an RTT of 0 so it'll
                // be selected.
                let new_desc = ServerDescription {
                    server_type: ServerType::LoadBalancer,
                    average_round_trip_time: Some(Duration::from_nanos(0)),
                    ..ServerDescription::new(server_address.clone(), None)
                };
                topology_state
                    .update(new_desc, &options, topology.downgrade())
                    .map_err(Error::internal)?;
            }
        }
        #[cfg(test)]
        let disable_monitoring_threads = options
            .test_options
            .map(|to| to.disable_monitoring_threads)
            .unwrap_or(false);
        #[cfg(not(test))]
        let disable_monitoring_threads = false;
        if !is_load_balanced && !disable_monitoring_threads {
            SrvPollingMonitor::start(topology.downgrade());
        }

        drop(topology_state);
        Ok(topology)
    }
/// Creates a new Topology given the `options`.
Creates a new Topology given the `options`.
Creates a new Topology given the ` options ` .
pub(crate) fn new(options: ClientOptions) -> Result<Self> {
        let description = TopologyDescription::new(options.clone())?;
        let is_load_balanced = description.topology_type() == TopologyType::LoadBalanced;
        let id = ObjectId::new();
        if let Some(ref handler) = options.sdam_event_handler {
            let event = TopologyOpeningEvent { topology_id: id };
            handler.handle_topology_opening_event(event);
        }
        let common = Common {
            is_alive: Arc::new(AtomicBool::new(true)),
            message_manager: TopologyMessageManager::new(),
            options: options.clone(),
            id,
        };
        let http_client = HttpClient::default();
        let topology_state = TopologyState {
            description,
            servers: Default::default(),
            http_client,
            options: options.clone(),
            id,
        };
        let state = Arc::new(RwLock::new(topology_state));
        let topology = Topology { state, common };
        // we can block in place here because we're the only ones with access to the lock, so it
        // should be acquired immediately.
        let mut topology_state = RUNTIME.block_in_place(topology.state.write());
        for address in &options.hosts {
            topology_state.add_new_server(address.clone(), options.clone(), &topology.downgrade());
        }
        if let Some(ref handler) = options.sdam_event_handler {
            let event = TopologyDescriptionChangedEvent {
                topology_id: id,
                previous_description: TopologyDescription::new_empty().into(),
                new_description: topology_state.description.clone().into(),
            };
            handler.handle_topology_description_changed_event(event);
            for server_address in &options.hosts {
                let event = ServerOpeningEvent {
                    topology_id: id,
                    address: server_address.clone(),
                };
                handler.handle_server_opening_event(event);
            }
        }
        if is_load_balanced {
            for server_address in &options.hosts {
                // Load-balanced clients don't have a heartbeat monitor, so we synthesize
                // updating the server to `ServerType::LoadBalancer` with an RTT of 0 so it'll
                // be selected.
                let new_desc = ServerDescription {
                    server_type: ServerType::LoadBalancer,
                    average_round_trip_time: Some(Duration::from_nanos(0)),
                    ..ServerDescription::new(server_address.clone(), None)
                };
                topology_state
                    .update(new_desc, &options, topology.downgrade())
                    .map_err(Error::internal)?;
            }
        }
        #[cfg(test)]
        let disable_monitoring_threads = options
            .test_options
            .map(|to| to.disable_monitoring_threads)
            .unwrap_or(false);
        #[cfg(not(test))]
        let disable_monitoring_threads = false;
        if !is_load_balanced && !disable_monitoring_threads {
            SrvPollingMonitor::start(topology.downgrade());
        }
        drop(topology_state);
        Ok(topology)
    }
pub ( crate ) fn new ( options : ClientOptions ) -> Result < Self > { let description = TopologyDescription :: new ( options . clone ( ) ) ? ; let is_load_balanced = description . topology_type ( ) == TopologyType :: LoadBalanced ; let id = ObjectId :: new ( ) ; if let Some ( ref handler ) = options . sdam_event_handler { let event = TopologyOpeningEvent { topology_id : id } ; handler . handle_topology_opening_event ( event ) ; } let common = Common { is_alive : Arc :: new ( AtomicBool :: new ( true ) ) , message_manager : TopologyMessageManager :: new ( ) , options : options . clone ( ) , id , } ; let http_client = HttpClient :: default ( ) ; let topology_state = TopologyState { description , servers : Default :: default ( ) , http_client , options : options . clone ( ) , id , } ; let state = Arc :: new ( RwLock :: new ( topology_state ) ) ; let topology = Topology { state , common } ; // we can block in place here because we're the only ones with access to the lock, so it // should be acquired immediately. let mut topology_state = RUNTIME . block_in_place ( topology . state . write ( ) ) ; for address in & options . hosts { topology_state . add_new_server ( address . clone ( ) , options . clone ( ) , & topology . downgrade ( ) ) ; } if let Some ( ref handler ) = options . sdam_event_handler { let event = TopologyDescriptionChangedEvent { topology_id : id , previous_description : TopologyDescription :: new_empty ( ) . into ( ) , new_description : topology_state . description . clone ( ) . into ( ) , } ; handler . handle_topology_description_changed_event ( event ) ; for server_address in & options . hosts { let event = ServerOpeningEvent { topology_id : id , address : server_address . clone ( ) , } ; handler . handle_server_opening_event ( event ) ; } } if is_load_balanced { for server_address in & options . hosts { // Load-balanced clients don't have a heartbeat monitor, so we synthesize // updating the server to `ServerType::LoadBalancer` with an RTT of 0 so it'll // be selected. let new_desc = ServerDescription { server_type : ServerType :: LoadBalancer , average_round_trip_time : Some ( Duration :: from_nanos ( 0 ) ) , .. ServerDescription :: new ( server_address . clone ( ) , None ) } ; topology_state . update ( new_desc , & options , topology . downgrade ( ) ) . map_err ( Error :: internal ) ? ; } } # [ cfg ( test ) ] let disable_monitoring_threads = options . test_options . map ( | to | to . disable_monitoring_threads ) . unwrap_or ( false ) ; # [ cfg ( not ( test ) ) ] let disable_monitoring_threads = false ; if ! is_load_balanced && ! disable_monitoring_threads { SrvPollingMonitor :: start ( topology . downgrade ( ) ) ; } drop ( topology_state ) ; Ok ( topology ) }
Creates a new Topology given the `options`.
Creates a new Topology given the ` options ` .
options
ClientOptions
options
ClientOptions