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
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
//! Transport layer traits
use crate::error::{OutOfMemoryError, ServiceSubscribeError};
use crate::time::{Clock, MicrosecondDuration32};
use crate::transfer::Transfer;
use crate::{ServiceId, SubjectId};
use alloc::vec::Vec;
use core::convert::TryFrom;
use core::fmt::Debug;
use core::hash::Hash;
use num_traits::Bounded;
/// Basic requirements for a transport that can be used to send and receive transfers
///
/// The [`Transmitter`] and [`Receiver`] sub-traits add additional functions.
pub trait Transport {
/// A node ID type that can hold the node ID values that this transport allows
type NodeId: Debug
+ Clone
+ PartialEq
+ Eq
+ Hash
+ Into<u16>
+ TryFrom<u16, Error: Debug>
+ Bounded;
/// A transfer ID type that can hold all supported transfer ID values
type TransferId: TransferId;
/// A priority type that can hold all supported priority values
type Priority: Clone + Debug + From<crate::Priority>;
}
/// A transmitter that can send outgoing transfers
pub trait Transmitter<C>
where
C: Clock,
{
/// The transport that this transmitter works with
type Transport: Transport;
/// The driver type that this transmitter uses to send frames
type Driver;
/// An error type
///
/// This type must have an out-of-memory variant that can hold an `OutOfMemoryError`.
type Error: Debug + From<OutOfMemoryError>;
/// Starts the process of sending an outgoing transfer
///
/// The transport implementation may block until the entire transfer is sent, or put frames in
/// a queue to be sent separately.
fn push<A>(
&mut self,
transfer: Transfer<A, Self::Transport>,
clock: &mut C,
driver: &mut Self::Driver,
) -> nb::Result<(), Self::Error>
where
A: AsRef<[u8]>;
/// Attempts to send all queued outgoing frames
///
/// If ths transport's `push` implementation blocks until all frames have been sent,
/// this function may be empty.
///
/// The transport implementation may block until all frames have been sent, or return
/// `Err(nb::Error::WouldBlock)` if not all frames can be sent.
///
/// Return values:
/// * `Ok(())`: All frames were sent
/// * `Err(nb::Error::WouldBlock)`: At least one frame could not be sent yet
/// * `Err(nb::Error::Other(e))`: Some other error occurred
fn flush(&mut self, clock: &mut C, driver: &mut Self::Driver) -> nb::Result<(), Self::Error>;
/// Returns the maximum transmission unit of this transport, in bytes
///
/// A message larger than this will need to be split into multiple frames.
///
/// For example, Cyphal/CAN over classic CAN can transfer up to 7 bytes per frame (the eighth
/// byte is used up by the tail byte), so it would return 7.
fn mtu(&self) -> usize;
}
/// A receiver that can assemble incoming frames into transfers
pub trait Receiver<C>
where
C: Clock,
{
/// The transport that this transmitter works with
type Transport: Transport;
/// The driver type that this transmitter uses to receive frames
type Driver;
/// An error type
///
/// This type must have an out-of-memory variant that can hold an `OutOfMemoryError`.
type Error: Debug + From<OutOfMemoryError>;
/// Checks for incoming frames and processes them, possibly returning a transfer
///
/// If the frame completes a transfer and the transfer matches an active subscription, the
/// transfer is returned.
///
/// This function must not block. If no frame can immediately be read, it should return `Ok(None)`.
///
/// If the transport reads a frame and processes it, but the frame does not complete a transfer,
/// this function must try again to read and process a frame. It must not return `Ok(None)`
/// if there are incoming frames that remain to be processed.
///
/// This function must not return any transfers for which the transport is not currently
/// subscribed. It also must not return any service transfers not addressed to this node.
///
/// The argument `now` should be the current time. This may be used to assign timestamps to
/// incoming frames and delete sessions that have timed out.
fn receive(
&mut self,
clock: &mut C,
driver: &mut Self::Driver,
) -> Result<Option<Transfer<Vec<u8>, Self::Transport>>, Self::Error>;
/// Subscribes to messages on a subject
///
/// This will enable incoming transfers from all nodes on the specified subject ID.
///
/// subject: The subject ID to subscribe to
///
/// payload_size_max: The maximum number of payload bytes expected on this subject
/// (longer transfers will be truncated)
///
/// timeout: The transfer-ID timeout, for removing duplicate transfers
///
fn subscribe_message(
&mut self,
subject: SubjectId,
payload_size_max: usize,
timeout: MicrosecondDuration32,
driver: &mut Self::Driver,
) -> Result<(), Self::Error>;
/// Unsubscribes from messages on a subject
fn unsubscribe_message(&mut self, subject: SubjectId, driver: &mut Self::Driver);
/// Subscribes to requests for a service
///
/// This will enable incoming service request transfers from all nodes on the specified service
/// ID.
///
/// service: The service ID to subscribe to
///
/// payload_size_max: The maximum number of payload bytes expected on this subject
/// (longer transfers will be truncated)
///
/// timeout: The transfer-ID timeout, for removing duplicate transfers
///
/// This function returns an error if this node is anonymous or some other transport error
/// occurs.
///
fn subscribe_request(
&mut self,
service: ServiceId,
payload_size_max: usize,
timeout: MicrosecondDuration32,
driver: &mut Self::Driver,
) -> Result<(), ServiceSubscribeError<Self::Error>>;
/// Unsubscribes from requests for a service
fn unsubscribe_request(&mut self, service: ServiceId, driver: &mut Self::Driver);
/// Subscribes to responses for a service
///
/// This will enable incoming service response transfers from all nodes on the specified service
/// ID.
///
/// service: The service ID to subscribe to
///
/// payload_size_max: The maximum number of payload bytes expected on this subject
/// (longer transfers will be truncated)
///
/// timeout: The transfer-ID timeout, for removing duplicate transfers
///
/// This function returns an error if this node is anonymous or some other transport error
/// occurs.
///
fn subscribe_response(
&mut self,
service: ServiceId,
payload_size_max: usize,
timeout: MicrosecondDuration32,
driver: &mut Self::Driver,
) -> Result<(), ServiceSubscribeError<Self::Error>>;
/// Unsubscribes from responses for a service
fn unsubscribe_response(&mut self, service: ServiceId, driver: &mut Self::Driver);
/// Updates the identifier of this node
///
/// This can be used after a node ID is identified to make this receiver capable of handling
/// service transfers.
fn set_id(&mut self, id: Option<<Self::Transport as Transport>::NodeId>);
/// Returns an iterator over all subscribers of this Receiver
fn subscribers(&self) -> impl Iterator<Item = SubjectId>;
/// Returns an iterator over all servers of this Receiver
fn servers(&self) -> impl Iterator<Item = ServiceId>;
}
/// Required operations for a transfer ID
pub trait TransferId: Default + Debug + Clone {
/// Increments the value of this transfer ID by 1
///
/// If this transfer ID is the maximum allowed value, this function must wrap around to the
/// minimum allowed value.
fn increment(self) -> Self;
}