scroll.server/services/gateway/src/session.rs

use std::net::SocketAddr;
use std::sync::Arc;
use titan::crypto::SessionCipher;
use titan::{MessageRegistry, Messaging, MessagingConfig};
use tokio::net::TcpStream;
use crate::backend::Backends;
use crate::config::GatewayConfig;
use crate::message_manager::{MessageManager, RoutingError};
#[derive(Debug, thiserror::Error)]
pub enum SessionError {
    #[error("io: {0}")]
    Io(#[from] std::io::Error),
    #[error("transport: {0}")]
    Transport(#[from] titan::MessagingError),
    #[error("routing: {0}")]
    Routing(#[from] RoutingError),
}
pub struct Session;
impl Session {
    pub async fn serve(
        stream: TcpStream,
        peer: SocketAddr,
        config: Arc<GatewayConfig>,
        backends: Arc<Backends>,
    ) -> Result<(), SessionError> {
        let registry: Arc<MessageRegistry> = Arc::new(logic::scroll_message_registry());
        let messaging_config = MessagingConfig {
            read_timeout: config.read_timeout,
            max_payload_len: config.max_payload_len,
            ..MessagingConfig::default()
        };
        let mut messaging = Messaging::attach(
            stream,
            SessionCipher::scroll_rc4(),
            registry,
            messaging_config,
        )?;
        let mut manager = MessageManager::new(peer, config, backends, messaging.sender());
        let outcome = loop {
            match messaging.next_message().await {
                Ok(None) => break Ok(()),
                Ok(Some(incoming)) => {
                    if let Err(error) = manager.receive_message(incoming).await {
                        break Err(SessionError::from(error));
                    }
                }
                Err(error) => break Err(SessionError::from(error)),
            }
        };
        manager.on_disconnect().await;
        messaging.shutdown().await;
        outcome
    }
}