juiz_core 0.0.4

use std::{sync::{Arc, Mutex}, time::Duration, ops::Deref};

use juiz_sdk::anyhow;
use crate::prelude::*;
use crate::brokers::messenger_broker_proxy_factory::create_messenger_broker_proxy_factory;

use super::local_broker::ProxySideSenderReceiverPair;

use crate::brokers::{BrokerProxyFactory, MessengerBrokerProxyCore, MessengerBrokerProxyCoreFactory};

//pub type LocalBrokerProxy = MessengerBrokerProxy;
pub struct LocalBrokerProxyCore {
    sender_receiver: Arc<Mutex<ProxySideSenderReceiverPair>>,
}

pub struct LocalBrokerProxyCoreFactory {
    sender_receiver: Arc<Mutex<ProxySideSenderReceiverPair>>,
}

impl LocalBrokerProxyCoreFactory {
    pub fn new(sender_receiver: Arc<Mutex<ProxySideSenderReceiverPair>>) -> JuizResult<Box<dyn MessengerBrokerProxyCoreFactory>> {
        Ok(Box::new(LocalBrokerProxyCoreFactory {sender_receiver}))
    }
}

impl MessengerBrokerProxyCoreFactory  for LocalBrokerProxyCoreFactory {
    fn create_core(&self, _object_name: &str) -> JuizResult<Box<dyn MessengerBrokerProxyCore>> {
        Ok(Box::new(LocalBrokerProxyCore{sender_receiver: self.sender_receiver.clone()}))
    }
}

impl MessengerBrokerProxyCore for LocalBrokerProxyCore {
    fn send_and_receive(&self, value: CapsuleMap, timeout: Duration) -> JuizResult<CapsulePtr> {
        let us = timeout.as_micros();
        log::trace!("LocaBrokerProxyCore::send_and_receive(timeout_us={us}) called");
        let sndr_recvr = self.sender_receiver.lock().map_err(|_e| return anyhow::Error::from(JuizError::BrokerSendCanNotLockSenderError{}))?;
        let ProxySideSenderReceiverPair(sndr, recvr) = sndr_recvr.deref();
        let _ = sndr.send(value).map_err(|_e| return anyhow::Error::from(JuizError::LocalBrokerProxySendError{}))?;
        let result = recvr.recv_timeout(timeout).map_err(|e| {
                log::error!("LocalBrokerProxyCore::send_and_receive() failed. Error is {e:}");
                return anyhow::Error::from(JuizError::LocalBrokerProxyReceiveTimeoutError{error: e});
        });
        log::trace!("LocaBrokerProxyCore::send_and_receive() exit");
        result
    }

    fn send_and_receive_output(&self, _v: CapsuleMap, _timeout: Duration) -> JuizResult<CapsulePtr> {
        todo!("Outputクラスをどう扱うか。")
    }
}

impl LocalBrokerProxyCore {
    // pub fn new(sender_receiver: Arc<Mutex<ProxySideSenderReceiverPair>> ) -> LocalBrokerProxyCore {
    //     LocalBrokerProxyCore{sender_receiver}
    // }
}

pub fn create_local_broker_proxy_factory(sender_receiver: Arc<Mutex<ProxySideSenderReceiverPair>>) -> JuizResult<Arc<Mutex<dyn BrokerProxyFactory>>> {
    log::trace!("create_local_broker_factory called");
    create_messenger_broker_proxy_factory("LocalBrokerProxyFactory", "local", LocalBrokerProxyCoreFactory::new(sender_receiver)?)
}