swctx 0.3.0

One-shot channel with some special semantics.
Documentation 
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

use std::{
  future::Future,
  pin::Pin,
  sync::Arc,
  task::{Context, Poll}
};

use crate::err::Error;

use super::{Shared, State};


/// End-point used to wait for a value to be sent from the paired
/// [`SetCtx`](super::SetCtx).
#[repr(transparent)]
pub struct WaitCtx<T, S, E>(pub(crate) Arc<Shared<T, S, E>>);

impl<T, S, E> WaitCtx<T, S, E> {
  /// Wait for the paired [`SetCtx`](super::SetCtx) to set a value or fail.
  ///
  /// # Errors
  /// Returns application-specific error wrapped in an [`Error::App`] if the
  /// `SetCtx` reported failure.
  pub fn wait(self) -> Result<T, Error<S, E>> {
    let mut inner = self.0.inner.lock();
    loop {
      match inner.state {
        State::Waiting => {
          self.0.signal.wait(&mut inner);
        }
        State::Data(_) => {
          let old = std::mem::replace(&mut inner.state, State::Finalized);
          drop(inner);
          let State::Data(data) = old else {
            unimplemented!("Unable to extract data");
          };
          break Ok(data);
        }
        State::Err(_) => {
          let old = std::mem::replace(&mut inner.state, State::Finalized);
          drop(inner);
          let State::Err(err) = old else {
            unimplemented!("Unable to extract error");
          };
          break Err(err);
        }
        State::Finalized => {
          // Shouldn't be possible
          unimplemented!("Unexpected state")
        }
      }
    }
  }

  /// Non-blocking attempt to get the get the stored value.
  ///
  /// Returns `Ok(Some(T))` if a value has been stored.  Returns `Ok(None)` if
  /// no value has been stored.
  ///
  /// # Errors
  /// Returns application-specific error wrapped in an [`Error::App`] if the
  /// [`SetCtx`](super::SetCtx) reported failure.
  ///
  /// # Panics
  /// This function will panic if called again after it has resolved to either
  /// data or error.
  pub fn try_get(&self) -> Result<Option<T>, Error<S, E>> {
    let mut inner = self.0.inner.lock();
    inner.try_get()
  }

  /// Return a `Future` that will wait for either data to be set or an error to
  /// occur.
  ///
  /// # Cancel safety
  /// The returned `Future` is cancel safe.
  ///
  /// # Panics
  /// This function will panic if called again after it has resolved to either
  /// data or error.
  #[must_use]
  pub const fn wait_async(&self) -> WaitFuture<T, S, E> {
    WaitFuture(self)
  }
}

impl<T, S, E> Drop for WaitCtx<T, S, E> {
  fn drop(&mut self) {
    let mut inner = self.0.inner.lock();
    inner.wctx_dropped = true;
  }
}

impl<T, S, E> Future for WaitCtx<T, S, E> {
  type Output = Result<T, Error<S, E>>;
  fn poll(self: Pin<&mut Self>, ctx: &mut Context<'_>) -> Poll<Self::Output> {
    let mut inner = self.0.inner.lock();
    match inner.try_get() {
      Ok(Some(v)) => Poll::Ready(Ok(v)),
      Ok(None) => {
        inner.waker = Some(ctx.waker().clone());
        Poll::Pending
      }
      Err(e) => Poll::Ready(Err(e))
    }
  }
}


/// Used to wait for the paired [`SetCtx`](super::SetCtx) to set a value in an
/// `async` context.
// A reference to the `WaitCtx` is used (rather than a clone of its
// Arc<Shared>) to tie the lifetime of the `WaitCtx` and the `WaitFuture`
// together, to avoid `WaitFuture` outliving the `WaitCtx` (which wouldn't make
// sense, because dropping `WaitCtx` sets `wctx_dropped` in the shared
// context).
#[repr(transparent)]
pub struct WaitFuture<'wctx, T, S, E>(&'wctx WaitCtx<T, S, E>);

impl<'wctx, T, S, E> Future for WaitFuture<'wctx, T, S, E> {
  type Output = Result<T, Error<S, E>>;
  fn poll(self: Pin<&mut Self>, ctx: &mut Context<'_>) -> Poll<Self::Output> {
    let mut inner = self.0 .0.inner.lock();
    match inner.try_get() {
      Ok(Some(v)) => Poll::Ready(Ok(v)),
      Ok(None) => {
        inner.waker = Some(ctx.waker().clone());
        Poll::Pending
      }
      Err(e) => Poll::Ready(Err(e))
    }
  }
}

// vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 :