ame-bus 1.0.1

Modern microservices framework based on NATS.
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

use serde::de::Error;
use std::sync::Arc;

#[async_trait::async_trait]
/// # NATS Message
///
/// Generic trait for NATS messages.
pub trait NatsMessage
where
    Self: Sized + Send,
{
    /// Error type for serialization.
    type SerError: std::error::Error + Send + Sync + 'static;

    /// Error type for deserialization.
    type DeError: std::error::Error + Send + Sync + 'static;

    /// serialize message to bytes. Can be any format.
    fn to_bytes(&self) -> Result<Arc<[u8]>, Self::SerError>;

    /// parse message from bytes. Can be any format.
    fn parse_from_bytes(bytes: impl AsRef<[u8]>) -> Result<Self, Self::DeError>;
}

impl<T: NatsJsonMessage> NatsMessage for T {
    type SerError = serde_json::Error;
    type DeError = serde_json::Error;

    fn to_bytes(&self) -> Result<Arc<[u8]>, Self::SerError> {
        self.to_json_bytes()
    }

    fn parse_from_bytes(bytes: impl AsRef<[u8]>) -> Result<Self, Self::DeError> {
        Self::from_json_bytes(bytes)
    }
}

/// # NATS JSON Message
///
/// Implement this trait will make the struct can be serialized and deserialized to JSON bytes.
///
/// Based on `serde_json` serialization and deseriali + async_nats::jetstream::consumer::FromConsumerzation.
///
/// implement `NatsJsonMessage` will automatically implement `NatsMessage` for the type.
pub trait NatsJsonMessage
where
    Self: serde::Serialize + for<'de> serde::de::Deserialize<'de> + Sized + Send,
{
    #[allow(missing_docs)]
    fn to_json_bytes(&self) -> Result<Arc<[u8]>, serde_json::Error> {
        let json = serde_json::to_string(self)?;
        Ok(Arc::from(json.into_bytes()))
    }
    #[allow(missing_docs)]
    fn from_json_bytes(bytes: impl AsRef<[u8]>) -> Result<Self, serde_json::Error> {
        let json = std::str::from_utf8(bytes.as_ref())
            .map_err(|_| serde_json::Error::custom("Failed to parse string from bytes"))?;
        let rpc_request = serde_json::from_str(json)?;
        Ok(rpc_request)
    }
}

impl NatsJsonMessage for () {}
impl NatsJsonMessage for serde_json::Value {}

#[async_trait::async_trait]
/// Message in NATS core (including service RPC).
pub trait NatsCoreMessageSendTrait: NatsMessage + DynamicSubjectNatsMessage {


    #[doc(hidden)]
    /// Publish the message to the NATS server.
    ///
    /// DO NOT OVERRIDE THIS FUNCTION.
    async fn publish(&self, nats: &async_nats::Client) -> anyhow::Result<()> {
        let subject = self.subject();
        let bytes = self.to_bytes()?;
        nats.publish(subject, bytes.to_vec().into()).await?;
        Ok(())
    }
}

/// NATS Message that has a subject. 
/// 
/// Can be dynamic or static. Can be NATS core message or JetStream message.
pub trait DynamicSubjectNatsMessage {
    /// The subject of the message. Can be dynamic.
    fn subject(&self) -> String;
}

/// NATS Message that has a subject.
/// 
/// Must be static. Can be NATS core message or JetStream message.
pub trait StaticSubjectNatsMessage {
    /// The subject of the message. Must be static.
    fn subject() -> String;
}

impl<T> DynamicSubjectNatsMessage for T
where
    T: StaticSubjectNatsMessage,
{
    fn subject(&self) -> String {
        T::subject()
    }
}