sawtooth_sdk/messaging/
stream.rs

1/*
2 * Copyright 2017 Intel Corporation
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7 *
8 *     http://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 * -----------------------------------------------------------------------------
16 */
17use crate::messages::validator::Message;
18use crate::messages::validator::Message_MessageType;
19use std::sync::mpsc::Receiver;
20use std::sync::mpsc::RecvError;
21use std::time::Duration;
22
23/// A Message Sender
24///
25/// A message
26pub trait MessageSender {
27    fn send(
28        &self,
29        destination: Message_MessageType,
30        correlation_id: &str,
31        contents: &[u8],
32    ) -> Result<MessageFuture, SendError>;
33
34    fn reply(
35        &self,
36        destination: Message_MessageType,
37        correlation_id: &str,
38        contents: &[u8],
39    ) -> Result<(), SendError>;
40
41    fn close(&mut self);
42}
43
44/// Result for a message received.
45pub type MessageResult = Result<Message, ReceiveError>;
46
47/// A message Receiver
48pub type MessageReceiver = Receiver<MessageResult>;
49
50/// A Message Connection
51///
52/// This denotes a connection which can create a MessageSender/Receiver pair.
53pub trait MessageConnection<MS: MessageSender> {
54    fn create(&self) -> (MS, MessageReceiver);
55}
56
57/// Errors that occur on sending a message.
58#[derive(Debug)]
59pub enum SendError {
60    DisconnectedError,
61    TimeoutError,
62    UnknownError,
63}
64
65impl std::error::Error for SendError {}
66
67impl std::fmt::Display for SendError {
68    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
69        match *self {
70            SendError::DisconnectedError => write!(f, "DisconnectedError"),
71            SendError::TimeoutError => write!(f, "TimeoutError"),
72            SendError::UnknownError => write!(f, "UnknownError"),
73        }
74    }
75}
76
77/// Errors that occur on receiving a message.
78#[derive(Debug, Clone)]
79pub enum ReceiveError {
80    TimeoutError,
81    ChannelError(RecvError),
82    DisconnectedError,
83}
84
85impl std::error::Error for ReceiveError {
86    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
87        match self {
88            ReceiveError::ChannelError(err) => Some(&*err),
89            _ => None,
90        }
91    }
92}
93
94impl std::fmt::Display for ReceiveError {
95    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
96        match *self {
97            ReceiveError::TimeoutError => write!(f, "TimeoutError"),
98            ReceiveError::ChannelError(ref err) => write!(f, "ChannelError: {}", err),
99            ReceiveError::DisconnectedError => write!(f, "DisconnectedError"),
100        }
101    }
102}
103/// MessageFuture is a promise for the reply to a sent message on connection.
104pub struct MessageFuture {
105    inner: Receiver<MessageResult>,
106    result: Option<MessageResult>,
107}
108
109impl MessageFuture {
110    pub fn new(inner: Receiver<MessageResult>) -> Self {
111        MessageFuture {
112            inner,
113            result: None,
114        }
115    }
116
117    pub fn get(&mut self) -> MessageResult {
118        if let Some(ref result) = self.result {
119            return result.clone();
120        }
121
122        match self.inner.recv() {
123            Ok(result) => {
124                self.result = Some(result.clone());
125                result
126            }
127            Err(err) => Err(ReceiveError::ChannelError(err)),
128        }
129    }
130
131    pub fn get_timeout(&mut self, timeout: Duration) -> MessageResult {
132        if let Some(ref result) = self.result {
133            return result.clone();
134        }
135
136        match self.inner.recv_timeout(timeout) {
137            Ok(result) => {
138                self.result = Some(result.clone());
139                result
140            }
141            Err(_) => Err(ReceiveError::TimeoutError),
142        }
143    }
144}
145
146/// Queue for inbound messages, sent directly to this stream.
147
148#[cfg(test)]
149mod tests {
150
151    use std::sync::mpsc::channel;
152    use std::thread;
153
154    use crate::messages::validator::Message;
155    use crate::messages::validator::Message_MessageType;
156
157    use super::MessageFuture;
158
159    fn make_ping(correlation_id: &str) -> Message {
160        let mut message = Message::new();
161        message.set_message_type(Message_MessageType::PING_REQUEST);
162        message.set_correlation_id(String::from(correlation_id));
163        message.set_content(String::from("PING").into_bytes());
164
165        message
166    }
167
168    #[test]
169    fn future_get() {
170        let (tx, rx) = channel();
171
172        let mut fut = MessageFuture::new(rx);
173
174        let t = thread::spawn(move || {
175            tx.send(Ok(make_ping("my_test"))).unwrap();
176        });
177
178        let msg = fut.get().expect("Should have a message");
179
180        t.join().unwrap();
181
182        assert_eq!(msg, make_ping("my_test"));
183    }
184}
sawtooth_sdk/messaging/stream.rs

sawtooth_sdk/messaging/
stream.rs
