mirror of https://github.com/Trivernis/bromine.git
GitHub
commit
891b688ff5
@ -1,288 +0,0 @@
|
||||
use std::future::Future;
|
||||
use std::mem;
|
||||
use std::ops::DerefMut;
|
||||
use std::pin::Pin;
|
||||
use std::sync::Arc;
|
||||
use std::task::Poll;
|
||||
use std::time::Duration;
|
||||
|
||||
use futures::future;
|
||||
use futures::future::Either;
|
||||
use tokio::io::{AsyncWrite, AsyncWriteExt};
|
||||
use tokio::sync::Mutex;
|
||||
use tracing;
|
||||
|
||||
use crate::error::{Error, Result};
|
||||
use crate::error_event::{ERROR_EVENT_NAME, ErrorEventData};
|
||||
use crate::events::event::Event;
|
||||
use crate::ipc::context::Context;
|
||||
use crate::payload::IntoPayload;
|
||||
use crate::protocol::AsyncProtocolStream;
|
||||
|
||||
macro_rules! poll_unwrap {
|
||||
($val:expr) => {
|
||||
if let Some(v) = $val {
|
||||
v
|
||||
} else {
|
||||
tracing::error!("Polling a future with an invalid state.");
|
||||
return Poll::Ready(Err(Error::InvalidState))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
type SendStream = Arc<Mutex<dyn AsyncWrite + Send + Sync + Unpin + 'static>>;
|
||||
|
||||
/// An abstraction over any type that implements the AsyncProtocolStream trait
|
||||
/// to emit events and share a connection across multiple
|
||||
/// contexts.
|
||||
#[derive(Clone)]
|
||||
pub struct StreamEmitter {
|
||||
stream: SendStream,
|
||||
}
|
||||
|
||||
impl StreamEmitter {
|
||||
pub fn new<P: AsyncProtocolStream + 'static>(stream: P::OwnedSplitWriteHalf) -> Self {
|
||||
Self {
|
||||
stream: Arc::new(Mutex::new(stream)),
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "trace", skip(self, ctx, payload))]
|
||||
fn _emit<P: IntoPayload>(
|
||||
&self,
|
||||
ctx: Context,
|
||||
namespace: Option<&str>,
|
||||
event: &str,
|
||||
payload: P,
|
||||
res_id: Option<u64>,
|
||||
) -> EmitMetadata<P> {
|
||||
EmitMetadata::new(ctx, self.stream.clone(), event.to_string(), namespace.map(|n| n.to_string()), payload, res_id)
|
||||
}
|
||||
|
||||
/// Emits an event
|
||||
#[inline]
|
||||
pub(crate) fn emit<S: AsRef<str>, P: IntoPayload>(
|
||||
&self,
|
||||
ctx: Context,
|
||||
event: S,
|
||||
payload: P,
|
||||
) -> EmitMetadata<P> {
|
||||
self._emit(ctx, None, event.as_ref(), payload, None)
|
||||
}
|
||||
|
||||
/// Emits an event to a specific namespace
|
||||
#[inline]
|
||||
pub(crate) fn emit_to<S1: AsRef<str>, S2: AsRef<str>, P: IntoPayload>(
|
||||
&self,
|
||||
ctx: Context,
|
||||
namespace: S1,
|
||||
event: S2,
|
||||
payload: P,
|
||||
) -> EmitMetadata<P> {
|
||||
self._emit(ctx, Some(namespace.as_ref()), event.as_ref(), payload, None)
|
||||
}
|
||||
|
||||
/// Emits a response to an event
|
||||
#[inline]
|
||||
pub(crate) fn emit_response<S: AsRef<str>, P: IntoPayload>(
|
||||
&self,
|
||||
ctx: Context,
|
||||
event_id: u64,
|
||||
event: S,
|
||||
payload: P,
|
||||
) -> EmitMetadata<P> {
|
||||
self._emit(ctx, None, event.as_ref(), payload, Some(event_id))
|
||||
}
|
||||
|
||||
/// Emits a response to an event to a namespace
|
||||
#[inline]
|
||||
pub(crate) fn emit_response_to<S1: AsRef<str>, S2: AsRef<str>, P: IntoPayload>(
|
||||
&self,
|
||||
ctx: Context,
|
||||
event_id: u64,
|
||||
namespace: S1,
|
||||
event: S2,
|
||||
payload: P,
|
||||
) -> EmitMetadata<P> {
|
||||
self._emit(
|
||||
ctx,
|
||||
Some(namespace.as_ref()),
|
||||
event.as_ref(),
|
||||
payload,
|
||||
Some(event_id),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
struct EventMetadata<P: IntoPayload> {
|
||||
event: Option<Event>,
|
||||
ctx: Option<Context>,
|
||||
event_namespace: Option<Option<String>>,
|
||||
event_name: Option<String>,
|
||||
res_id: Option<Option<u64>>,
|
||||
payload: Option<P>,
|
||||
}
|
||||
|
||||
impl<P: IntoPayload> EventMetadata<P> {
|
||||
pub fn get_event(&mut self) -> Result<&Event> {
|
||||
if self.event.is_none() {
|
||||
self.build_event()?;
|
||||
}
|
||||
Ok(self.event.as_ref().unwrap())
|
||||
}
|
||||
|
||||
pub fn take_event(mut self) -> Result<Option<Event>> {
|
||||
if self.event.is_none() {
|
||||
self.build_event()?;
|
||||
}
|
||||
Ok(mem::take(&mut self.event))
|
||||
}
|
||||
|
||||
fn build_event(&mut self) -> Result<()> {
|
||||
let ctx = self.ctx.take().ok_or(Error::InvalidState)?;
|
||||
let event = self.event_name.take().ok_or(Error::InvalidState)?;
|
||||
let namespace = self.event_namespace.take().ok_or(Error::InvalidState)?;
|
||||
let payload = self.payload.take().ok_or(Error::InvalidState)?;
|
||||
let res_id = self.res_id.take().ok_or(Error::InvalidState)?;
|
||||
let payload_bytes = payload.into_payload(&ctx)?;
|
||||
|
||||
let event = if let Some(namespace) = namespace {
|
||||
Event::with_namespace(namespace.to_string(), event.to_string(), payload_bytes, res_id)
|
||||
} else {
|
||||
Event::new(event.to_string(), payload_bytes, res_id)
|
||||
};
|
||||
self.event = Some(event);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
/// A metadata object returned after emitting an event.
|
||||
/// To send the event this object needs to be awaited
|
||||
/// This object can be used to wait for a response to an event.
|
||||
/// The result contains the emitted event id.
|
||||
pub struct EmitMetadata<P: IntoPayload> {
|
||||
event_metadata: Option<EventMetadata<P>>,
|
||||
stream: Option<SendStream>,
|
||||
fut: Option<Pin<Box<dyn Future<Output=Result<u64>> + Send + Sync>>>,
|
||||
}
|
||||
|
||||
/// A metadata object returned after waiting for a reply to an event
|
||||
/// This object needs to be awaited for to get the actual reply
|
||||
pub struct EmitMetadataWithResponse<P: IntoPayload> {
|
||||
timeout: Option<Duration>,
|
||||
fut: Option<Pin<Box<dyn Future<Output=Result<Event>> + Send + Sync>>>,
|
||||
emit_metadata: Option<EmitMetadata<P>>,
|
||||
}
|
||||
|
||||
impl<P: IntoPayload> EmitMetadata<P> {
|
||||
#[inline]
|
||||
pub(crate) fn new(ctx: Context, stream: SendStream, event_name: String, event_namespace: Option<String>, payload: P, res_id: Option<u64>) -> Self {
|
||||
Self {
|
||||
event_metadata: Some(EventMetadata {
|
||||
event: None,
|
||||
ctx: Some(ctx),
|
||||
event_name: Some(event_name),
|
||||
event_namespace: Some(event_namespace),
|
||||
payload: Some(payload),
|
||||
res_id: Some(res_id),
|
||||
}),
|
||||
stream: Some(stream),
|
||||
fut: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Waits for a reply to the given message.
|
||||
#[tracing::instrument(skip(self), fields(self.message_id))]
|
||||
pub fn await_reply(self) -> EmitMetadataWithResponse<P> {
|
||||
EmitMetadataWithResponse {
|
||||
timeout: None,
|
||||
fut: None,
|
||||
emit_metadata: Some(self),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<P: IntoPayload> EmitMetadataWithResponse<P> {
|
||||
/// Sets a timeout for awaiting replies to this emitted event
|
||||
#[inline]
|
||||
pub fn with_timeout(mut self, timeout: Duration) -> Self {
|
||||
self.timeout = Some(timeout);
|
||||
|
||||
self
|
||||
}
|
||||
}
|
||||
|
||||
impl<P: IntoPayload> Unpin for EmitMetadata<P> {}
|
||||
|
||||
impl<P: IntoPayload> Unpin for EmitMetadataWithResponse<P> {}
|
||||
|
||||
impl<P: IntoPayload + Send + Sync + 'static> Future for EmitMetadata<P> {
|
||||
type Output = Result<u64>;
|
||||
|
||||
fn poll(mut self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Self::Output> {
|
||||
if self.fut.is_none() {
|
||||
let event_metadata = poll_unwrap!(self.event_metadata.take());
|
||||
let stream = poll_unwrap!(self.stream.take());
|
||||
|
||||
let event = match event_metadata.take_event() {
|
||||
Ok(m) => { m }
|
||||
Err(e) => { return Poll::Ready(Err(e)); }
|
||||
}.expect("poll after future was done");
|
||||
|
||||
self.fut = Some(Box::pin(async move {
|
||||
let event_id = event.id();
|
||||
let event_bytes = event.into_bytes()?;
|
||||
let mut stream = stream.lock().await;
|
||||
stream.deref_mut().write_all(&event_bytes[..]).await?;
|
||||
tracing::trace!(bytes_len = event_bytes.len());
|
||||
|
||||
Ok(event_id)
|
||||
}));
|
||||
}
|
||||
self.fut.as_mut().unwrap().as_mut().poll(cx)
|
||||
}
|
||||
}
|
||||
|
||||
impl<P: IntoPayload + Send + Sync + 'static> Future for EmitMetadataWithResponse<P> {
|
||||
type Output = Result<Event>;
|
||||
|
||||
fn poll(mut self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Self::Output> {
|
||||
if self.fut.is_none() {
|
||||
let mut emit_metadata = poll_unwrap!(self.emit_metadata.take());
|
||||
let ctx = poll_unwrap!(emit_metadata.event_metadata.as_ref().and_then(|m| m.ctx.clone()));
|
||||
let timeout = self.timeout.clone().unwrap_or(ctx.default_reply_timeout.clone());
|
||||
|
||||
let event_id = match poll_unwrap!(emit_metadata.event_metadata.as_mut()).get_event() {
|
||||
Ok(e) => { e.id() }
|
||||
Err(e) => { return Poll::Ready(Err(e)); }
|
||||
};
|
||||
|
||||
self.fut = Some(Box::pin(async move {
|
||||
let tx = ctx.register_reply_listener(event_id).await?;
|
||||
emit_metadata.await?;
|
||||
|
||||
let result = future::select(
|
||||
Box::pin(tx),
|
||||
Box::pin(tokio::time::sleep(timeout)),
|
||||
)
|
||||
.await;
|
||||
|
||||
let reply = match result {
|
||||
Either::Left((tx_result, _)) => Ok(tx_result?),
|
||||
Either::Right(_) => {
|
||||
let mut listeners = ctx.reply_listeners.lock().await;
|
||||
listeners.remove(&event_id);
|
||||
Err(Error::Timeout)
|
||||
}
|
||||
}?;
|
||||
if reply.name() == ERROR_EVENT_NAME {
|
||||
Err(reply.payload::<ErrorEventData>()?.into())
|
||||
} else {
|
||||
Ok(reply)
|
||||
}
|
||||
}))
|
||||
}
|
||||
self.fut.as_mut().unwrap().as_mut().poll(cx)
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,101 @@
|
||||
use crate::context::Context;
|
||||
use crate::error::Error;
|
||||
use crate::event::EventType;
|
||||
use crate::ipc::stream_emitter::emit_metadata_with_response::EmitMetadataWithResponse;
|
||||
use crate::ipc::stream_emitter::emit_metadata_with_response_stream::EmitMetadataWithResponseStream;
|
||||
use crate::ipc::stream_emitter::event_metadata::EventMetadata;
|
||||
use crate::ipc::stream_emitter::SendStream;
|
||||
use crate::payload::IntoPayload;
|
||||
use crate::{error, poll_unwrap};
|
||||
use std::future::Future;
|
||||
use std::ops::DerefMut;
|
||||
use std::pin::Pin;
|
||||
use std::task::Poll;
|
||||
use tokio::io::AsyncWriteExt;
|
||||
|
||||
/// A metadata object returned after emitting an event.
|
||||
/// To send the event this object needs to be awaited
|
||||
/// This object can be used to wait for a response to an event.
|
||||
/// The result contains the emitted event id.
|
||||
pub struct EmitMetadata<P: IntoPayload> {
|
||||
pub(crate) event_metadata: Option<EventMetadata<P>>,
|
||||
stream: Option<SendStream>,
|
||||
fut: Option<Pin<Box<dyn Future<Output = error::Result<u64>> + Send + Sync>>>,
|
||||
}
|
||||
|
||||
impl<P: IntoPayload> EmitMetadata<P> {
|
||||
#[inline]
|
||||
pub(crate) fn new(
|
||||
ctx: Context,
|
||||
stream: SendStream,
|
||||
event_name: String,
|
||||
event_namespace: Option<String>,
|
||||
payload: P,
|
||||
res_id: Option<u64>,
|
||||
event_type: EventType,
|
||||
) -> Self {
|
||||
Self {
|
||||
event_metadata: Some(EventMetadata {
|
||||
event: None,
|
||||
ctx: Some(ctx),
|
||||
event_name: Some(event_name),
|
||||
event_namespace: Some(event_namespace),
|
||||
payload: Some(payload),
|
||||
res_id: Some(res_id),
|
||||
event_type: Some(event_type),
|
||||
}),
|
||||
stream: Some(stream),
|
||||
fut: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Waits for a reply to the given message.
|
||||
#[tracing::instrument(skip(self), fields(self.message_id))]
|
||||
pub fn await_reply(self) -> EmitMetadataWithResponse<P> {
|
||||
EmitMetadataWithResponse {
|
||||
timeout: None,
|
||||
fut: None,
|
||||
emit_metadata: Some(self),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn stream_replies(self) -> EmitMetadataWithResponseStream<P> {
|
||||
EmitMetadataWithResponseStream {
|
||||
timeout: None,
|
||||
fut: None,
|
||||
emit_metadata: Some(self),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<P: IntoPayload> Unpin for EmitMetadata<P> {}
|
||||
|
||||
impl<P: IntoPayload + Send + Sync + 'static> Future for EmitMetadata<P> {
|
||||
type Output = error::Result<u64>;
|
||||
|
||||
fn poll(mut self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Self::Output> {
|
||||
if self.fut.is_none() {
|
||||
let event_metadata = poll_unwrap!(self.event_metadata.take());
|
||||
let stream = poll_unwrap!(self.stream.take());
|
||||
|
||||
let event = match event_metadata.take_event() {
|
||||
Ok(m) => m,
|
||||
Err(e) => {
|
||||
return Poll::Ready(Err(e));
|
||||
}
|
||||
}
|
||||
.expect("poll after future was done");
|
||||
|
||||
self.fut = Some(Box::pin(async move {
|
||||
let event_id = event.id();
|
||||
let event_bytes = event.into_bytes()?;
|
||||
let mut stream = stream.lock().await;
|
||||
stream.deref_mut().write_all(&event_bytes[..]).await?;
|
||||
tracing::trace!(bytes_len = event_bytes.len());
|
||||
|
||||
Ok(event_id)
|
||||
}));
|
||||
}
|
||||
self.fut.as_mut().unwrap().as_mut().poll(cx)
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,84 @@
|
||||
use crate::context::Context;
|
||||
use crate::error::Error;
|
||||
use crate::error_event::ErrorEventData;
|
||||
use crate::event::{Event, EventType};
|
||||
use crate::ipc::stream_emitter::emit_metadata::EmitMetadata;
|
||||
use crate::payload::IntoPayload;
|
||||
use crate::{error, poll_unwrap};
|
||||
use std::future::Future;
|
||||
use std::pin::Pin;
|
||||
use std::task::Poll;
|
||||
use std::time::Duration;
|
||||
|
||||
/// A metadata object returned after waiting for a reply to an event
|
||||
/// This object needs to be awaited for to get the actual reply
|
||||
pub struct EmitMetadataWithResponse<P: IntoPayload> {
|
||||
pub(crate) timeout: Option<Duration>,
|
||||
pub(crate) fut: Option<Pin<Box<dyn Future<Output = error::Result<Event>> + Send + Sync>>>,
|
||||
pub(crate) emit_metadata: Option<EmitMetadata<P>>,
|
||||
}
|
||||
|
||||
impl<P: IntoPayload> EmitMetadataWithResponse<P> {
|
||||
/// Sets a timeout for awaiting replies to this emitted event
|
||||
#[inline]
|
||||
pub fn with_timeout(mut self, timeout: Duration) -> Self {
|
||||
self.timeout = Some(timeout);
|
||||
|
||||
self
|
||||
}
|
||||
}
|
||||
|
||||
impl<P: IntoPayload> Unpin for EmitMetadataWithResponse<P> {}
|
||||
|
||||
impl<P: IntoPayload + Send + Sync + 'static> Future for EmitMetadataWithResponse<P> {
|
||||
type Output = error::Result<Event>;
|
||||
|
||||
fn poll(mut self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Self::Output> {
|
||||
if self.fut.is_none() {
|
||||
let mut emit_metadata = poll_unwrap!(self.emit_metadata.take());
|
||||
let ctx = poll_unwrap!(emit_metadata
|
||||
.event_metadata
|
||||
.as_ref()
|
||||
.and_then(|m| m.ctx.clone()));
|
||||
let timeout = self
|
||||
.timeout
|
||||
.take()
|
||||
.unwrap_or_else(|| ctx.default_reply_timeout.clone());
|
||||
|
||||
let event_id = match poll_unwrap!(emit_metadata.event_metadata.as_mut()).get_event() {
|
||||
Ok(e) => e.id(),
|
||||
Err(e) => {
|
||||
return Poll::Ready(Err(e));
|
||||
}
|
||||
};
|
||||
|
||||
self.fut = Some(Box::pin(async move {
|
||||
let mut tx = ctx.register_reply_listener(event_id).await?;
|
||||
emit_metadata.await?;
|
||||
|
||||
let reply = tokio::select! {
|
||||
tx_result = tx.recv() => {
|
||||
Ok(tx_result.ok_or_else(|| Error::SendError)?)
|
||||
}
|
||||
_ = tokio::time::sleep(timeout) => {
|
||||
Err(Error::Timeout)
|
||||
}
|
||||
}?;
|
||||
|
||||
remove_reply_listener(&ctx, event_id).await;
|
||||
|
||||
if reply.event_type() == EventType::Error {
|
||||
Err(reply.payload::<ErrorEventData>()?.into())
|
||||
} else {
|
||||
Ok(reply)
|
||||
}
|
||||
}))
|
||||
}
|
||||
self.fut.as_mut().unwrap().as_mut().poll(cx)
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) async fn remove_reply_listener(ctx: &Context, event_id: u64) {
|
||||
let mut listeners = ctx.reply_listeners.lock().await;
|
||||
listeners.remove(&event_id);
|
||||
}
|
||||
@ -0,0 +1,151 @@
|
||||
use crate::context::Context;
|
||||
use crate::error::{Error, Result};
|
||||
use crate::event::{Event, EventType};
|
||||
use crate::ipc::stream_emitter::emit_metadata::EmitMetadata;
|
||||
use crate::ipc::stream_emitter::emit_metadata_with_response::remove_reply_listener;
|
||||
use crate::payload::IntoPayload;
|
||||
use crate::poll_unwrap;
|
||||
use futures_core::Stream;
|
||||
use std::future::Future;
|
||||
use std::pin::Pin;
|
||||
use std::task::Poll;
|
||||
use std::time::Duration;
|
||||
use tokio::sync::mpsc::Receiver;
|
||||
|
||||
/// A metadata object returned after waiting for a reply to an event
|
||||
/// This object needs to be awaited for to get the actual reply
|
||||
pub struct EmitMetadataWithResponseStream<P: IntoPayload> {
|
||||
pub(crate) timeout: Option<Duration>,
|
||||
pub(crate) fut: Option<Pin<Box<dyn Future<Output = Result<ResponseStream>> + Send + Sync>>>,
|
||||
pub(crate) emit_metadata: Option<EmitMetadata<P>>,
|
||||
}
|
||||
|
||||
/// An asynchronous stream one can read all responses to a specific event from.
|
||||
pub struct ResponseStream {
|
||||
event_id: u64,
|
||||
ctx: Option<Context>,
|
||||
receiver: Option<Receiver<Event>>,
|
||||
timeout: Duration,
|
||||
fut: Option<Pin<Box<dyn Future<Output = Result<(Option<Event>, Context, Receiver<Event>)>>>>>,
|
||||
}
|
||||
|
||||
impl ResponseStream {
|
||||
pub(crate) fn new(
|
||||
event_id: u64,
|
||||
timeout: Duration,
|
||||
ctx: Context,
|
||||
receiver: Receiver<Event>,
|
||||
) -> Self {
|
||||
Self {
|
||||
event_id,
|
||||
ctx: Some(ctx),
|
||||
receiver: Some(receiver),
|
||||
timeout,
|
||||
fut: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<P: IntoPayload> Unpin for EmitMetadataWithResponseStream<P> {}
|
||||
|
||||
impl<P: IntoPayload> EmitMetadataWithResponseStream<P> {
|
||||
/// Sets a timeout for awaiting replies to this emitted event
|
||||
#[inline]
|
||||
pub fn with_timeout(mut self, timeout: Duration) -> Self {
|
||||
self.timeout = Some(timeout);
|
||||
|
||||
self
|
||||
}
|
||||
}
|
||||
|
||||
impl<P: IntoPayload + Send + Sync + 'static> Future for EmitMetadataWithResponseStream<P> {
|
||||
type Output = Result<ResponseStream>;
|
||||
|
||||
fn poll(mut self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Self::Output> {
|
||||
if self.fut.is_none() {
|
||||
let mut emit_metadata = poll_unwrap!(self.emit_metadata.take());
|
||||
let ctx = poll_unwrap!(emit_metadata
|
||||
.event_metadata
|
||||
.as_ref()
|
||||
.and_then(|m| m.ctx.clone()));
|
||||
let timeout = self
|
||||
.timeout
|
||||
.take()
|
||||
.unwrap_or_else(|| ctx.default_reply_timeout.clone());
|
||||
|
||||
let event_id = match poll_unwrap!(emit_metadata.event_metadata.as_mut()).get_event() {
|
||||
Ok(e) => e.id(),
|
||||
Err(e) => {
|
||||
return Poll::Ready(Err(e));
|
||||
}
|
||||
};
|
||||
|
||||
self.fut = Some(Box::pin(async move {
|
||||
let tx = ctx.register_reply_listener(event_id).await?;
|
||||
emit_metadata.await?;
|
||||
|
||||
Ok(ResponseStream::new(event_id, timeout, ctx, tx))
|
||||
}))
|
||||
}
|
||||
self.fut.as_mut().unwrap().as_mut().poll(cx)
|
||||
}
|
||||
}
|
||||
|
||||
impl Unpin for ResponseStream {}
|
||||
|
||||
impl Stream for ResponseStream {
|
||||
type Item = Result<Event>;
|
||||
|
||||
fn poll_next(
|
||||
mut self: Pin<&mut Self>,
|
||||
cx: &mut std::task::Context<'_>,
|
||||
) -> Poll<Option<Self::Item>> {
|
||||
if self.fut.is_none() {
|
||||
if self.ctx.is_none() || self.receiver.is_none() {
|
||||
return Poll::Ready(None);
|
||||
}
|
||||
let ctx = self.ctx.take().unwrap();
|
||||
let mut receiver = self.receiver.take().unwrap();
|
||||
let timeout = self.timeout;
|
||||
let event_id = self.event_id;
|
||||
|
||||
self.fut = Some(Box::pin(async move {
|
||||
let event: Option<Event> = tokio::select! {
|
||||
tx_result = receiver.recv() => {
|
||||
Ok(tx_result)
|
||||
}
|
||||
_ = tokio::time::sleep(timeout) => {
|
||||
Err(Error::Timeout)
|
||||
}
|
||||
}?;
|
||||
|
||||
if event.is_none() || event.as_ref().unwrap().event_type() == EventType::End {
|
||||
remove_reply_listener(&ctx, event_id).await;
|
||||
}
|
||||
|
||||
Ok((event, ctx, receiver))
|
||||
}));
|
||||
}
|
||||
|
||||
match self.fut.as_mut().unwrap().as_mut().poll(cx) {
|
||||
Poll::Ready(r) => match r {
|
||||
Ok((event, ctx, tx)) => {
|
||||
self.fut = None;
|
||||
|
||||
if let Some(event) = event {
|
||||
if event.event_type() != EventType::End {
|
||||
self.ctx = Some(ctx);
|
||||
self.receiver = Some(tx);
|
||||
}
|
||||
|
||||
Poll::Ready(Some(Ok(event)))
|
||||
} else {
|
||||
Poll::Ready(None)
|
||||
}
|
||||
}
|
||||
Err(e) => Poll::Ready(Some(Err(e))),
|
||||
},
|
||||
Poll::Pending => Poll::Pending,
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,54 @@
|
||||
use crate::context::Context;
|
||||
use crate::error;
|
||||
use crate::error::Error;
|
||||
use crate::event::{Event, EventType};
|
||||
use crate::payload::IntoPayload;
|
||||
use std::mem;
|
||||
|
||||
pub(crate) struct EventMetadata<P: IntoPayload> {
|
||||
pub(crate) event: Option<Event>,
|
||||
pub(crate) ctx: Option<Context>,
|
||||
pub(crate) event_namespace: Option<Option<String>>,
|
||||
pub(crate) event_name: Option<String>,
|
||||
pub(crate) res_id: Option<Option<u64>>,
|
||||
pub(crate) event_type: Option<EventType>,
|
||||
pub(crate) payload: Option<P>,
|
||||
}
|
||||
|
||||
impl<P: IntoPayload> EventMetadata<P> {
|
||||
pub fn get_event(&mut self) -> error::Result<&Event> {
|
||||
if self.event.is_none() {
|
||||
self.build_event()?;
|
||||
}
|
||||
Ok(self.event.as_ref().unwrap())
|
||||
}
|
||||
|
||||
pub fn take_event(mut self) -> error::Result<Option<Event>> {
|
||||
if self.event.is_none() {
|
||||
self.build_event()?;
|
||||
}
|
||||
Ok(mem::take(&mut self.event))
|
||||
}
|
||||
|
||||
fn build_event(&mut self) -> error::Result<()> {
|
||||
let ctx = self.ctx.take().ok_or(Error::InvalidState)?;
|
||||
let event = self.event_name.take().ok_or(Error::InvalidState)?;
|
||||
let namespace = self.event_namespace.take().ok_or(Error::InvalidState)?;
|
||||
let payload = self.payload.take().ok_or(Error::InvalidState)?;
|
||||
let res_id = self.res_id.take().ok_or(Error::InvalidState)?;
|
||||
let event_type = self.event_type.take().ok_or(Error::InvalidState)?;
|
||||
let payload_bytes = payload.into_payload(&ctx)?;
|
||||
|
||||
let event = Event::new(
|
||||
namespace,
|
||||
event.to_string(),
|
||||
payload_bytes,
|
||||
res_id,
|
||||
event_type,
|
||||
);
|
||||
|
||||
self.event = Some(event);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,159 @@
|
||||
pub mod emit_metadata;
|
||||
pub mod emit_metadata_with_response;
|
||||
pub mod emit_metadata_with_response_stream;
|
||||
mod event_metadata;
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use tokio::io::AsyncWrite;
|
||||
use tokio::sync::Mutex;
|
||||
use tracing;
|
||||
|
||||
use crate::event::EventType;
|
||||
use crate::ipc::context::Context;
|
||||
use crate::payload::IntoPayload;
|
||||
use crate::protocol::AsyncProtocolStream;
|
||||
|
||||
pub use emit_metadata_with_response_stream::ResponseStream;
|
||||
use crate::prelude::emit_metadata::EmitMetadata;
|
||||
|
||||
#[macro_export]
|
||||
macro_rules! poll_unwrap {
|
||||
($val:expr) => {
|
||||
if let Some(v) = $val {
|
||||
v
|
||||
} else {
|
||||
tracing::error!("Polling a future with an invalid state.");
|
||||
return Poll::Ready(Err(Error::InvalidState));
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
type SendStream = Arc<Mutex<dyn AsyncWrite + Send + Sync + Unpin + 'static>>;
|
||||
|
||||
/// An abstraction over any type that implements the AsyncProtocolStream trait
|
||||
/// to emit events and share a connection across multiple
|
||||
/// contexts.
|
||||
#[derive(Clone)]
|
||||
pub struct StreamEmitter {
|
||||
stream: SendStream,
|
||||
}
|
||||
|
||||
impl StreamEmitter {
|
||||
pub fn new<P: AsyncProtocolStream + 'static>(stream: P::OwnedSplitWriteHalf) -> Self {
|
||||
Self {
|
||||
stream: Arc::new(Mutex::new(stream)),
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "trace", skip(self, ctx, payload))]
|
||||
fn _emit<P: IntoPayload>(
|
||||
&self,
|
||||
ctx: Context,
|
||||
namespace: Option<String>,
|
||||
event: &str,
|
||||
payload: P,
|
||||
res_id: Option<u64>,
|
||||
event_type: EventType,
|
||||
) -> EmitMetadata<P> {
|
||||
EmitMetadata::new(
|
||||
ctx,
|
||||
self.stream.clone(),
|
||||
event.to_string(),
|
||||
namespace,
|
||||
payload,
|
||||
res_id,
|
||||
event_type,
|
||||
)
|
||||
}
|
||||
|
||||
/// Emits an event
|
||||
#[inline]
|
||||
pub(crate) fn emit<S: AsRef<str>, P: IntoPayload>(
|
||||
&self,
|
||||
ctx: Context,
|
||||
event: S,
|
||||
payload: P,
|
||||
) -> EmitMetadata<P> {
|
||||
self._emit(
|
||||
ctx,
|
||||
None,
|
||||
event.as_ref(),
|
||||
payload,
|
||||
None,
|
||||
EventType::Initiator,
|
||||
)
|
||||
}
|
||||
|
||||
/// Emits an event to a specific namespace
|
||||
#[inline]
|
||||
pub(crate) fn emit_to<S1: AsRef<str>, S2: AsRef<str>, P: IntoPayload>(
|
||||
&self,
|
||||
ctx: Context,
|
||||
namespace: S1,
|
||||
event: S2,
|
||||
payload: P,
|
||||
) -> EmitMetadata<P> {
|
||||
self._emit(
|
||||
ctx,
|
||||
Some(namespace.as_ref().to_string()),
|
||||
event.as_ref(),
|
||||
payload,
|
||||
None,
|
||||
EventType::Initiator,
|
||||
)
|
||||
}
|
||||
|
||||
/// Emits a raw event
|
||||
#[inline]
|
||||
pub(crate) fn emit_raw<P: IntoPayload>(
|
||||
&self,
|
||||
ctx: Context,
|
||||
res_id: Option<u64>,
|
||||
event: &str,
|
||||
namespace: Option<String>,
|
||||
event_type: EventType,
|
||||
payload: P,
|
||||
) -> EmitMetadata<P> {
|
||||
self._emit(ctx, namespace, event, payload, res_id, event_type)
|
||||
}
|
||||
|
||||
/// Emits a response to an event
|
||||
#[inline]
|
||||
pub(crate) fn emit_response<S: AsRef<str>, P: IntoPayload>(
|
||||
&self,
|
||||
ctx: Context,
|
||||
event_id: u64,
|
||||
event: S,
|
||||
payload: P,
|
||||
) -> EmitMetadata<P> {
|
||||
self._emit(
|
||||
ctx,
|
||||
None,
|
||||
event.as_ref(),
|
||||
payload,
|
||||
Some(event_id),
|
||||
EventType::Response,
|
||||
)
|
||||
}
|
||||
|
||||
/// Emits a response to an event to a namespace
|
||||
#[inline]
|
||||
pub(crate) fn emit_response_to<S1: AsRef<str>, S2: AsRef<str>, P: IntoPayload>(
|
||||
&self,
|
||||
ctx: Context,
|
||||
event_id: u64,
|
||||
namespace: S1,
|
||||
event: S2,
|
||||
payload: P,
|
||||
) -> EmitMetadata<P> {
|
||||
self._emit(
|
||||
ctx,
|
||||
Some(namespace.as_ref().to_string()),
|
||||
event.as_ref(),
|
||||
payload,
|
||||
Some(event_id),
|
||||
EventType::Response,
|
||||
)
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,88 @@
|
||||
use crate::utils::call_counter::{get_counter_from_context, increment_counter_for_event};
|
||||
use crate::utils::protocol::TestProtocolListener;
|
||||
use crate::utils::{get_free_port, start_server_and_client};
|
||||
use bromine::prelude::*;
|
||||
use byteorder::ReadBytesExt;
|
||||
use futures::StreamExt;
|
||||
use std::io::Read;
|
||||
use std::time::Duration;
|
||||
|
||||
mod utils;
|
||||
|
||||
/// When awaiting the reply to an event the handler for the event doesn't get called.
|
||||
/// Therefore we expect it to have a call count of 0.
|
||||
#[tokio::test]
|
||||
async fn it_receives_responses() {
|
||||
let port = get_free_port();
|
||||
let ctx = get_client_with_server(port).await;
|
||||
let mut reply_stream = ctx
|
||||
.emit("stream", EmptyPayload)
|
||||
.stream_replies()
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let mut reply_stream_2 = ctx
|
||||
.emit("stream", EmptyPayload)
|
||||
.stream_replies()
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
for i in 0u8..=100 {
|
||||
if let Some(Ok(event)) = reply_stream.next().await {
|
||||
assert_eq!(event.payload::<NumberPayload>().unwrap().0, i)
|
||||
} else {
|
||||
panic!("stream 1 has no value {}", i);
|
||||
}
|
||||
if let Some(Ok(event)) = reply_stream_2.next().await {
|
||||
assert_eq!(event.payload::<NumberPayload>().unwrap().0, i)
|
||||
} else {
|
||||
panic!("stream 2 has no value {}", i);
|
||||
}
|
||||
}
|
||||
let counter = get_counter_from_context(&ctx).await;
|
||||
assert_eq!(counter.get("stream").await, 2);
|
||||
}
|
||||
|
||||
async fn get_client_with_server(port: u8) -> Context {
|
||||
start_server_and_client(move || get_builder(port)).await
|
||||
}
|
||||
|
||||
fn get_builder(port: u8) -> IPCBuilder<TestProtocolListener> {
|
||||
IPCBuilder::new()
|
||||
.address(port)
|
||||
.timeout(Duration::from_millis(100))
|
||||
.on("stream", callback!(handle_stream_event))
|
||||
}
|
||||
|
||||
async fn handle_stream_event(ctx: &Context, event: Event) -> IPCResult<Response> {
|
||||
increment_counter_for_event(ctx, &event).await;
|
||||
for i in 0u8..=99 {
|
||||
ctx.emit("number", NumberPayload(i)).await?;
|
||||
}
|
||||
|
||||
ctx.response(NumberPayload(100))
|
||||
}
|
||||
|
||||
pub struct EmptyPayload;
|
||||
|
||||
impl IntoPayload for EmptyPayload {
|
||||
fn into_payload(self, _: &Context) -> IPCResult<Vec<u8>> {
|
||||
Ok(vec![])
|
||||
}
|
||||
}
|
||||
|
||||
pub struct NumberPayload(u8);
|
||||
|
||||
impl IntoPayload for NumberPayload {
|
||||
fn into_payload(self, _: &Context) -> IPCResult<Vec<u8>> {
|
||||
Ok(vec![self.0])
|
||||
}
|
||||
}
|
||||
|
||||
impl FromPayload for NumberPayload {
|
||||
fn from_payload<R: Read>(mut reader: R) -> IPCResult<Self> {
|
||||
let num = reader.read_u8()?;
|
||||
|
||||
Ok(NumberPayload(num))
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue