/// Creates a SettingProxy that is listening to requests from the
/// provided receiver and will send responses/updates on the given sender.
Creates a SettingProxy that is listening to requests from the
provided receiver and will send responses/updates on the given sender.
Creates a SettingProxy that is listening to requests from the provided receiver and will send responses / updates on the given sender .
pub async fn create(
        setting_type: SettingType,
        handler_factory: Arc<Mutex<dyn SettingHandlerFactory + Send + Sync>>,
        messenger_factory: service::message::Factory,
        max_attempts: u64,
        teardown_timeout: Duration,
        request_timeout: Option<Duration>,
        retry_on_timeout: bool,
    ) -> Result<service::message::Signature, Error> {
        let (messenger, receptor) = messenger_factory
            .create(MessengerType::Addressable(service::Address::Handler(setting_type)))
            .await
            .map_err(Error::new)?;
        let service_signature = receptor.get_signature();
        // TODO(fxbug.dev/67536): Remove receptors below as their logic is
        // migrated to the MessageHub defined above.
        let event_publisher = event::Publisher::create(
            &messenger_factory,
            MessengerType::Addressable(service::Address::EventSource(
                event::Address::SettingProxy(setting_type),
            )),
        )
        .await;
        let (proxy_request_sender, proxy_request_receiver) =
            futures::channel::mpsc::unbounded::<ProxyRequest>();
        // We must create handle here rather than return back the value as we
        // reference the proxy in the async tasks below.
        let mut proxy = Self {
            setting_type,
            handler_factory,
            next_request_id: 0,
            client_signature: None,
            active_request: None,
            pending_requests: VecDeque::new(),
            listen_requests: Vec::new(),
            messenger_factory,
            messenger,
            signature: service_signature,
            event_publisher: event_publisher,
            proxy_request_sender,
            max_attempts,
            teardown_timeout,
            request_timeout: request_timeout,
            retry_on_timeout,
            teardown_cancellation: None,
        };
        // Main task loop for receiving and processing incoming messages.
        fasync::Task::spawn(async move {
            let receptor_fuse = receptor.fuse();
            let proxy_fuse = proxy_request_receiver.fuse();
            futures::pin_mut!(receptor_fuse, proxy_fuse);
            loop {
                futures::select! {
                    // Handles requests from the service MessageHub and
                    // communication from the setting controller.
                    event = receptor_fuse.select_next_some() => {
                        proxy.process_service_event(event).await;
                    }
                    // Handles messages for enqueueing requests and processing
                    // results on the main event loop for proxy.
                    request = proxy_fuse.select_next_some() => {
                        proxy.process_proxy_request(request).await;
                    }
                }
            }
        })
        .detach();
        Ok(service_signature)
    }
pub async fn create ( setting_type : SettingType , handler_factory : Arc < Mutex < dyn SettingHandlerFactory + Send + Sync > > , messenger_factory : service :: message :: Factory , max_attempts : u64 , teardown_timeout : Duration , request_timeout : Option < Duration > , retry_on_timeout : bool , ) -> Result < service :: message :: Signature , Error > { let ( messenger , receptor ) = messenger_factory . create ( MessengerType :: Addressable ( service :: Address :: Handler ( setting_type ) ) ) . await . map_err ( Error :: new ) ? ; let service_signature = receptor . get_signature ( ) ; // TODO(fxbug.dev/67536): Remove receptors below as their logic is // migrated to the MessageHub defined above. let event_publisher = event :: Publisher :: create ( & messenger_factory , MessengerType :: Addressable ( service :: Address :: EventSource ( event :: Address :: SettingProxy ( setting_type ) , ) ) , ) . await ; let ( proxy_request_sender , proxy_request_receiver ) = futures :: channel :: mpsc :: unbounded :: < ProxyRequest > ( ) ; // We must create handle here rather than return back the value as we // reference the proxy in the async tasks below. let mut proxy = Self { setting_type , handler_factory , next_request_id : 0 , client_signature : None , active_request : None , pending_requests : VecDeque :: new ( ) , listen_requests : Vec :: new ( ) , messenger_factory , messenger , signature : service_signature , event_publisher : event_publisher , proxy_request_sender , max_attempts , teardown_timeout , request_timeout : request_timeout , retry_on_timeout , teardown_cancellation : None , } ; // Main task loop for receiving and processing incoming messages. fasync :: Task :: spawn ( async move { let receptor_fuse = receptor . fuse ( ) ; let proxy_fuse = proxy_request_receiver . fuse ( ) ; futures :: pin_mut ! ( receptor_fuse proxy_fuse ) ; loop { futures :: select ! { // Handles requests from the service MessageHub and // communication from the setting controller. event receptor_fuse select_next_some ( ) { proxy process_service_event ( event ) await } // Handles messages for enqueueing requests and processing // results on the main event loop for proxy. request proxy_fuse select_next_some ( ) { proxy process_proxy_request ( request ) await } } } } ) . detach ( ) ; Ok ( service_signature ) }
Creates a SettingProxy that is listening to requests from the
provided receiver and will send responses/updates on the given sender.
Creates a SettingProxy that is listening to requests from the provided receiver and will send responses / updates on the given sender .
setting_type
SettingType
handler_factory
Arc<Mutex<dyn SettingHandlerFactory + Send + Sync>>
messenger_factory
service::message::Factory
max_attempts
u64
teardown_timeout
Duration
request_timeout
Option<Duration>
retry_on_timeout
bool