use std::any::TypeId;
use std::marker::PhantomData;
use std::sync::Arc;
use async_std::stream::StreamExt;
use async_std::sync::Mutex;
use futures::{channel::mpsc, SinkExt};
use ng_repo::errors::{NgError, ProtocolError, ServerError};
use ng_repo::log::*;
use crate::utils::{spawn_and_log_error, Receiver, ResultSend, Sender};
use crate::{connection::*, types::ProtocolMessage};
impl TryFrom<ProtocolMessage> for () {
type Error = ProtocolError;
fn try_from(_msg: ProtocolMessage) -> Result<Self, Self::Error> {
Ok(())
}
}
#[doc(hidden)]
#[async_trait::async_trait]
pub trait EActor: Send + Sync + std::fmt::Debug {
async fn respond(
&mut self,
msg: ProtocolMessage,
fsm: Arc<Mutex<NoiseFSM>>,
) -> Result<(), ProtocolError>;
fn set_id(&mut self, _id: i64) {}
}
#[derive(Debug)]
pub(crate) struct Actor<
'a,
A: Into<ProtocolMessage> + std::fmt::Debug,
B: TryFrom<ProtocolMessage, Error = ProtocolError> + std::fmt::Debug + Sync,
> {
id: i64,
phantom_a: PhantomData<&'a A>,
phantom_b: PhantomData<&'a B>,
receiver: Option<Receiver<ConnectionCommand>>,
receiver_tx: Sender<ConnectionCommand>,
}
#[derive(Debug)]
pub enum SoS<B> {
Single(B),
Stream(Receiver<B>),
}
impl<B> SoS<B> {
pub fn is_single(&self) -> bool {
if let Self::Single(_b) = self {
true
} else {
false
}
}
pub fn is_stream(&self) -> bool {
!self.is_single()
}
pub fn unwrap_single(self) -> B {
match self {
Self::Single(s) => s,
Self::Stream(_s) => {
panic!("called `unwrap_single()` on a `Stream` value")
}
}
}
pub fn unwrap_stream(self) -> Receiver<B> {
match self {
Self::Stream(s) => s,
Self::Single(_s) => {
panic!("called `unwrap_stream()` on a `Single` value")
}
}
}
}
impl<
A: Into<ProtocolMessage> + std::fmt::Debug + 'static,
B: TryFrom<ProtocolMessage, Error = ProtocolError> + Sync + Send + std::fmt::Debug + 'static,
> Actor<'_, A, B>
{
pub fn new(id: i64, _initiator: bool) -> Self {
let (receiver_tx, receiver) = mpsc::unbounded::<ConnectionCommand>();
Self {
id,
receiver: Some(receiver),
receiver_tx,
phantom_a: PhantomData,
phantom_b: PhantomData,
}
}
pub fn detach_receiver(&mut self) -> Receiver<ConnectionCommand> {
self.receiver.take().unwrap()
}
pub async fn request(
&mut self,
msg: ProtocolMessage,
fsm: Arc<Mutex<NoiseFSM>>,
) -> Result<SoS<B>, NgError> {
fsm.lock().await.send(msg).await?;
let mut receiver = self.receiver.take().unwrap();
match receiver.next().await {
Some(ConnectionCommand::Msg(msg)) => {
if let Some(bm) = msg.is_streamable() {
if bm.result() == Into::<u16>::into(ServerError::PartialContent)
&& TypeId::of::<B>() != TypeId::of::<()>()
{
let (mut b_sender, b_receiver) = mpsc::unbounded::<B>();
let response = msg.try_into().map_err(|e| {
log_err!("msg.try_into {}", e);
ProtocolError::ActorError
})?;
b_sender
.send(response)
.await
.map_err(|_err| ProtocolError::IoError)?;
async fn pump_stream<C: TryFrom<ProtocolMessage, Error = ProtocolError>>(
mut actor_receiver: Receiver<ConnectionCommand>,
mut sos_sender: Sender<C>,
fsm: Arc<Mutex<NoiseFSM>>,
id: i64,
) -> ResultSend<()> {
async move {
while let Some(ConnectionCommand::Msg(msg)) =
actor_receiver.next().await
{
if let Some(bm) = msg.is_streamable() {
if bm.result()
== Into::<u16>::into(ServerError::EndOfStream)
{
break;
}
let response = msg.try_into();
if response.is_err() {
break;
}
if sos_sender.send(response.unwrap()).await.is_err() {
break;
}
} else {
break;
}
}
fsm.lock().await.remove_actor(id).await;
}
.await;
Ok(())
}
spawn_and_log_error(pump_stream::<B>(
receiver,
b_sender,
Arc::clone(&fsm),
self.id,
));
return Ok(SoS::<B>::Stream(b_receiver));
}
}
fsm.lock().await.remove_actor(self.id).await;
let server_error: Result<ServerError, NgError> = (&msg).try_into();
if server_error.is_ok() {
return Err(NgError::ServerError(server_error.unwrap()));
}
let response: B = match msg.try_into() {
Ok(b) => b,
Err(ProtocolError::ServerError) => {
return Err(NgError::ServerError(server_error?));
}
Err(e) => return Err(NgError::ProtocolError(e)),
};
Ok(SoS::<B>::Single(response))
}
Some(ConnectionCommand::ProtocolError(e)) => Err(e.into()),
Some(ConnectionCommand::Error(e)) => Err(ProtocolError::from(e).into()),
Some(ConnectionCommand::Close) => Err(ProtocolError::Closing.into()),
_ => Err(ProtocolError::ActorError.into()),
}
}
pub fn new_responder(id: i64) -> Box<Self> {
Box::new(Self::new(id, false))
}
pub fn get_receiver_tx(&self) -> Sender<ConnectionCommand> {
self.receiver_tx.clone()
}
pub fn id(&self) -> i64 {
self.id
}
}
#[cfg(test)]
mod test {
use crate::actor::*;
use crate::actors::*;
#[async_std::test]
pub async fn test_actor() {
let _a = Actor::<Noise, Noise>::new(1, true);
}
}