camber 0.1.6

Opinionated async Rust for IO-bound services on top of Tokio
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151

use super::{block_on, mq_error};
use crate::RuntimeError;
use crate::runtime;
use bytes::Bytes;
use std::time::Duration;

/// A received NATS message.
pub struct Message {
    inner: async_nats::Message,
}

impl Message {
    /// The message payload as bytes.
    pub fn payload(&self) -> &[u8] {
        &self.inner.payload
    }

    /// The subject this message was published to.
    pub fn subject(&self) -> &str {
        self.inner.subject.as_str()
    }
}

/// A NATS subscription that receives messages synchronously.
///
/// Wraps async-nats `Subscriber` with blocking receive operations
/// suitable for use in sync handlers.
pub struct Subscription {
    inner: async_nats::Subscriber,
}

impl Subscription {
    /// Block until the next message arrives or the timeout elapses.
    pub fn next_timeout(&mut self, timeout: Duration) -> Result<Message, RuntimeError> {
        runtime::check_cancel()?;
        let msg = block_on(async {
            tokio::time::timeout(timeout, {
                use futures_util::StreamExt;
                self.inner.next()
            })
            .await
        });
        runtime::check_cancel()?;
        match msg {
            Ok(Some(m)) => Ok(Message { inner: m }),
            Ok(None) => Err(RuntimeError::ChannelClosed),
            Err(_) => Err(RuntimeError::Timeout),
        }
    }

    /// Try to receive a message without blocking. Returns `None` if no message is ready.
    pub fn try_next(&mut self) -> Option<Message> {
        use futures_util::StreamExt;
        block_on(async {
            match tokio::time::timeout(Duration::from_millis(1), self.inner.next()).await {
                Ok(Some(m)) => Some(Message { inner: m }),
                _ => None,
            }
        })
    }
}

/// A sync NATS connection for use in sync handlers.
///
/// All operations use `block_in_place` internally, so they are safe to call
/// from Camber sync handlers without blocking the Tokio runtime.
#[derive(Clone)]
pub struct Connection {
    client: async_nats::Client,
}

/// Connect to a NATS server. Blocks until connected.
///
/// Suitable for sync handlers. For async handlers, use [`connect_async`].
pub fn connect(url: &str) -> Result<Connection, RuntimeError> {
    runtime::check_cancel()?;
    let client = block_on(async_nats::connect(url)).map_err(mq_error)?;
    Ok(Connection { client })
}

/// Connect to a NATS server asynchronously. For use in async handlers.
///
/// Returns the same `Connection` type — usable from both sync and async contexts.
/// The underlying client is inherently async; sync methods use `block_in_place`.
pub async fn connect_async(url: &str) -> Result<Connection, RuntimeError> {
    let client = async_nats::connect(url).await.map_err(mq_error)?;
    Ok(Connection { client })
}

impl Connection {
    /// Publish a message to the given subject.
    pub fn publish(&self, subject: &str, payload: &[u8]) -> Result<(), RuntimeError> {
        runtime::check_cancel()?;
        block_on(
            self.client
                .publish(subject.to_owned(), Bytes::copy_from_slice(payload)),
        )
        .map_err(mq_error)?;
        block_on(self.client.flush()).map_err(mq_error)?;
        runtime::check_cancel()?;
        Ok(())
    }

    /// Subscribe to a subject. Returns a [`Subscription`] that delivers messages synchronously.
    pub fn subscribe(&self, subject: &str) -> Result<Subscription, RuntimeError> {
        runtime::check_cancel()?;
        let sub = block_on(self.client.subscribe(subject.to_owned())).map_err(mq_error)?;
        Ok(Subscription { inner: sub })
    }

    /// Subscribe to a subject with a queue group.
    ///
    /// Messages are distributed across subscribers in the same queue group —
    /// each message is delivered to exactly one member.
    pub fn queue_subscribe(
        &self,
        subject: &str,
        queue_group: &str,
    ) -> Result<Subscription, RuntimeError> {
        runtime::check_cancel()?;
        let sub = block_on(
            self.client
                .queue_subscribe(subject.to_owned(), queue_group.to_owned()),
        )
        .map_err(mq_error)?;
        Ok(Subscription { inner: sub })
    }

    /// Publish a message asynchronously. For use in async handlers.
    pub async fn publish_async(&self, subject: &str, payload: &[u8]) -> Result<(), RuntimeError> {
        self.client
            .publish(subject.to_owned(), Bytes::copy_from_slice(payload))
            .await
            .map_err(mq_error)?;
        self.client.flush().await.map_err(mq_error)?;
        Ok(())
    }

    /// Subscribe asynchronously. For use in async handlers.
    ///
    /// Returns the same [`Subscription`] type — you can still use `next_timeout`
    /// from an async context (it uses `block_in_place` internally).
    pub async fn subscribe_async(&self, subject: &str) -> Result<Subscription, RuntimeError> {
        let sub = self
            .client
            .subscribe(subject.to_owned())
            .await
            .map_err(mq_error)?;
        Ok(Subscription { inner: sub })
    }
}