up_rust/communication/
simple_publisher.rs

1/********************************************************************************
2 * Copyright (c) 2024 Contributors to the Eclipse Foundation
3 *
4 * See the NOTICE file(s) distributed with this work for additional
5 * information regarding copyright ownership.
6 *
7 * This program and the accompanying materials are made available under the
8 * terms of the Apache License Version 2.0 which is available at
9 * https://www.apache.org/licenses/LICENSE-2.0
10 *
11 * SPDX-License-Identifier: Apache-2.0
12 ********************************************************************************/
13
14// [impl->dsn~communication-layer-impl-default~1]
15
16use std::sync::Arc;
17
18use async_trait::async_trait;
19
20use crate::{
21    communication::{
22        apply_common_options, build_message, CallOptions, PubSubError, Publisher, UPayload,
23    },
24    LocalUriProvider, UMessageBuilder, UTransport,
25};
26
27/// A [`Publisher`] that uses the uProtocol Transport Layer API for publishing events to topics.
28pub struct SimplePublisher<T, P> {
29    transport: Arc<T>,
30    uri_provider: Arc<P>,
31}
32
33impl<T, P> core::fmt::Debug for SimplePublisher<T, P> {
34    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
35        f.debug_struct("SimplePublisher").finish_non_exhaustive()
36    }
37}
38
39impl<T: UTransport, P: LocalUriProvider> SimplePublisher<T, P> {
40    /// Creates a new client.
41    ///
42    /// # Arguments
43    ///
44    /// * `transport` - The transport to use for sending messages.
45    /// * `uri_provider` - The service to use for creating the event messages' _sink_ address.
46    pub fn new(transport: Arc<T>, uri_provider: Arc<P>) -> Self {
47        SimplePublisher {
48            transport,
49            uri_provider,
50        }
51    }
52}
53
54#[async_trait]
55impl<T: UTransport, P: LocalUriProvider> Publisher for SimplePublisher<T, P> {
56    async fn publish(
57        &self,
58        resource_id: u16,
59        call_options: CallOptions,
60        payload: Option<UPayload>,
61    ) -> Result<(), PubSubError> {
62        let mut builder = UMessageBuilder::publish(self.uri_provider.get_resource_uri(resource_id));
63        apply_common_options(call_options, &mut builder);
64        match build_message(&mut builder, payload) {
65            Ok(publish_message) => self
66                .transport
67                .send(publish_message)
68                .await
69                .map_err(Box::from)
70                .map_err(PubSubError::PublishError),
71            Err(e) => Err(PubSubError::InvalidArgument(format!(
72                "failed to create Publish message from parameters: {e}"
73            ))),
74        }
75    }
76}
77
78#[cfg(test)]
79mod tests {
80
81    // [utest->dsn~communication-layer-impl-default~1]
82
83    use super::*;
84
85    use crate::{utransport::MockTransport, StaticUriProvider, UCode, UPriority, UStatus, UUID};
86
87    fn new_uri_provider() -> Arc<StaticUriProvider> {
88        Arc::new(StaticUriProvider::new("", 0x0005, 0x02).expect("failed to create URI provider"))
89    }
90
91    #[tokio::test]
92    async fn test_publish_fails_for_invalid_topic() {
93        // GIVEN a publisher
94        let uri_provider = new_uri_provider();
95        let mut transport = MockTransport::new();
96
97        transport.expect_do_send().never();
98        let publisher = SimplePublisher::new(Arc::new(transport), uri_provider);
99
100        // WHEN publishing to an invalid topic
101        let options = CallOptions::for_publish(None, None, None);
102        let publish_result = publisher
103            // resource ID for topic must be >= 0x8000
104            .publish(0x1000, options, None)
105            .await;
106
107        // THEN publishing fails with an InvalidArgument error
108        assert!(publish_result.is_err_and(|e| matches!(e, PubSubError::InvalidArgument(_msg))));
109    }
110
111    #[tokio::test]
112    async fn test_publish_fails_with_transport_error() {
113        let message_id = UUID::build();
114        // GIVEN a publisher
115        let uri_provider = new_uri_provider();
116        let mut transport = MockTransport::new();
117        // that is not connected to the underlying messaging infrastructure
118        let expected_message_id = message_id.clone();
119        transport
120            .expect_do_send()
121            .once()
122            .withf(move |msg| msg.id() == &expected_message_id)
123            .returning(|_msg| {
124                Err(UStatus::fail_with_code(
125                    UCode::Unavailable,
126                    "transport not available",
127                ))
128            });
129        let publisher = SimplePublisher::new(Arc::new(transport), uri_provider);
130
131        // WHEN publishing to a valid topic
132        let options = CallOptions::for_publish(None, Some(message_id), None);
133        let publish_result = publisher.publish(0x9A00, options, None).await;
134
135        // THEN publishing fails with a PublishError
136        assert!(publish_result.is_err_and(|e| matches!(e, PubSubError::PublishError(_status))));
137    }
138
139    #[tokio::test]
140    async fn test_publish_succeeds() {
141        // GIVEN a publisher
142        let uri_provider = new_uri_provider();
143        let mut transport = MockTransport::new();
144        let message_id = UUID::build();
145        let expected_message_id = message_id.clone();
146        let value = b"Hello";
147
148        transport
149            .expect_do_send()
150            .once()
151            .withf(move |message| {
152                message.is_publish()
153                    && message.id() == &expected_message_id
154                    && message.priority_unchecked() == UPriority::CS3
155                    && message.ttl_unchecked() == 5_000
156                    && message.payload() == Some(value.as_slice().into())
157            })
158            .returning(|_msg| Ok(()));
159
160        let publisher = SimplePublisher::new(Arc::new(transport), uri_provider);
161
162        // WHEN publishing some data to a valid topic
163        let call_options = CallOptions::for_publish(
164            Some(5_000),
165            Some(message_id.clone()),
166            Some(crate::UPriority::CS3),
167        );
168        let publish_result = publisher
169            .publish(
170                0x9A00,
171                call_options,
172                Some(UPayload::new(value.as_slice(), crate::UPayloadFormat::Raw)),
173            )
174            .await;
175
176        // THEN a corresponding Publish message has been sent via the transport
177        assert!(publish_result.is_ok());
178    }
179}
up_rust/communication/simple_publisher.rs

up_rust/communication/
simple_publisher.rs
