nostralink 0.2.1

//! Query channel

use super::manager::RRCache;
use super::result_set::TRdfResultSet;
use crossbeam_channel::{
    bounded, Receiver as BeamReceiver, Sender as BeamSender,
};
use std::{
    clone::Clone,
    collections::BTreeMap,
    sync::atomic::{AtomicUsize, Ordering},
    sync::{Arc, Mutex},
};

#[derive(Debug)]
pub enum ChannelError {
    TotallyEmpty,
    ReceiveError,
    SendResultsError,
    NoSuchChannelError,
    UnknownError,
    CacheNoResultsError,
}

pub type TX<T> = BeamSender<Arc<TRdfResultSet<T>>>;
pub type RX<T> = BeamReceiver<Arc<TRdfResultSet<T>>>;

const DEF_CHANNEL_SIZE: usize = 8;

#[derive(Debug)]
struct Channel<T: Clone + Send + Sync + 'static> {
    fallback: Option<Arc<TRdfResultSet<T>>>,
    pub tx: TX<T>,
    pub rx: RX<T>,
    pub status: usize,
}

impl<T: Clone + Send + Sync + 'static> Channel<T> {
    pub fn new() -> Self {
        let (tx, rx) = bounded(DEF_CHANNEL_SIZE);

        Self {
            tx,
            rx,
            status: 0,
            fallback: None,
        }
    }
}

#[derive(Debug)]
pub struct QueryChannel<T: Clone + Send + Sync + 'static> {
    /// ID-based channels
    channels_map: Arc<Mutex<BTreeMap<String, Channel<T>>>>,

    pub status: Arc<AtomicUsize>,
    pub rcache: RRCache<T>,
}

impl<T: Clone + Send + Sync + 'static> QueryChannel<T> {
    pub fn new(ttl_secs: Option<u64>, idle_secs: Option<u64>) -> Self {
        let channels_map = Arc::new(Mutex::new(BTreeMap::new()));

        Self {
            rcache: RRCache::new(ttl_secs, idle_secs),
            channels_map,
            status: Arc::new(AtomicUsize::new(0)),
        }
    }

    pub fn new_arc() -> Arc<Self> {
        Arc::new(Self::new(None, None))
    }

    pub fn new_arc_custom(ttl_secs: u64, idle_secs: u64) -> Arc<Self> {
        Arc::new(Self::new(Some(ttl_secs), Some(idle_secs)))
    }

    pub fn get_status(&self) -> usize {
        self.status.load(Ordering::Acquire)
    }

    pub fn set_thread_status(&self, key: &String, val: usize) {
        let mut channels = self.channels_map.lock().unwrap();
        channels
            .entry(key.clone())
            .or_insert(Channel::<T>::new())
            .status = val;
    }

    pub fn get_thread_status(&self, key: &String) -> usize {
        let mut channels = self.channels_map.lock().unwrap();
        channels
            .entry(key.clone())
            .or_insert(Channel::<T>::new())
            .status
    }

    pub fn get_endpoints(&self, key: &String) -> (TX<T>, RX<T>) {
        let mut channels = self.channels_map.lock().unwrap();
        let chan = channels.entry(key.clone()).or_insert(Channel::<T>::new());

        (chan.tx.clone(), chan.rx.clone())
    }

    pub fn reader_for_key(&self, key: &String) -> RX<T> {
        self.get_endpoints(key).1
    }

    pub fn writer_for_key(&self, key: &String) -> TX<T> {
        self.get_endpoints(key).0
    }

    pub fn send(
        &self,
        key: &String,
        results: Arc<TRdfResultSet<T>>,
    ) -> Result<(), ChannelError> {
        match self.get_endpoints(key).0.try_send(results.clone()) {
            Err(_) => Err(ChannelError::SendResultsError),
            Ok(_) => {
                if let Some(chan) =
                    self.channels_map.lock().unwrap().get_mut(key)
                {
                    chan.fallback = Some(results.clone());
                }
                Ok(())
            }
        }
    }

    pub fn rcache_get(
        &self,
        key: &String,
    ) -> Result<Arc<TRdfResultSet<T>>, ChannelError> {
        match self.rcache.cache.get(key) {
            Some(cached) => Ok(cached.clone()),
            None => Err(ChannelError::CacheNoResultsError),
        }
    }

    pub fn recv(
        &self,
        key: &String,
    ) -> Result<Arc<TRdfResultSet<T>>, ChannelError> {
        match self.reader_for_key(key).try_recv() {
            Ok(set) => Ok(set),
            Err(_) => {
                let mut channels = self.channels_map.lock().unwrap();

                if let Some(chan) = channels.get_mut(key) {
                    if let Some(set) = &chan.fallback {
                        return Ok(set.clone());
                    } else {
                        return Err(ChannelError::TotallyEmpty);
                    }
                } else {
                    return Err(ChannelError::NoSuchChannelError);
                }
            }
        }
    }
}

#[macro_export]
macro_rules! qchannel_peek {
    ($channel:ident) => {
        match $channel.reader().try_recv() {
            Ok(set) => Ok(Some(set)),
            Err(_) => Ok(None),
        }
    };
}

#[macro_export]
macro_rules! qchannel_peek_key {
    ($channel:ident, $key:ident) => {
        match $channel.reader_for_key(&$key).try_recv() {
            Ok(set) => Ok(set),
            Err(e) => Err(RdfStoreError::QueryChannelReceiveError),
        }
    };
}