midi_io/
connection.rs

1use std::pin::Pin;
2use std::sync::Arc;
3use std::task::Context;
4use std::task::Poll;
5
6use futures_channel::mpsc;
7use futures_util::stream::select;
8use futures_util::Stream;
9use futures_util::StreamExt;
10
11use crate::platform::PlatformClient;
12use crate::port::VirtualPortId;
13use crate::Decoded;
14use crate::Destination;
15use crate::DestinationChange;
16use crate::Error;
17use crate::MidiMessage;
18use crate::PortId;
19use crate::RawMidiMessage;
20use crate::Source;
21use crate::SourceChange;
22use crate::SysEx;
23
24#[derive(Debug, Clone)]
25pub struct Timed<T> {
26    pub timestamp: std::time::Instant,
27    pub payload: T,
28}
29
30impl<T> Timed<T> {
31    pub fn map<U, F: FnOnce(T) -> U>(self, f: F) -> Timed<U> {
32        Timed {
33            timestamp: self.timestamp,
34            payload: f(self.payload),
35        }
36    }
37}
38
39#[derive(Debug)]
40enum ConnectionId {
41    Source(PortId),
42    Destination(PortId),
43    VirtualSource(VirtualPortId),
44    VirtualDestination(VirtualPortId),
45}
46
47#[derive(Debug)]
48pub(crate) struct ConnectionGuard {
49    target: ConnectionId,
50    platform: Arc<PlatformClient>,
51}
52
53impl Drop for ConnectionGuard {
54    fn drop(&mut self) {
55        match &self.target {
56            ConnectionId::Source(port) => self.platform.disconnect(*port),
57            ConnectionId::Destination(port) => self.platform.disconnect_destination(*port),
58            ConnectionId::VirtualSource(id) => self.platform.destroy_virtual_source(*id),
59            ConnectionId::VirtualDestination(id) => self.platform.destroy_virtual_destination(*id),
60        }
61    }
62}
63
64#[derive(Debug)]
65struct ReceiveStreams {
66    message_rx: mpsc::Receiver<Timed<MidiMessage>>,
67    sysex_rx: mpsc::Receiver<Timed<SysEx>>,
68    error_rx: mpsc::Receiver<Timed<Error>>,
69    guard: Arc<ConnectionGuard>,
70}
71
72impl ReceiveStreams {
73    fn into_streams(self) -> Streams {
74        Streams {
75            messages: MessageStream {
76                inner: self.message_rx,
77                _guard: self.guard.clone(),
78            },
79            sysex: SysexStream {
80                inner: self.sysex_rx,
81                _guard: self.guard.clone(),
82            },
83            errors: ErrorStream {
84                inner: self.error_rx,
85                _guard: self.guard,
86            },
87        }
88    }
89
90    fn into_messages(self) -> MessageStream {
91        MessageStream {
92            inner: self.message_rx,
93            _guard: self.guard,
94        }
95    }
96
97    fn into_sysex(self) -> SysexStream {
98        SysexStream {
99            inner: self.sysex_rx,
100            _guard: self.guard,
101        }
102    }
103
104    fn into_errors(self) -> ErrorStream {
105        ErrorStream {
106            inner: self.error_rx,
107            _guard: self.guard,
108        }
109    }
110
111    fn into_events(self) -> EventStream {
112        EventStream::new(self.message_rx, self.sysex_rx, self.error_rx, self.guard)
113    }
114}
115
116#[derive(Debug)]
117pub struct SourceConnection {
118    streams: ReceiveStreams,
119}
120
121impl SourceConnection {
122    pub(crate) fn new(
123        message_rx: mpsc::Receiver<Timed<MidiMessage>>,
124        sysex_rx: mpsc::Receiver<Timed<SysEx>>,
125        error_rx: mpsc::Receiver<Timed<Error>>,
126        port: PortId,
127        platform: Arc<crate::platform::PlatformClient>,
128    ) -> Self {
129        Self {
130            streams: ReceiveStreams {
131                message_rx,
132                sysex_rx,
133                error_rx,
134                guard: Arc::new(ConnectionGuard {
135                    target: ConnectionId::Source(port),
136                    platform,
137                }),
138            },
139        }
140    }
141
142    pub fn into_streams(self) -> Streams {
143        self.streams.into_streams()
144    }
145
146    pub fn into_messages(self) -> MessageStream {
147        self.streams.into_messages()
148    }
149
150    pub fn into_sysex(self) -> SysexStream {
151        self.streams.into_sysex()
152    }
153
154    pub fn into_errors(self) -> ErrorStream {
155        self.streams.into_errors()
156    }
157
158    pub fn into_events(self) -> EventStream {
159        self.streams.into_events()
160    }
161}
162
163#[derive(Debug)]
164pub struct MessageStream {
165    inner: mpsc::Receiver<Timed<MidiMessage>>,
166    _guard: Arc<ConnectionGuard>,
167}
168
169impl MessageStream {
170    pub async fn recv(&mut self) -> Option<Timed<MidiMessage>> {
171        self.next().await
172    }
173}
174
175impl Stream for MessageStream {
176    type Item = Timed<MidiMessage>;
177
178    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
179        Pin::new(&mut self.inner).poll_next(cx)
180    }
181}
182
183#[derive(Debug)]
184pub struct SysexStream {
185    inner: mpsc::Receiver<Timed<SysEx>>,
186    _guard: Arc<ConnectionGuard>,
187}
188
189impl SysexStream {
190    pub async fn recv(&mut self) -> Option<Timed<SysEx>> {
191        self.next().await
192    }
193}
194
195impl Stream for SysexStream {
196    type Item = Timed<SysEx>;
197
198    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
199        Pin::new(&mut self.inner).poll_next(cx)
200    }
201}
202
203#[derive(Debug)]
204pub struct ErrorStream {
205    inner: mpsc::Receiver<Timed<Error>>,
206    _guard: Arc<ConnectionGuard>,
207}
208
209impl ErrorStream {
210    pub async fn recv(&mut self) -> Option<Timed<Error>> {
211        self.next().await
212    }
213}
214
215impl Stream for ErrorStream {
216    type Item = Timed<Error>;
217
218    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
219        Pin::new(&mut self.inner).poll_next(cx)
220    }
221}
222
223#[derive(Debug)]
224pub struct Streams {
225    pub messages: MessageStream,
226    pub sysex: SysexStream,
227    pub errors: ErrorStream,
228}
229
230#[allow(clippy::type_complexity)]
231pub struct EventStream {
232    inner: Pin<Box<dyn Stream<Item = Timed<Result<Decoded, Error>>> + Send>>,
233    _guard: Arc<ConnectionGuard>,
234}
235
236impl EventStream {
237    fn new(
238        message_rx: mpsc::Receiver<Timed<MidiMessage>>,
239        sysex_rx: mpsc::Receiver<Timed<SysEx>>,
240        error_rx: mpsc::Receiver<Timed<Error>>,
241        guard: Arc<ConnectionGuard>,
242    ) -> Self {
243        let messages = message_rx.map(|t| t.map(|m| Ok::<Decoded, Error>(Decoded::Message(m))));
244        let sysex = sysex_rx.map(|t| t.map(|s| Ok::<Decoded, Error>(Decoded::SysEx(s))));
245        let errors = error_rx.map(|t| t.map(Err::<Decoded, Error>));
246        Self {
247            inner: Box::pin(select(select(messages, sysex), errors)),
248            _guard: guard,
249        }
250    }
251
252    pub async fn recv(&mut self) -> Option<Timed<Result<Decoded, Error>>> {
253        self.next().await
254    }
255}
256
257impl Stream for EventStream {
258    type Item = Timed<Result<Decoded, Error>>;
259
260    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
261        self.inner.as_mut().poll_next(cx)
262    }
263}
264
265impl std::fmt::Debug for EventStream {
266    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
267        f.debug_struct("EventStream").finish_non_exhaustive()
268    }
269}
270
271#[derive(Debug, Clone)]
272pub struct DestinationConnection {
273    pub(crate) port: PortId,
274    pub(crate) guard: Arc<ConnectionGuard>,
275}
276
277impl DestinationConnection {
278    pub(crate) fn new(port: PortId, platform: Arc<crate::platform::PlatformClient>) -> Self {
279        Self {
280            port,
281            guard: Arc::new(ConnectionGuard {
282                target: ConnectionId::Destination(port),
283                platform,
284            }),
285        }
286    }
287
288    pub async fn send(&self, msg: &MidiMessage) -> Result<(), Error> {
289        let raw = RawMidiMessage::from(msg);
290        self.guard.platform.send_midi(self.port, raw).await
291    }
292
293    pub async fn send_sysex(&self, sysex: &SysEx) -> Result<(), Error> {
294        self.guard
295            .platform
296            .send_sysex(self.port, sysex.to_wire_bytes())
297            .await
298    }
299}
300
301#[derive(Debug)]
302pub struct VirtualDestination {
303    streams: ReceiveStreams,
304    port: PortId,
305    name: String,
306}
307
308impl VirtualDestination {
309    pub(crate) fn new(
310        message_rx: mpsc::Receiver<Timed<MidiMessage>>,
311        sysex_rx: mpsc::Receiver<Timed<SysEx>>,
312        error_rx: mpsc::Receiver<Timed<Error>>,
313        id: VirtualPortId,
314        port: PortId,
315        name: String,
316        platform: Arc<crate::platform::PlatformClient>,
317    ) -> Self {
318        Self {
319            streams: ReceiveStreams {
320                message_rx,
321                sysex_rx,
322                error_rx,
323                guard: Arc::new(ConnectionGuard {
324                    target: ConnectionId::VirtualDestination(id),
325                    platform,
326                }),
327            },
328            port,
329            name,
330        }
331    }
332
333    pub fn as_destination(&self) -> Destination {
334        Destination {
335            id: self.port,
336            name: self.name.clone(),
337            is_virtual: true,
338        }
339    }
340
341    pub fn into_streams(self) -> Streams {
342        self.streams.into_streams()
343    }
344
345    pub fn into_messages(self) -> MessageStream {
346        self.streams.into_messages()
347    }
348
349    pub fn into_sysex(self) -> SysexStream {
350        self.streams.into_sysex()
351    }
352
353    pub fn into_errors(self) -> ErrorStream {
354        self.streams.into_errors()
355    }
356
357    pub fn into_events(self) -> EventStream {
358        self.streams.into_events()
359    }
360}
361
362#[derive(Debug, Clone)]
363pub struct VirtualSource {
364    pub(crate) id: VirtualPortId,
365    pub(crate) port: PortId,
366    pub(crate) name: String,
367    pub(crate) guard: Arc<ConnectionGuard>,
368}
369
370impl VirtualSource {
371    pub(crate) fn new(
372        id: VirtualPortId,
373        port: PortId,
374        name: String,
375        platform: Arc<crate::platform::PlatformClient>,
376    ) -> Self {
377        Self {
378            id,
379            port,
380            name,
381            guard: Arc::new(ConnectionGuard {
382                target: ConnectionId::VirtualSource(id),
383                platform,
384            }),
385        }
386    }
387
388    pub fn as_source(&self) -> Source {
389        Source {
390            id: self.port,
391            name: self.name.clone(),
392            is_virtual: true,
393        }
394    }
395
396    pub async fn send(&self, msg: &MidiMessage) -> Result<(), Error> {
397        let raw = RawMidiMessage::from(msg);
398        self.guard.platform.send_virtual_midi(self.id, raw).await
399    }
400
401    pub async fn send_sysex(&self, sysex: &SysEx) -> Result<(), Error> {
402        self.guard
403            .platform
404            .send_virtual_sysex(self.id, sysex.to_wire_bytes())
405            .await
406    }
407}
408
409#[derive(Debug)]
410pub struct SourceChanges(pub(crate) mpsc::UnboundedReceiver<SourceChange>);
411
412impl SourceChanges {
413    pub async fn recv(&mut self) -> Option<SourceChange> {
414        self.next().await
415    }
416}
417
418impl Stream for SourceChanges {
419    type Item = SourceChange;
420
421    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
422        Pin::new(&mut self.0).poll_next(cx)
423    }
424}
425
426#[derive(Debug)]
427pub struct DestinationChanges(pub(crate) mpsc::UnboundedReceiver<DestinationChange>);
428
429impl DestinationChanges {
430    pub async fn recv(&mut self) -> Option<DestinationChange> {
431        self.next().await
432    }
433}
434
435impl Stream for DestinationChanges {
436    type Item = DestinationChange;
437
438    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
439        Pin::new(&mut self.0).poll_next(cx)
440    }
441}
midi_io/connection.rs

midi_io/
connection.rs
