liminal-cli 0.2.1

//! One SDK subscription per tapped subject, read without a timer.

use super::{RailError, TapCloser, TapReader, TapSubscription};
use liminal_sdk::{SdkError, SubscriptionStream};
use serde_json::Value;
use std::sync::Arc;

#[derive(Debug)]
struct Reader {
    stream: SubscriptionStream,
    record_bytes: usize,
}

impl TapReader for Reader {
    fn receive(&self) -> Result<Value, RailError> {
        let record = self
            .stream
            .recv()
            .map_err(|error| RailError::Refused(error.to_string()))?;
        if record.payload().len() > self.record_bytes {
            return Err(RailError::RecordTooLarge {
                limit: self.record_bytes,
            });
        }
        serde_json::from_slice(record.payload()).map_err(|error| RailError::Unreadable {
            message: error.to_string(),
        })
    }
}

struct Closer<F>(F);

impl<F> std::fmt::Debug for Closer<F> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str("SubscriptionClose")
    }
}

impl<F: Fn() -> Result<(), SdkError> + Send + Sync> TapCloser for Closer<F> {
    fn close(&self) -> Result<(), RailError> {
        (self.0)().map_err(|error| RailError::Refused(error.to_string()))
    }
}

pub(super) fn open(
    server: &str,
    channel: &str,
    credential: &[u8],
    record_bytes: usize,
) -> Result<TapSubscription, RailError> {
    let stream = SubscriptionStream::open_with_auth(server, channel, Vec::new(), credential)
        .map_err(|error| RailError::Refused(error.to_string()))?;
    let close = stream
        .close_handle()
        .map_err(|error| RailError::Refused(error.to_string()))?;
    Ok(TapSubscription {
        reader: Box::new(Reader {
            stream,
            record_bytes,
        }),
        closer: Arc::new(Closer(close)),
    })
}