|
|
@ -10,23 +10,26 @@ use std::future::Future;
|
|
|
|
use std::pin::Pin;
|
|
|
|
use std::pin::Pin;
|
|
|
|
use std::task::Poll;
|
|
|
|
use std::task::Poll;
|
|
|
|
use std::time::Duration;
|
|
|
|
use std::time::Duration;
|
|
|
|
|
|
|
|
use futures_core::future::BoxFuture;
|
|
|
|
use tokio::sync::mpsc::Receiver;
|
|
|
|
use tokio::sync::mpsc::Receiver;
|
|
|
|
|
|
|
|
|
|
|
|
/// A metadata object returned after waiting for a reply to an event
|
|
|
|
/// A metadata object returned after waiting for a reply to an event
|
|
|
|
/// This object needs to be awaited for to get the actual reply
|
|
|
|
/// This object needs to be awaited for to get the actual reply
|
|
|
|
pub struct EmitMetadataWithResponseStream<P: IntoPayload> {
|
|
|
|
pub struct EmitMetadataWithResponseStream<P: IntoPayload> {
|
|
|
|
pub(crate) timeout: Option<Duration>,
|
|
|
|
pub(crate) timeout: Option<Duration>,
|
|
|
|
pub(crate) fut: Option<Pin<Box<dyn Future<Output = Result<ResponseStream>> + Send + Sync>>>,
|
|
|
|
pub(crate) fut: Option<BoxFuture<'static, Result<ResponseStream>>>,
|
|
|
|
pub(crate) emit_metadata: Option<EmitMetadata<P>>,
|
|
|
|
pub(crate) emit_metadata: Option<EmitMetadata<P>>,
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
type StreamFutureResult = Result<(Option<Event>, Context, Receiver<Event>)>;
|
|
|
|
|
|
|
|
|
|
|
|
/// An asynchronous stream one can read all responses to a specific event from.
|
|
|
|
/// An asynchronous stream one can read all responses to a specific event from.
|
|
|
|
pub struct ResponseStream {
|
|
|
|
pub struct ResponseStream {
|
|
|
|
event_id: u64,
|
|
|
|
event_id: u64,
|
|
|
|
ctx: Option<Context>,
|
|
|
|
ctx: Option<Context>,
|
|
|
|
receiver: Option<Receiver<Event>>,
|
|
|
|
receiver: Option<Receiver<Event>>,
|
|
|
|
timeout: Duration,
|
|
|
|
timeout: Duration,
|
|
|
|
fut: Option<Pin<Box<dyn Future<Output = Result<(Option<Event>, Context, Receiver<Event>)>>>>>,
|
|
|
|
fut: Option<BoxFuture<'static, StreamFutureResult>>,
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
impl ResponseStream {
|
|
|
|
impl ResponseStream {
|
|
|
@ -71,7 +74,7 @@ impl<P: IntoPayload + Send + Sync + 'static> Future for EmitMetadataWithResponse
|
|
|
|
let timeout = self
|
|
|
|
let timeout = self
|
|
|
|
.timeout
|
|
|
|
.timeout
|
|
|
|
.take()
|
|
|
|
.take()
|
|
|
|
.unwrap_or_else(|| ctx.default_reply_timeout.clone());
|
|
|
|
.unwrap_or(ctx.default_reply_timeout);
|
|
|
|
|
|
|
|
|
|
|
|
let event_id = match poll_unwrap!(emit_metadata.event_metadata.as_mut()).get_event() {
|
|
|
|
let event_id = match poll_unwrap!(emit_metadata.event_metadata.as_mut()).get_event() {
|
|
|
|
Ok(e) => e.id(),
|
|
|
|
Ok(e) => e.id(),
|
|
|
|