rofr 0.1.1

Framework for building robotics APIs that use NATS messaging
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

use crate::ServiceContext;
use crate::ServiceState;
use crate::header;
use async_nats::HeaderMap;
use async_nats::ToSubject;
use async_nats::jetstream::context::PublishAckFuture;
use async_nats::jetstream::context::PublishError;
use async_nats::jetstream::message::PublishMessage;
use async_trait::async_trait;
use bytes::Bytes;
use serde::Serialize;
use std::fmt::Debug;
use std::sync::Arc;

/// Stream context.
#[non_exhaustive]
pub struct StreamContext<Context: ServiceContext> {
    pub(crate) subject_prefix: String,
    pub(crate) jetstream: async_nats::jetstream::Context,
    /// Service shared state.
    pub service: Arc<ServiceState<Context>>,
}

impl<Context: ServiceContext> StreamContext<Context> {
    /// Shared context.
    pub fn context(&self) -> &Context {
        &self.service.private
    }

    /// NATS connection.
    pub fn nats(&self) -> async_nats::Client {
        self.jetstream.client()
    }

    /// NATS JetStream connection.
    pub fn jetstream(&self) -> &async_nats::jetstream::Context {
        &self.jetstream
    }

    /// Publish a message to the stream.
    pub async fn send<Subject: ToSubject>(
        &self,
        subject: Subject,
        message: &impl Serialize,
    ) -> Result<PublishAckFuture, PublishError> {
        self.jetstream
            .send_publish(
                format!("{}.{}", self.subject_prefix, subject.to_subject()),
                PublishMessage::build()
                    .payload(Bytes::from(serde_json::to_vec(message).unwrap()))
                    .header(header::MESSAGE_ID, ulid::Ulid::new().to_string()),
            )
            .await
    }

    /// Publish a message with headers to the stream.
    pub async fn send_with_headers<Subject: ToSubject>(
        &self,
        subject: Subject,
        headers: HeaderMap,
        message: &impl Serialize,
    ) -> Result<PublishAckFuture, PublishError> {
        self.jetstream
            .send_publish(
                format!("{}.{}", self.subject_prefix, subject.to_subject()),
                PublishMessage::build()
                    .payload(Bytes::from(serde_json::to_vec(message).unwrap()))
                    .headers(headers)
                    .header(header::MESSAGE_ID, ulid::Ulid::new().to_string()),
            )
            .await
    }
}

#[async_trait]
pub trait StreamHandler<Context>: Debug + Send + Sync
where
    Context: ServiceContext,
{
    async fn handle_stream(
        &self,
        rqctx: StreamContext<Context>,
    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>>;
}