scatterbrain 0.0.1

use std::io::{ErrorKind, Read, Write};

pub use crate::api::proto::ProtoUuid;
use crate::api::proto::{MessageType, TypePrefix, UnitResponse};
use crate::constants::{MESSAGE_SIZE_CAP, TYPE_SIZE_CAP};
use crate::error::{Error, IntoRemoteErr, SbResult};
use crate::types::GetType;
use byteorder::{BigEndian, ReadBytesExt, WriteBytesExt};
use crc::Crc;
#[cfg(feature = "flutter")]
use flutter_rust_bridge::DartFnFuture;
use prost::{bytes::BufMut, Message};
use tokio::io::{AsyncReadExt, AsyncWriteExt};

#[cfg(feature = "flutter")]
pub use super::api::SbSession;

// java uses CRC32 from GZIP RFC1952
const JAVA_ALG: crc::Algorithm<u32> = crc::Algorithm {
    width: 32,
    poly: 0x04C11DB7,
    init: 0xFFFFFFFF,
    refin: true,
    refout: true,
    xorout: 0xFFFFFFFF,
    check: 0xaee7,
    residue: 0x0000,
};

pub trait ToUuid {
    fn as_uuid(&self) -> uuid::Uuid;
    fn as_proto(&self) -> ProtoUuid;
}

impl ToUuid for ProtoUuid {
    fn as_uuid(&self) -> uuid::Uuid {
        uuid::Uuid::from_u64_pair(self.upper, self.lower)
    }

    fn as_proto(&self) -> ProtoUuid {
        *self
    }
}

impl ToUuid for uuid::Uuid {
    fn as_uuid(&self) -> uuid::Uuid {
        *self
    }

    fn as_proto(&self) -> ProtoUuid {
        let (upper, lower) = self.as_u64_pair();
        ProtoUuid { upper, lower }
    }
}

pub struct ProtoStream<A> {
    stream: A,
    pub is_disconnected: bool,
    #[cfg(feature = "flutter")]
    pub(crate) on_connect:
        Option<Box<dyn Fn(Option<SbSession>) -> DartFnFuture<()> + Send + Sync + 'static>>,
}

impl<A> Clone for ProtoStream<A>
where
    A: Clone,
{
    fn clone(&self) -> Self {
        ProtoStream {
            stream: self.stream.clone(),
            is_disconnected: self.is_disconnected,
            #[cfg(feature = "flutter")]
            on_connect: None,
        }
    }
}

#[derive(Debug, Default)]
pub struct TypedMessage<M>
where
    M: Message + GetType + Default + Send,
{
    pub message: M,
    pub message_type: MessageType,
}

impl<M> TypedMessage<M>
where
    M: Message + GetType + Default + Send,
{
    pub fn new_typed(message_type: MessageType) -> Self {
        Self {
            message: M::default(),
            message_type,
        }
    }

    pub fn new(message: M) -> Self {
        let message_type = M::get_type();
        Self {
            message,
            message_type,
        }
    }
}

impl<M> Message for TypedMessage<M>
where
    M: Message + GetType + Default + Send,
{
    fn clear(&mut self) {
        self.message.clear()
    }

    fn encode_raw(&self, buf: &mut impl BufMut)
    where
        Self: Sized,
    {
        self.message.encode_raw(buf)
    }

    fn merge_field(
        &mut self,
        tag: u32,
        wire_type: ::prost::encoding::WireType,
        buf: &mut impl prost::bytes::Buf,
        ctx: ::prost::encoding::DecodeContext,
    ) -> std::result::Result<(), prost::DecodeError>
    where
        Self: Sized,
    {
        self.message.merge_field(tag, wire_type, buf, ctx)
    }

    fn encode(&self, buf: &mut impl BufMut) -> std::result::Result<(), prost::EncodeError>
    where
        Self: Sized,
    {
        self.message.encode(buf)
    }

    fn encoded_len(&self) -> usize {
        self.message.encoded_len()
    }

    fn encode_to_vec(&self) -> Vec<u8>
    where
        Self: Sized,
    {
        self.message.encode_to_vec()
    }

    fn merge(&mut self, buf: impl prost::bytes::Buf) -> std::result::Result<(), prost::DecodeError>
    where
        Self: Sized,
    {
        self.message.merge(buf)
    }

    fn decode(buf: impl prost::bytes::Buf) -> std::result::Result<Self, prost::DecodeError>
    where
        Self: Default,
    {
        Ok(Self::new(M::decode(buf)?))
    }
}

impl<A> ProtoStream<A>
where
    A: Unpin + Send,
{
    pub fn new(sock: A) -> Self {
        Self {
            stream: sock,
            is_disconnected: false,
            #[cfg(feature = "flutter")]
            on_connect: None,
        }
    }

    pub fn write_message_sync<M>(&mut self, message: &M) -> SbResult<()>
    where
        M: Message + GetType + Default + Send,
        A: Write,
    {
        let crc = Crc::<u32>::new(&JAVA_ALG);
        let mut digest = crc.digest();
        let message = message.encode_to_vec();
        let size = message.len() as i32;

        let tp = TypePrefix {
            message_type: M::get_type().into(),
        };

        let tp = tp.encode_to_vec();

        let typesize = tp.len() as i32;
        digest.update(&typesize.to_be_bytes());
        digest.update(&size.to_be_bytes());
        digest.update(&tp);
        digest.update(&message);
        self.stream.write_i32::<BigEndian>(typesize)?;
        self.stream.write_i32::<BigEndian>(size)?;

        self.stream.write(&tp)?;
        self.stream.write(&message)?;

        self.stream.write_u32::<BigEndian>(digest.finalize())?;

        Ok(())
    }

    pub async fn write_message<M>(&mut self, message: &M) -> SbResult<()>
    where
        M: Message + GetType + Default + Send,
        A: AsyncWriteExt,
    {
        let crc = Crc::<u32>::new(&JAVA_ALG);
        let mut digest = crc.digest();
        let message = message.encode_to_vec();
        let size = message.len() as i32;

        let tp = TypePrefix {
            message_type: M::get_type().into(),
        };

        let tp = tp.encode_to_vec();

        let typesize = tp.len() as i32;
        digest.update(&typesize.to_be_bytes());
        digest.update(&size.to_be_bytes());
        digest.update(&tp);
        digest.update(&message);
        self.stream.write_i32(typesize).await?;
        self.stream.write_i32(size).await?;

        self.stream.write(&tp).await?;
        self.stream.write(&message).await?;

        self.stream.write_u32(digest.finalize()).await?;

        Ok(())
    }

    pub fn read_message_sync<M>(&mut self) -> SbResult<M>
    where
        M: Message + GetType + Default + Send,
        A: Read,
    {
        let crc = Crc::<u32>::new(&JAVA_ALG);
        let mut digest = crc.digest();

        let typesize: i32 = self.stream.read_i32::<BigEndian>()?;
        let size = self.stream.read_i32::<BigEndian>()?;

        log::debug!("receivied message sizes test {} {}", typesize, size);

        digest.update(&typesize.to_be_bytes());
        digest.update(&size.to_be_bytes());

        if size > MESSAGE_SIZE_CAP as i32 {
            return Err(Error::MessageSizeError(size as usize));
        }

        if typesize > TYPE_SIZE_CAP as i32 {
            return Err(Error::MessageSizeError(typesize as usize));
        }

        let mut mb = vec![0; typesize as usize];
        self.stream.read_exact(mb.as_mut_slice())?;
        digest.update(mb.as_slice());
        let tp = TypePrefix::decode(mb.as_slice())?;

        log::debug!(
            "read type prefix: expected={} got={}",
            M::get_type().as_str_name(),
            tp.message_type().as_str_name()
        );

        if M::get_type() != tp.message_type() {
            return Err(Error::TypeMismatch {
                expected: M::get_type().as_str_name().to_owned(),
                actual: tp.message_type().as_str_name().to_owned(),
            });
        }

        let mut mb = vec![0; size as usize];
        self.stream.read_exact(mb.as_mut_slice())?;
        digest.update(mb.as_slice());
        let m = M::decode(mb.as_slice())?;
        let crc = self.stream.read_u32::<BigEndian>()?;
        let mycrc = digest.finalize();
        log::debug!("message = {:?}", m);
        log::debug!("received CRC thiers={} ours={}", crc, mycrc);
        if crc != mycrc {
            return Err(Error::CrcMismatch);
        }
        Ok(m)
    }

    pub async fn read_message<M>(&mut self) -> SbResult<M>
    where
        M: Message + GetType + Default + Send,
        A: AsyncReadExt,
    {
        match self.read_message_impl().await {
            Ok(m) => Ok(m),
            Err(err) => match err {
                Error::IoError(err) => {
                    match err.kind() {
                        ErrorKind::ConnectionAborted
                        | ErrorKind::UnexpectedEof
                        | ErrorKind::ConnectionReset => {
                            self.is_disconnected = true;
                            #[cfg(feature = "flutter")]
                            if let Some(on_disconnect) = self.on_connect.as_ref() {
                                on_disconnect(None).await;
                            }
                        }
                        _ => (),
                    }
                    Err(Error::IoError(err))
                }
                e => Err(e),
            },
        }
    }

    async fn read_message_impl<M>(&mut self) -> SbResult<M>
    where
        M: Message + GetType + Default + Send,
        A: AsyncReadExt,
    {
        let crc = Crc::<u32>::new(&JAVA_ALG);
        let mut digest = crc.digest();

        let typesize = self.stream.read_i32().await?;
        let size = self.stream.read_i32().await?;

        log::debug!("receivied message sizes test {} {}", typesize, size);
        digest.update(&typesize.to_be_bytes());
        digest.update(&size.to_be_bytes());

        if size > MESSAGE_SIZE_CAP as i32 {
            return Err(Error::MessageSizeError(size as usize));
        }

        if typesize > TYPE_SIZE_CAP as i32 {
            return Err(Error::MessageSizeError(typesize as usize));
        }

        let mut mb = vec![0; typesize as usize];
        self.stream.read_exact(mb.as_mut_slice()).await?;
        digest.update(mb.as_slice());
        let tp = TypePrefix::decode(mb.as_slice())?;

        log::debug!(
            "read type prefix: expected={} got={}",
            M::get_type().as_str_name(),
            tp.message_type().as_str_name()
        );

        if M::get_type() != tp.message_type() {
            if tp.message_type() == MessageType::UnitResponse {
                let mut mb = vec![0; size as usize];
                self.stream.read_exact(mb.as_mut_slice()).await?;
                digest.update(mb.as_slice());
                let m = UnitResponse::decode(mb.as_slice())?;
                let crc = self.stream.read_u32().await?;
                let mycrc = digest.finalize();
                log::debug!("received CRC thiers={} ours={}", crc, mycrc);
                if crc != mycrc {
                    return Err(Error::CrcMismatch);
                }
                m.into_remote_err()?;
            }
            return Err(Error::TypeMismatch {
                expected: M::get_type().as_str_name().to_owned(),
                actual: tp.message_type().as_str_name().to_owned(),
            });
        }

        let mut mb = vec![0; size as usize];
        self.stream.read_exact(mb.as_mut_slice()).await?;
        digest.update(mb.as_slice());
        let m = M::decode(mb.as_slice())?;
        let crc = self.stream.read_u32().await?;
        let mycrc = digest.finalize();
        log::debug!("message = {:?}", m);
        log::debug!("received CRC thiers={} ours={}", crc, mycrc);
        if crc != mycrc {
            return Err(Error::CrcMismatch);
        }
        Ok(m)
    }
}

#[cfg(test)]
mod test {

    use std::sync::Arc;
    use std::time::Duration;

    use crate::connection::SessionTrait;
    use crate::crypto::SessionState;
    use crate::{
        api::proto::{ack::*, *},
        crypto::{CryptoMessageWrapper, KxSession},
        types::{GetMessagesCmd, MessageResponse},
    };
    use chrono::Utc;
    use dryoc::constants::CRYPTO_KX_SESSIONKEYBYTES;
    use dryoc::types::ByteArray;
    use serde::Serialize;
    use tokio::sync::RwLock;
    use uuid::Uuid;
    use zeroize::Zeroize;

    #[derive(Serialize)]
    pub struct FakeSession<SessionKey: ByteArray<{ CRYPTO_KX_SESSIONKEYBYTES }> + Zeroize> {
        rx_key: SessionKey,
        tx_key: SessionKey,
    }

    use super::*;
    #[tokio::test]
    async fn test_kotlin_ack() {
        let _ = env_logger::try_init();
        let st = tokio::fs::File::open("./src/test/ack-stream")
            .await
            .expect("failed to open test file");
        let mut reader = ProtoStream::new(st);
        let mesage: Ack = reader.read_message().await.expect("failed to read message");
        assert!(mesage.success);
    }

    #[tokio::test]
    async fn test_readwrite() {
        let ack = Ack {
            success: true,
            status: 1,
            ack_maybe_message: Some(AckMaybeMessage::Text("tests".to_owned())),
        };

        let (client, server) = tokio::io::duplex(64);
        let mut client = ProtoStream::new(client);
        let mut server = ProtoStream::new(server);
        client.write_message(&ack).await.expect("failed to write");
        let newack: Ack = server.read_message().await.expect("failed to read");

        assert_eq!(ack, newack);
    }

    #[tokio::test]
    async fn bad_message() {
        let bytes: Vec<u8> = vec![
            0x00, 0x00, 0x00, 0x02, 0x00, 0x00, 0x06, 0x68, 0x08, 0x38, 0x0a, 0x18, 0xe7, 0xa4,
            0x39, 0x34, 0x7c, 0x1e, 0xeb, 0x8c, 0xe6, 0xba, 0x42, 0x75, 0x29, 0x86, 0x97, 0xab,
            0x74, 0x82, 0x13, 0xcf, 0x76, 0x16, 0x5e, 0xd9, 0x12, 0xcb, 0x0c, 0x05, 0x9d, 0x06,
            0xb3, 0x84, 0x4e, 0x15, 0x62, 0xdc, 0xe6, 0xc4, 0xbf, 0x05, 0x9d, 0x17, 0x63, 0x8c,
            0x77, 0x83, 0x6a, 0x8f, 0x58, 0x4b, 0x54, 0xd1, 0x06, 0x81, 0x1b, 0xd2, 0xd5, 0x4c,
            0x8e, 0xda, 0x19, 0xb3, 0x90, 0xd1, 0x3b, 0x08, 0xb7, 0xd3, 0xb2, 0x6d, 0xb1, 0x85,
            0xd8, 0xfd, 0x67, 0x4f, 0xea, 0xe7, 0x58, 0x6c, 0x0b, 0x28, 0xcf, 0xdf, 0x10, 0x21,
            0x48, 0x78, 0x3d, 0x22, 0x2c, 0x90, 0x26, 0xde, 0x44, 0x15, 0xbb, 0x31, 0x52, 0x0a,
            0xac, 0xd9, 0x95, 0xc1, 0x08, 0x6e, 0x26, 0x96, 0x67, 0x92, 0xa5, 0xa0, 0x92, 0xee,
            0xed, 0xa5, 0xc5, 0x7f, 0xc8, 0xb5, 0xcc, 0x15, 0xf1, 0xf2, 0x52, 0xd4, 0x52, 0x24,
            0xb8, 0x57, 0x5b, 0x20, 0x4f, 0x26, 0x38, 0xd8, 0x27, 0x65, 0xfc, 0x34, 0x76, 0xfe,
            0xf5, 0x6c, 0xd5, 0x69, 0xba, 0x43, 0x21, 0x50, 0x04, 0x44, 0xc7, 0x96, 0xdf, 0x99,
            0xf7, 0xac, 0xf5, 0x1b, 0x5f, 0x72, 0xbc, 0xb2, 0x19, 0x59, 0xdd, 0x6e, 0x3a, 0xc0,
            0x10, 0xef, 0x66, 0x19, 0xa2, 0x77, 0xa2, 0xc2, 0x98, 0xad, 0x29, 0x5e, 0xc6, 0xf1,
            0x32, 0xc7, 0x44, 0x5d, 0x27, 0xde, 0x63, 0x52, 0x75, 0x2b, 0x3e, 0x6b, 0x72, 0x00,
            0xf9, 0x2c, 0x51, 0xb1, 0x77, 0x89, 0x70, 0x0d, 0xa4, 0xde, 0x25, 0x61, 0xea, 0x4d,
            0x4e, 0xad, 0x46, 0xed, 0xd3, 0xea, 0xbc, 0x66, 0x1f, 0xf5, 0x70, 0x03, 0x72, 0xe7,
            0xe6, 0x20, 0x24, 0xae, 0x27, 0x1f, 0xda, 0xd3, 0x07, 0xe9, 0x64, 0x33, 0x41, 0x4c,
            0x1a, 0x50, 0x9f, 0xc3, 0x30, 0x95, 0x20, 0x30, 0x62, 0xc0, 0xaa, 0x01, 0xf6, 0x6e,
            0x2f, 0x5c, 0x06, 0x5d, 0xf1, 0x34, 0x78, 0x39, 0xd5, 0x98, 0x09, 0x43, 0x20, 0xb0,
            0xcd, 0x53, 0xc3, 0xdf, 0xd6, 0x28, 0xeb, 0xca, 0x33, 0x4d, 0xa4, 0x63, 0xe7, 0x28,
            0xa2, 0x63, 0xb2, 0xed, 0xc7, 0x15, 0xdb, 0x03, 0x3a, 0x8c, 0x97, 0xe5, 0x5b, 0x76,
            0x9b, 0x07, 0x91, 0xe3, 0x5f, 0xa6, 0x28, 0x1c, 0xf9, 0xf2, 0xc7, 0x30, 0x8d, 0xd3,
            0x16, 0xfc, 0xad, 0x06, 0x73, 0xa1, 0x17, 0x78, 0xe2, 0x2f, 0x03, 0x02, 0xd5, 0x50,
            0xbd, 0x16, 0x21, 0x11, 0xf4, 0x68, 0xe2, 0xa6, 0x35, 0xf0, 0x84, 0x01, 0xaa, 0x17,
            0x3a, 0xef, 0xf7, 0x10, 0xba, 0xf6, 0x23, 0x6b, 0xe4, 0x3f, 0xfa, 0x94, 0xfe, 0x9b,
            0x57, 0xbc, 0x8d, 0x3c, 0xc8, 0x17, 0xcf, 0xf8, 0xef, 0x84, 0x30, 0x70, 0x38, 0xa2,
            0x41, 0xf2, 0x73, 0x8e, 0xf3, 0x72, 0x80, 0x98, 0xf3, 0xdc, 0x5c, 0x7c, 0xa1, 0x51,
            0x27, 0x7e, 0x23, 0x6c, 0x50, 0x9e, 0x13, 0xff, 0x45, 0xf4, 0xdc, 0xac, 0xae, 0x4e,
            0xb8, 0x20, 0xcf, 0xe4, 0x19, 0x9d, 0xc3, 0xf7, 0x66, 0xf6, 0x00, 0xb2, 0xbd, 0xc7,
            0xca, 0x5f, 0x3c, 0x30, 0xf9, 0x7d, 0x52, 0xa9, 0x1d, 0x82, 0xea, 0x14, 0x53, 0xd3,
            0x17, 0xb4, 0x9f, 0xe4, 0x8b, 0xe0, 0x18, 0xc3, 0x4a, 0xa7, 0xd6, 0x73, 0x26, 0x1b,
            0xe3, 0x50, 0x3a, 0x3d, 0xa6, 0x44, 0xd4, 0x8e, 0x12, 0xe6, 0xde, 0x81, 0xa1, 0x02,
            0x34, 0x3a, 0x87, 0x6b, 0x50, 0xc5, 0x85, 0x67, 0x4e, 0x94, 0x58, 0x0a, 0x8f, 0xfc,
            0x6e, 0xab, 0x5b, 0x3d, 0x1e, 0x82, 0x5c, 0x40, 0x91, 0x43, 0x5f, 0xe4, 0xfe, 0x0d,
            0x2b, 0xe9, 0x90, 0x1f, 0xf7, 0xfa, 0x58, 0x99, 0xef, 0x6a, 0x4c, 0x6d, 0xf5, 0xe3,
            0x2e, 0xed, 0x72, 0x59, 0xb0, 0x43, 0x63, 0x43, 0x5f, 0xc5, 0x2d, 0xf1, 0x3e, 0x5a,
            0x41, 0x35, 0x14, 0x42, 0x1c, 0x51, 0x7f, 0x53, 0x94, 0x6b, 0x17, 0xb8, 0xf0, 0xb1,
            0xe5, 0xc0, 0x2d, 0x23, 0xea, 0x33, 0x6c, 0x7f, 0x4a, 0xf7, 0xa9, 0xe1, 0x20, 0x55,
            0x10, 0x99, 0xe8, 0x52, 0xbc, 0xd5, 0xc4, 0x8a, 0x22, 0x82, 0x12, 0x80, 0x0b, 0x90,
            0x3a, 0x44, 0x81, 0x4b, 0x62, 0x65, 0x12, 0x14, 0x19, 0xa4, 0x67, 0x3a, 0x8b, 0xd0,
            0x4e, 0x9a, 0xc2, 0xb2, 0x85, 0xd4, 0x9d, 0x23, 0xa3, 0xfb, 0x41, 0x7a, 0xed, 0x52,
            0xb7, 0xf1, 0xf6, 0x75, 0x35, 0x55, 0x7f, 0xcf, 0x8c, 0x4b, 0x41, 0x69, 0xdf, 0xb5,
            0x2d, 0xa6, 0x9b, 0xf7, 0xa9, 0xc2, 0xce, 0x45, 0x0f, 0x70, 0x5a, 0x30, 0x3c, 0x52,
            0x55, 0xc0, 0xfd, 0x5f, 0x1a, 0x44, 0xf3, 0xbc, 0x75, 0x9c, 0x38, 0x96, 0xd4, 0x63,
            0x4d, 0x9b, 0xef, 0x0d, 0x9a, 0x18, 0x6f, 0xae, 0xbb, 0xae, 0xe5, 0x29, 0xbb, 0x9c,
            0x8a, 0xec, 0xe2, 0xf6, 0x1c, 0x0d, 0xd0, 0x3f, 0xa2, 0x46, 0x91, 0xef, 0x56, 0x30,
            0x00, 0xd0, 0x0a, 0x80, 0xd3, 0x0a, 0x79, 0x3e, 0xcd, 0x5e, 0x0a, 0xbc, 0xb8, 0xef,
            0xb7, 0x51, 0x30, 0xa7, 0xd1, 0xcd, 0xfb, 0x87, 0x05, 0xe6, 0xa2, 0x9b, 0xeb, 0xf6,
            0xb8, 0x80, 0xd6, 0x64, 0x43, 0x49, 0xaa, 0xb2, 0x47, 0x72, 0x4e, 0xec, 0xbb, 0xec,
            0x00, 0x62, 0x14, 0x44, 0x59, 0x12, 0x6b, 0x11, 0x56, 0x13, 0x1f, 0x51, 0x0d, 0xb5,
            0x5a, 0xb8, 0x6f, 0x5b, 0x02, 0xc6, 0x38, 0xb7, 0xad, 0xe3, 0x58, 0xbc, 0xf0, 0x46,
            0x36, 0xda, 0x92, 0xaf, 0xf7, 0xce, 0xc4, 0x4f, 0xc1, 0xb9, 0x0e, 0xdd, 0x94, 0x82,
            0x84, 0x5d, 0xa2, 0xe4, 0x2b, 0xb3, 0x1c, 0xbb, 0x0b, 0x21, 0x1b, 0x4d, 0xbb, 0xf1,
            0xa4, 0x3b, 0x68, 0xb5, 0xe5, 0xca, 0x2d, 0x9d, 0x43, 0x1e, 0xe6, 0xb1, 0xaf, 0xe0,
            0x3f, 0xce, 0x34, 0xaa, 0x27, 0x80, 0xea, 0xb7, 0x6e, 0x59, 0x5f, 0x65, 0xa6, 0x58,
            0xab, 0x9d, 0xac, 0x36, 0x97, 0x1a, 0x7a, 0xec, 0xdb, 0x16, 0x7f, 0x35, 0x0c, 0x55,
            0x91, 0x13, 0xe8, 0xb2, 0xdc, 0xa4, 0x1b, 0x2f, 0xe6, 0x60, 0xb9, 0x07, 0x2f, 0x47,
            0x5e, 0x46, 0x0a, 0x8f, 0x4e, 0x0b, 0x58, 0xb3, 0x52, 0x4e, 0xba, 0x73, 0xfd, 0x13,
            0x14, 0x78, 0xe9, 0xea, 0x67, 0x92, 0x72, 0x89, 0x56, 0xa0, 0xb2, 0x79, 0x98, 0xf0,
            0x92, 0x49, 0x70, 0x52, 0xbd, 0xb6, 0x06, 0x9e, 0x7f, 0x4f, 0xec, 0x6c, 0x12, 0x6b,
            0x7d, 0x02, 0x4a, 0x19, 0xa5, 0x79, 0x49, 0x63, 0x92, 0x90, 0x5e, 0xad, 0x53, 0x21,
            0x56, 0x31, 0x3a, 0x48, 0x1b, 0xa1, 0xfe, 0x95, 0xae, 0xcf, 0xa1, 0x82, 0x3f, 0x15,
            0x40, 0x6e, 0x4f, 0xa9, 0x69, 0x80, 0xb0, 0x48, 0x91, 0x29, 0x14, 0x76, 0xe8, 0x37,
            0x65, 0xce, 0xb6, 0x68, 0x81, 0xdc, 0xbb, 0xc0, 0x4d, 0xef, 0x5c, 0xd3, 0x8e, 0x88,
            0x31, 0xd9, 0x54, 0x6a, 0xc5, 0x83, 0x2b, 0x73, 0xee, 0x1c, 0x95, 0xa6, 0x79, 0x3e,
            0x15, 0xab, 0xb9, 0x4f, 0x59, 0xc8, 0xcf, 0xed, 0xdc, 0xa6, 0x10, 0xb7, 0x8a, 0xd4,
            0x9d, 0xfe, 0x88, 0x5f, 0x7c, 0x1e, 0xd3, 0x4b, 0xbd, 0x3b, 0x91, 0x2c, 0x1d, 0xe8,
            0x68, 0x89, 0xd1, 0x03, 0x29, 0x24, 0x1c, 0x56, 0x2d, 0x1e, 0x60, 0xaf, 0xa3, 0x64,
            0x73, 0xfb, 0x70, 0x24, 0xf3, 0x4c, 0xd0, 0x5f, 0x19, 0x9b, 0xde, 0x5a, 0x93, 0xad,
            0x4b, 0xf7, 0x63, 0x86, 0x38, 0x82, 0x34, 0x15, 0xd0, 0x26, 0x7f, 0xe3, 0xd0, 0xc7,
            0x8e, 0x92, 0xea, 0xb6, 0x3b, 0xb0, 0x0c, 0xce, 0x0a, 0x32, 0x65, 0x6b, 0x33, 0x46,
            0x35, 0x3a, 0x92, 0xee, 0x1d, 0x01, 0x6a, 0x4d, 0x4b, 0x36, 0xb1, 0x19, 0xdf, 0x07,
            0xfe, 0x7b, 0x29, 0x8a, 0xcb, 0x03, 0x34, 0x86, 0x31, 0xe2, 0x81, 0xe2, 0xe2, 0xf9,
            0xc3, 0xe9, 0x65, 0x0c, 0x2b, 0xa9, 0xdd, 0x76, 0xa9, 0x2c, 0x46, 0x82, 0xd9, 0x80,
            0x30, 0x2d, 0x47, 0xbf, 0xeb, 0xdd, 0x74, 0x8f, 0xab, 0xaf, 0xf6, 0xd9, 0x2f, 0xb0,
            0x10, 0x11, 0x84, 0x44, 0xcb, 0x9a, 0xdc, 0x3e, 0x9e, 0x9a, 0x9a, 0x34, 0xe8, 0xcc,
            0xd1, 0xb2, 0x50, 0x46, 0xb5, 0x46, 0x44, 0xc8, 0xc6, 0x37, 0x30, 0xbb, 0xd3, 0xd6,
            0x1a, 0x5d, 0x4d, 0xbd, 0x9e, 0xbb, 0xb4, 0xae, 0x7d, 0xc8, 0x95, 0x33, 0xba, 0x52,
            0xa5, 0xba, 0x43, 0xc6, 0x88, 0x66, 0xc3, 0xc6, 0x46, 0xe5, 0x4a, 0xe0, 0xf6, 0xb6,
            0xca, 0xd5, 0x97, 0x98, 0xd3, 0xf1, 0x15, 0x99, 0x66, 0xb5, 0x02, 0xb2, 0x93, 0x39,
            0x82, 0x95, 0x0c, 0xdf, 0xc2, 0xe4, 0x38, 0x42, 0xc4, 0x4c, 0xf1, 0x3d, 0xcf, 0x1d,
            0xc6, 0x40, 0x25, 0xb7, 0x6d, 0x43, 0x45, 0xf7, 0x28, 0xb1, 0x36, 0x60, 0x5e, 0xaf,
            0xfe, 0xc8, 0x19, 0x5b, 0xc4, 0x18, 0xf8, 0x62, 0x14, 0x2d, 0xc2, 0xd2, 0xb3, 0x49,
            0x7c, 0xe3, 0xd0, 0xab, 0xef, 0x23, 0x07, 0xcb, 0x95, 0xac, 0x35, 0xe4, 0x22, 0xec,
            0xe1, 0x2f, 0xf8, 0x55, 0x67, 0x10, 0xc1, 0x74, 0xe1, 0xf8, 0xac, 0x18, 0x12, 0x55,
            0xc4, 0x53, 0xd0, 0x34, 0xc7, 0x65, 0xa1, 0xb0, 0xbd, 0x64, 0xd0, 0x59, 0xf7, 0xea,
            0x67, 0x01, 0x99, 0xec, 0xab, 0x3a, 0x48, 0xb5, 0xa6, 0x3e, 0x22, 0xab, 0x49, 0xe4,
            0xce, 0x0b, 0x17, 0xac, 0xf8, 0xd1, 0xe9, 0x81, 0x3e, 0x3f, 0x7c, 0xc1, 0xb4, 0x1e,
            0x75, 0x4f, 0x78, 0x7a, 0xa7, 0x25, 0x54, 0x38, 0x11, 0xf2, 0x58, 0x51, 0x6d, 0x56,
            0x11, 0x83, 0x28, 0x70, 0x91, 0x5e, 0x2b, 0x9c, 0x77, 0xb2, 0x2a, 0xec, 0x32, 0xee,
            0x11, 0x99, 0xcd, 0x9f, 0xbc, 0x7a, 0x00, 0x85, 0x70, 0x4e, 0xa6, 0x78, 0x0d, 0xfd,
            0x95, 0xa0, 0xb7, 0x09, 0x3a, 0x57, 0x9c, 0x05, 0x15, 0x56, 0xb2, 0x01, 0x93, 0x32,
            0x76, 0xc6, 0x3c, 0xc3, 0xf0, 0x27, 0x85, 0x1a, 0xa1, 0xca, 0xf1, 0xee, 0xbe, 0x67,
            0x35, 0x9a, 0x9a, 0xa8, 0x5c, 0xbf, 0xef, 0xef, 0xf1, 0x3a, 0xa7, 0x5a, 0xab, 0x42,
            0xf3, 0x84, 0x77, 0x41, 0x96, 0x5d, 0x2e, 0x3f, 0x3f, 0x0f, 0xd3, 0xb0, 0x25, 0x86,
            0x87, 0x51, 0xc9, 0x01, 0x6f, 0x02, 0xaf, 0x94, 0x14, 0x5a, 0x07, 0x5e, 0xb4, 0xa0,
            0x97, 0xcb, 0x79, 0x97, 0xa2, 0xea, 0xf5, 0x93, 0xd7, 0xbd, 0x94, 0xaa, 0xf9, 0x79,
            0x74, 0x2e, 0x0f, 0x2c, 0x9f, 0xe5, 0xa4, 0xef, 0x26, 0x19, 0x99, 0x97, 0x2c, 0xdd,
            0x9e, 0x57, 0x97, 0xdf, 0x35, 0x6a, 0x37, 0x29, 0x66, 0x3b, 0x53, 0x9b, 0x9c, 0xbd,
            0x82, 0x52, 0x11, 0x43, 0x12, 0xee, 0x08, 0x49, 0x92, 0xed, 0xd3, 0x1b, 0x26, 0x18,
            0x91, 0xf3, 0x16, 0x84, 0xe4, 0xe9, 0x00, 0xcf, 0x1a, 0xc9, 0xa1, 0x2b, 0x3f, 0xfb,
            0xeb, 0x68, 0x18, 0x56, 0x9a, 0x59, 0x43, 0x5b, 0xa9, 0x64, 0xe6, 0xe7, 0xc2, 0x6f,
            0x77, 0xc4, 0x0d, 0xc6, 0x29, 0xb9, 0xfe, 0x28, 0xe4, 0x0a, 0x0e, 0x8c, 0x5f, 0xca,
            0xa6, 0xdc, 0xe0, 0x11, 0x2f, 0xac, 0xfa, 0x9b, 0x88, 0x3d, 0x9c, 0x5b, 0x88, 0x98,
            0xdd, 0x50, 0x5b, 0x25, 0xc3, 0x00, 0x03, 0xa4, 0x1e, 0xf2, 0x22, 0x3d, 0x05, 0x1d,
            0xec, 0x5b, 0x28, 0x21, 0xc8, 0x13, 0xa7, 0x8c, 0x6e, 0x5a, 0xe9, 0xae, 0x6f, 0x17,
            0x33, 0x0e, 0x81, 0x82, 0x5f, 0x24, 0xee, 0x33, 0xde, 0x26, 0x40, 0xed, 0x9e, 0x58,
            0xfd, 0x44, 0x57, 0xf7, 0xb4, 0xc4, 0xa6, 0x56, 0x9d, 0x40, 0xc1, 0xf7, 0x7d, 0xbf,
            0x85, 0x4a, 0xa1, 0xcc, 0x10, 0x9b, 0x14, 0x36, 0x90, 0xb5, 0xee, 0x59, 0x70, 0xf3,
            0x49, 0x39, 0x7f, 0xcf, 0x0d, 0x40, 0x7e, 0x21, 0xd0, 0xd4, 0xf2, 0x67, 0x16, 0xc3,
            0xfb, 0x73, 0x28, 0x13, 0x52, 0x00, 0x48, 0x8c, 0xc7, 0x04, 0x95, 0x8b, 0xab, 0x92,
            0x5e, 0xff, 0x74, 0x7d, 0x7b, 0x53, 0x07, 0x74, 0xc8, 0xc2, 0x80, 0x7b, 0x31, 0x08,
            0xcd, 0x47, 0xc5, 0x5d, 0xca, 0x2b, 0x57, 0x4b, 0xab, 0x71, 0xe3, 0xfa, 0x55, 0x54,
            0x62, 0xdc, 0x1e, 0xe0, 0x63, 0x4a, 0x95, 0x4f, 0x0a, 0x15, 0xd6, 0x29, 0xdd, 0x52,
            0xb3, 0x18, 0xb2, 0xf4, 0xee, 0x78, 0xbf, 0xf7, 0xac, 0x67, 0x96, 0xff, 0x61, 0x99,
            0x3b, 0xe7, 0x3f, 0x12, 0xb4, 0x67, 0x8a, 0x79, 0x81, 0x59, 0xa9, 0x42, 0xbf, 0xc5,
            0x01, 0x5a, 0x83, 0x03, 0x1d, 0x7b, 0xf6, 0xbe, 0x6a, 0xb7, 0x87, 0xdf, 0x01, 0x8e,
            0xfd, 0x32,
        ];

        let tx_key: [u8; 32] = [
            52, 143, 85, 11, 74, 246, 144, 9, 220, 173, 186, 173, 20, 4, 160, 35, 217, 9, 162, 34,
            197, 83, 228, 66, 181, 38, 110, 222, 40, 204, 247, 9,
        ];
        let rx_key: [u8; 32] = [
            25, 134, 42, 107, 166, 25, 248, 136, 83, 6, 247, 66, 88, 1, 81, 44, 208, 207, 241, 97,
            234, 20, 206, 231, 153, 19, 68, 159, 128, 92, 93, 244,
        ];

        //let expected_crc = 3624957289;

        let (client, mut server) = tokio::io::duplex(9000);
        let mut client = ProtoStream::new(client);

        let keys: KxSession =
            serde_json::from_str(&serde_json::to_string(&FakeSession { rx_key, tx_key }).unwrap())
                .unwrap();

        let mut session = crate::crypto::Session {
            session: Uuid::new_v4(),
            session_keys: keys,
            state: SessionState {
                kp: dryoc::keypair::KeyPair::gen(),
                remotekey: None,
            },
            stream: client,
        };

        #[cfg(feature = "flutter")]
        let session = SbSession(Arc::new(RwLock::new(session)));

        tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(100)).await;
            server.write(&bytes).await.unwrap();
        });
        let v = session
            .get_messages_recieve_date(
                "newsnet".to_owned(),
                None,
                Some(Utc::now().naive_utc()),
                None,
            )
            .await
            .unwrap();

        // let out: CryptoMessage = client.read_message().await.unwrap();
        // let out = CryptoMessageWrapper::new(out);
        // let v: MessageResponse = out.decrypt(&tx_key).unwrap();
    }

    #[tokio::test]
    async fn test_readwrite_multiple() {
        let ack = Ack {
            success: true,
            status: 1,
            ack_maybe_message: Some(AckMaybeMessage::Text("tests".to_owned())),
        };

        let (client, server) = tokio::io::duplex(64);
        let mut client = ProtoStream::new(client);
        let mut server = ProtoStream::new(server);
        for _ in 0..20 {
            client.write_message(&ack).await.expect("failed to write");
            let newack: Ack = server.read_message().await.expect("failed to read");

            assert_eq!(ack, newack);
        }
    }

    #[tokio::test]
    async fn test_readwrite_sync() {
        let ack = Ack {
            success: true,
            status: 1,
            ack_maybe_message: Some(AckMaybeMessage::Text("tests".to_owned())),
        };

        let (mut c, server) = tokio::io::duplex(64);
        let mut v = Vec::new();
        let mut client = ProtoStream::new(&mut v);
        let mut server = ProtoStream::new(server);
        client.write_message_sync(&ack).expect("failed to write");
        c.write(v.as_slice()).await.unwrap();
        let newack: Ack = server.read_message().await.expect("failed to read");

        assert_eq!(ack, newack);
    }
}