pub(crate) fn send_request<C, T, F>(&mut self, component: &mut C, f: F) -> T
    where
        C: Component<REv>,
        <C as Component<REv>>::Event: Send + 'static,
        T: Send + 'static,
        F: FnOnce(Responder<T>) -> C::Event,
        REv: ReactorEvent,
    {
        // Prepare a channel.
        let (sender, receiver) = oneshot::channel();

        // Create response function.
        let responder = Responder::without_shutdown(sender);

        // Create the event for the component.
        let request_event = f(responder);

        // Send directly to component.
        let returned_effects = self.send_event(component, request_event);

        // Execute the effects on our dedicated runtime, hopefully creating the responses.
        let mut join_handles = Vec::new();
        for effect in returned_effects {
            join_handles.push(self.runtime.spawn(effect));
        }

        // Wait for a response to arrive.
        self.runtime.block_on(receiver).unwrap_or_else(|err| {
            // A channel was closed and this is usually an error. However, we consider all pending
            // events, in case we did get a control announcement requiring us to fatal error instead
            // before panicking on the basis of the missing response.

            // We give each of them a little time to produce the desired event. Note that `join_all`
            // should be safe to cancel, since we are only awaiting join handles.
            let join_all = async {
                for handle in join_handles {
                    if let Err(err) = handle.await {
                        warn!("Join error while waiting for an effect to finish: {}", err);
                    };
                }
            };

            if let Err(_timeout) = self.runtime.block_on(async move {
                // Note: timeout can only be called from within a running running, this is why
                // we use an extra `async` block here.
                tokio::time::timeout(FATAL_GRACE_TIME, join_all).await
            }) {
                warn!(grace_time=?FATAL_GRACE_TIME, "while a responder was dropped in a unit test, \
                I waited for all other pending effects to complete in case the output of a \
                `fatal!` was among them but none of them completed");
            }

            // Iterate over all events that currently are inside the queue and fish out any fatal.
            for _ in 0..(self.scheduler.item_count()) {
                let ((_ancestor, ev), _queue_kind) = self.runtime.block_on(self.scheduler.pop());

                if let Some(ctrl_ann) = ev.as_control() {
                    match ctrl_ann {
                        fatal @ ControlAnnouncement::FatalError { .. } => {
                            panic!(
                                "a control announcement requesting a fatal error was received: {}",
                                fatal
                            )
                        }
                    }
                } else {
                    debug!(?ev, "ignoring event while looking for a fatal")
                }
            }

            // Barring a `fatal`, the channel should never be closed, ever.
            panic!(
                "request for {} channel closed with return value \"{}\" in unit test harness",
                type_name::<T>(),
                err,
            );
        })
    }
/// Sends a request, expecting an immediate response.
///
/// Sends a request by creating a channel for the response, then mapping it using the function
/// `f`. Executes all returned effects, then awaits a response.
Sends a request, expecting an immediate response.
Sends a request by creating a channel for the response, then mapping it using the function
`f`. Executes all returned effects, then awaits a response.
Sends a request expecting an immediate response . Sends a request by creating a channel for the response then mapping it using the function ` f ` . Executes all returned effects then awaits a response .
pub(crate) fn send_request<C, T, F>(&mut self, component: &mut C, f: F) -> T
    where
        C: Component<REv>,
        <C as Component<REv>>::Event: Send + 'static,
        T: Send + 'static,
        F: FnOnce(Responder<T>) -> C::Event,
        REv: ReactorEvent,
    {
        // Prepare a channel.
        let (sender, receiver) = oneshot::channel();
        // Create response function.
        let responder = Responder::without_shutdown(sender);
        // Create the event for the component.
        let request_event = f(responder);
        // Send directly to component.
        let returned_effects = self.send_event(component, request_event);
        // Execute the effects on our dedicated runtime, hopefully creating the responses.
        let mut join_handles = Vec::new();
        for effect in returned_effects {
            join_handles.push(self.runtime.spawn(effect));
        }
        // Wait for a response to arrive.
        self.runtime.block_on(receiver).unwrap_or_else(|err| {
            // A channel was closed and this is usually an error. However, we consider all pending
            // events, in case we did get a control announcement requiring us to fatal error instead
            // before panicking on the basis of the missing response.
            // We give each of them a little time to produce the desired event. Note that `join_all`
            // should be safe to cancel, since we are only awaiting join handles.
            let join_all = async {
                for handle in join_handles {
                    if let Err(err) = handle.await {
                        warn!("Join error while waiting for an effect to finish: {}", err);
                    };
                }
            };
            if let Err(_timeout) = self.runtime.block_on(async move {
                // Note: timeout can only be called from within a running running, this is why
                // we use an extra `async` block here.
                tokio::time::timeout(FATAL_GRACE_TIME, join_all).await
            }) {
                warn!(grace_time=?FATAL_GRACE_TIME, "while a responder was dropped in a unit test, \
                I waited for all other pending effects to complete in case the output of a \
                `fatal!` was among them but none of them completed");
            }
            // Iterate over all events that currently are inside the queue and fish out any fatal.
            for _ in 0..(self.scheduler.item_count()) {
                let ((_ancestor, ev), _queue_kind) = self.runtime.block_on(self.scheduler.pop());
                if let Some(ctrl_ann) = ev.as_control() {
                    match ctrl_ann {
                        fatal @ ControlAnnouncement::FatalError { .. } => {
                            panic!(
                                "a control announcement requesting a fatal error was received: {}",
                                fatal
                            )
                        }
                    }
                } else {
                    debug!(?ev, "ignoring event while looking for a fatal")
                }
            }
            // Barring a `fatal`, the channel should never be closed, ever.
            panic!(
                "request for {} channel closed with return value \"{}\" in unit test harness",
                type_name::<T>(),
                err,
            );
        })
    }
pub ( crate ) fn send_request < C , T , F > ( & mut self , component : & mut C , f : F ) -> T where C : Component < REv > , < C as Component < REv > > :: Event : Send + ' static , T : Send + ' static , F : FnOnce ( Responder < T > ) -> C :: Event , REv : ReactorEvent , { // Prepare a channel. let ( sender , receiver ) = oneshot :: channel ( ) ; // Create response function. let responder = Responder :: without_shutdown ( sender ) ; // Create the event for the component. let request_event = f ( responder ) ; // Send directly to component. let returned_effects = self . send_event ( component , request_event ) ; // Execute the effects on our dedicated runtime, hopefully creating the responses. let mut join_handles = Vec :: new ( ) ; for effect in returned_effects { join_handles . push ( self . runtime . spawn ( effect ) ) ; } // Wait for a response to arrive. self . runtime . block_on ( receiver ) . unwrap_or_else ( | err | { // A channel was closed and this is usually an error. However, we consider all pending // events, in case we did get a control announcement requiring us to fatal error instead // before panicking on the basis of the missing response. // We give each of them a little time to produce the desired event. Note that `join_all` // should be safe to cancel, since we are only awaiting join handles. let join_all = async { for handle in join_handles { if let Err ( err ) = handle . await { warn ! ( " " err ) ; } ; } } ; if let Err ( _timeout ) = self . runtime . block_on ( async move { // Note: timeout can only be called from within a running running, this is why // we use an extra `async` block here. tokio :: time :: timeout ( FATAL_GRACE_TIME , join_all ) . await } ) { warn ! ( grace_time FATAL_GRACE_TIME " \
 \
 " ) ; } // Iterate over all events that currently are inside the queue and fish out any fatal. for _ in 0 .. ( self . scheduler . item_count ( ) ) { let ( ( _ancestor , ev ) , _queue_kind ) = self . runtime . block_on ( self . scheduler . pop ( ) ) ; if let Some ( ctrl_ann ) = ev . as_control ( ) { match ctrl_ann { fatal @ ControlAnnouncement :: FatalError { .. } => { panic ! ( " " fatal ) } } } else { debug ! ( ev " " ) } } // Barring a `fatal`, the channel should never be closed, ever. panic ! ( " \" \" " type_name T ( ) err ) ; } ) }
Sends a request, expecting an immediate response.
Sends a request expecting an immediate response .
&mut self
component
&mut C
f
F
&mut self
component
&mut C
f
F