use std::marker::PhantomData; use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed}; use futures::unsync::mpsc; use crate::dispatcher::FramedMessage; use crate::sink::Sink; pub struct Connect { io: Io, _t: PhantomData<(St, Codec)>, } impl Connect where Io: AsyncRead + AsyncWrite, { pub(crate) fn new(io: Io) -> Self { Self { io, _t: PhantomData, } } pub fn codec(self, codec: Codec) -> ConnectResult where Codec: Encoder + Decoder, { let (tx, rx) = mpsc::unbounded(); let sink = Sink::new(tx); ConnectResult { state: (), framed: Framed::new(self.io, codec), rx, sink, } } } pub struct ConnectResult { pub(crate) state: St, pub(crate) framed: Framed, pub(crate) rx: mpsc::UnboundedReceiver::Item>>, pub(crate) sink: Sink<::Item>, } impl ConnectResult { #[inline] pub fn sink(&self) -> &Sink<::Item> { &self.sink } #[inline] pub fn get_ref(&self) -> &Io { self.framed.get_ref() } #[inline] pub fn get_mut(&mut self) -> &mut Io { self.framed.get_mut() } #[inline] pub fn state(self, state: S) -> ConnectResult { ConnectResult { state, framed: self.framed, rx: self.rx, sink: self.sink, } } } impl futures::Stream for ConnectResult where Io: AsyncRead + AsyncWrite, Codec: Encoder + Decoder, { type Item = ::Item; type Error = ::Error; fn poll(&mut self) -> futures::Poll, Self::Error> { self.framed.poll() } } impl futures::Sink for ConnectResult where Io: AsyncRead + AsyncWrite, Codec: Encoder + Decoder, { type SinkItem = ::Item; type SinkError = ::Error; fn start_send( &mut self, item: Self::SinkItem, ) -> futures::StartSend { self.framed.start_send(item) } fn poll_complete(&mut self) -> futures::Poll<(), Self::SinkError> { self.framed.poll_complete() } fn close(&mut self) -> futures::Poll<(), Self::SinkError> { self.framed.close() } }