livekit_protocol/
promise.rs

1// Copyright 2024 LiveKit, Inc.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use tokio::sync::{oneshot, Mutex, RwLock};
16
17pub struct Promise<T> {
18    tx: Mutex<Option<oneshot::Sender<T>>>,
19    rx: Mutex<Option<oneshot::Receiver<T>>>,
20    result: RwLock<Option<T>>,
21}
22
23impl<T: Clone> Promise<T> {
24    pub fn new() -> Self {
25        let (tx, rx) = oneshot::channel();
26        Self { tx: Mutex::new(Some(tx)), rx: Mutex::new(Some(rx)), result: Default::default() }
27    }
28
29    pub fn resolve(&self, result: T) -> Result<(), &'static str> {
30        let mut tx = self.tx.try_lock().unwrap();
31        if tx.is_some() {
32            let _ = tx.take().unwrap().send(result);
33            Ok(())
34        } else {
35            Err("promise already used")
36        }
37    }
38
39    pub async fn result(&self) -> T {
40        {
41            let result_read = self.result.read().await;
42            if let Some(result) = result_read.clone() {
43                return result;
44            }
45        }
46
47        let mut rx = self.rx.lock().await;
48        if let Some(rx) = rx.take() {
49            let result = rx.await.unwrap();
50            *self.result.write().await = Some(result.clone());
51            result
52        } else {
53            self.result.read().await.clone().unwrap()
54        }
55    }
56
57    pub fn try_result(&self) -> Option<T> {
58        self.result.try_read().ok().and_then(|result| result.clone())
59    }
60}
livekit_protocol/promise.rs

livekit_protocol/
promise.rs
