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
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
// Copyright 2020 The Fuchsia Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

use bt_avdtp::MediaStream;
use fuchsia_bluetooth::types::PeerId;
use fuchsia_inspect::Node;
use fuchsia_inspect_derive::AttachError;
use futures::{
    future::{BoxFuture, Shared},
    FutureExt,
};
use std::time::Duration;
use thiserror::Error;

use crate::codec::MediaCodecConfig;

#[derive(Debug, Error, Clone)]
#[non_exhaustive]
pub enum MediaTaskError {
    #[error("Operation or configuration not supported")]
    NotSupported,
    #[error("Peer closed the media stream")]
    PeerClosed,
    #[error("Resources needed are already being used")]
    ResourcesInUse,
    #[error("Other Media Task Error: {}", _0)]
    Other(String),
}

impl From<bt_avdtp::Error> for MediaTaskError {
    fn from(error: bt_avdtp::Error) -> Self {
        Self::Other(format!("AVDTP Error: {}", error))
    }
}

/// MediaTaskRunners are configured with information about the media codec when either peer in a
/// conversation configures a stream endpoint.  When successfully configured, they can start
/// MediaTasks by accepting a MediaStream, which will provide or consume media on that stream until
/// dropped or stopped.
///
/// A builder that will make media task runners from requested configurations.
pub trait MediaTaskBuilder: Send + Sync {
    /// Set up to stream based on the given `codec_config` parameters.
    /// Returns a MediaTaskRunner if the configuration is supported, and
    /// MediaTaskError::NotSupported otherwise.
    fn configure(
        &self,
        peer_id: &PeerId,
        codec_config: &MediaCodecConfig,
    ) -> Result<Box<dyn MediaTaskRunner>, MediaTaskError>;
}

/// MediaTaskRunners represent an ability of the media system to start streaming media.
/// They are configured for a specific codec by `MediaTaskBuilder::configure`
/// Typically a MediaTaskRunner can start multiple streams without needing to be reconfigured,
/// although possibly not simultaneously.
pub trait MediaTaskRunner: Send {
    /// Start a MediaTask using the MediaStream given.
    /// If the task started, returns a MediaTask which will finish if the stream ends or an
    /// error occurs, and can be stopped using `MediaTask::stop` or by dropping the MediaTask.
    /// This can fail with MediaTaskError::ResourcesInUse if a MediaTask cannot be started because
    /// one is already running.
    fn start(&mut self, stream: MediaStream) -> Result<Box<dyn MediaTask>, MediaTaskError>;

    /// Try to reconfigure the MediaTask to accept a new configuration.  This differs from
    /// `MediaTaskBuilder::configure` as it attempts to preserve the same configured session.
    /// The runner remains configured with the initial configuration on an error.
    fn reconfigure(&mut self, _config: &MediaCodecConfig) -> Result<(), MediaTaskError> {
        Err(MediaTaskError::NotSupported)
    }

    /// Set the delay reported from the peer for this media task.
    /// This should configure the media source or sink to attempt to compensate.
    /// Typically this is zero for Sink tasks, but Source tasks can receive this info from the peer.
    /// May only be supported before start.
    /// If an Error is returned, the delay has not been set.
    fn set_delay(&mut self, _delay: Duration) -> Result<(), MediaTaskError> {
        Err(MediaTaskError::NotSupported)
    }

    /// Add information from the running media task to the inspect tree
    /// (i.e. data transferred, jitter, etc)
    fn iattach(&mut self, _parent: &Node, _name: &str) -> Result<(), AttachError> {
        Err("attach not implemented".into())
    }
}

/// MediaTasks represent a media stream being actively processed (sent or received from a peer).
/// They are are created by `MediaTaskRunner::start`.
/// Typically a MediaTask will run a background task that is active until dropped or
/// `MediaTask::stop` is called.
pub trait MediaTask: Send {
    /// Returns a Future that finishes when the running media task finshes for any reason.
    /// Should return a future that immediately resolves if this task is finished.
    fn finished(&mut self) -> BoxFuture<'static, Result<(), MediaTaskError>>;

    /// Returns the result if this task has finished, and None otherwise
    fn result(&mut self) -> Option<Result<(), MediaTaskError>> {
        self.finished().now_or_never()
    }

    /// Stops the task normally, signalling to all waiters Ok(()).
    /// Returns the result sent to MediaTask::finished futures, which may be different from Ok(()).
    /// When this function returns, is is good practice to ensure the MediaStream that started
    /// this task is also dropped.
    fn stop(&mut self) -> Result<(), MediaTaskError>;
}

pub mod tests {
    use super::*;

    use futures::{
        channel::{mpsc, oneshot},
        stream::StreamExt,
        Future, TryFutureExt,
    };
    use std::fmt;
    use std::sync::{Arc, Mutex};

    #[derive(Clone)]
    pub struct TestMediaTask {
        /// The PeerId that was used to make this Task
        pub peer_id: PeerId,
        /// The configuration used to make this task
        pub codec_config: MediaCodecConfig,
        /// If still started, this holds the MediaStream.
        pub stream: Arc<Mutex<Option<MediaStream>>>,
        /// Sender for the shared result future. None if already sent.
        sender: Arc<Mutex<Option<oneshot::Sender<Result<(), MediaTaskError>>>>>,
        /// Shared result future.
        result: Shared<BoxFuture<'static, Result<(), MediaTaskError>>>,
        /// Delay the task was started with.
        pub delay: Duration,
    }

    impl fmt::Debug for TestMediaTask {
        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
            f.debug_struct("TestMediaTask")
                .field("peer_id", &self.peer_id)
                .field("codec_config", &self.codec_config)
                .field("result", &self.result.clone().now_or_never())
                .finish()
        }
    }

    impl TestMediaTask {
        pub fn new(
            peer_id: PeerId,
            codec_config: MediaCodecConfig,
            stream: MediaStream,
            delay: Duration,
        ) -> Self {
            let (sender, receiver) = oneshot::channel();
            let result = receiver
                .map_ok_or_else(
                    |_err| Err(MediaTaskError::Other(format!("Nothing sent"))),
                    |result| result,
                )
                .boxed()
                .shared();
            Self {
                peer_id,
                codec_config,
                stream: Arc::new(Mutex::new(Some(stream))),
                sender: Arc::new(Mutex::new(Some(sender))),
                result,
                delay,
            }
        }

        /// Return true if the background media task is running.
        pub fn is_started(&self) -> bool {
            // The stream being held represents the task running.
            self.stream.lock().expect("stream lock").is_some()
        }

        /// End the streaming task without an external stop().
        /// Sends an optional result from the task.
        pub fn end_prematurely(&self, task_result: Option<Result<(), MediaTaskError>>) {
            let _removed_stream = self.stream.lock().expect("mutex").take();
            let mut lock = self.sender.lock().expect("sender lock");
            let sender = lock.take();
            if let (Some(result), Some(sender)) = (task_result, sender) {
                sender.send(result).expect("send ok");
            }
        }
    }

    impl MediaTask for TestMediaTask {
        fn finished(&mut self) -> BoxFuture<'static, Result<(), MediaTaskError>> {
            self.result.clone().boxed()
        }

        fn stop(&mut self) -> Result<(), MediaTaskError> {
            let _ = self.stream.lock().expect("stream lock").take();
            {
                let mut lock = self.sender.lock().expect("sender lock");
                if let Some(sender) = lock.take() {
                    let _ = sender.send(Ok(()));
                    return Ok(());
                }
            }
            // Result should be available.
            self.finished().now_or_never().unwrap()
        }
    }

    pub struct TestMediaTaskRunner {
        /// The peer_id this was started with.
        pub peer_id: PeerId,
        /// The config that this runner will start tasks for
        pub codec_config: MediaCodecConfig,
        /// If this is reconfigurable
        pub reconfigurable: bool,
        /// If this supports delay reporting
        pub supports_set_delay: bool,
        /// What the delay is right now
        pub set_delay: Option<std::time::Duration>,
        /// The Sender that will send a clone of the started tasks to the builder.
        pub sender: mpsc::Sender<TestMediaTask>,
    }

    impl MediaTaskRunner for TestMediaTaskRunner {
        fn start(&mut self, stream: MediaStream) -> Result<Box<dyn MediaTask>, MediaTaskError> {
            let task = TestMediaTask::new(
                self.peer_id.clone(),
                self.codec_config.clone(),
                stream,
                self.set_delay.unwrap_or(Duration::ZERO),
            );
            // Don't particularly care if the receiver got dropped.
            let _ = self.sender.try_send(task.clone());
            Ok(Box::new(task))
        }

        fn set_delay(&mut self, delay: std::time::Duration) -> Result<(), MediaTaskError> {
            if self.supports_set_delay {
                self.set_delay = Some(delay);
                Ok(())
            } else {
                Err(MediaTaskError::NotSupported)
            }
        }

        fn reconfigure(&mut self, config: &MediaCodecConfig) -> Result<(), MediaTaskError> {
            if self.reconfigurable {
                self.codec_config = config.clone();
                Ok(())
            } else {
                Err(MediaTaskError::NotSupported)
            }
        }
    }

    /// A TestMediaTask expects to be configured once, and then started and stopped as appropriate.
    /// It will Error if started again while started or stopped while stopped, or if it was
    /// configured multiple times.
    pub struct TestMediaTaskBuilder {
        sender: Mutex<mpsc::Sender<TestMediaTask>>,
        receiver: mpsc::Receiver<TestMediaTask>,
        reconfigurable: bool,
        supports_set_delay: bool,
    }

    impl TestMediaTaskBuilder {
        pub fn new() -> Self {
            let (sender, receiver) = mpsc::channel(5);
            Self {
                sender: Mutex::new(sender),
                receiver,
                reconfigurable: false,
                supports_set_delay: false,
            }
        }

        pub fn new_reconfigurable() -> Self {
            Self { reconfigurable: true, ..Self::new() }
        }

        pub fn new_delayable() -> Self {
            Self { supports_set_delay: true, ..Self::new() }
        }

        /// Returns a type that implements MediaTaskBuilder.  When a MediaTask is built using
        /// configure(), it will be avialable from `next_task`.
        pub fn builder(&self) -> impl MediaTaskBuilder {
            TestMediaTaskBuilderBuilder(
                self.sender.lock().expect("locking").clone(),
                self.reconfigurable,
                self.supports_set_delay,
            )
        }

        /// Gets a future that will return a handle to the next TestMediaTask that gets started
        /// from a Runner that was retrieved from this builder.
        /// The TestMediaTask, can tell you when it's started and give you a handle to the MediaStream.
        pub fn next_task(&mut self) -> impl Future<Output = Option<TestMediaTask>> + '_ {
            self.receiver.next()
        }

        /// Expects that a task had been built, and retrieves that task, or panics.
        #[track_caller]
        pub fn expect_task(&mut self) -> TestMediaTask {
            self.receiver
                .try_next()
                .expect("should have made a task")
                .expect("shouldn't have dropped all senders")
        }
    }

    struct TestMediaTaskBuilderBuilder(mpsc::Sender<TestMediaTask>, bool, bool);

    impl MediaTaskBuilder for TestMediaTaskBuilderBuilder {
        fn configure(
            &self,
            peer_id: &PeerId,
            codec_config: &MediaCodecConfig,
        ) -> Result<Box<dyn MediaTaskRunner>, MediaTaskError> {
            let runner = TestMediaTaskRunner {
                peer_id: peer_id.clone(),
                codec_config: codec_config.clone(),
                sender: self.0.clone(),
                reconfigurable: self.1,
                supports_set_delay: self.2,
                set_delay: None,
            };
            Ok::<Box<dyn MediaTaskRunner>, _>(Box::new(runner))
        }
    }
}