see 0.1.2

A high-performance, asynchronous runtime-agnostic alternative to tokio::sync::watch
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
136
137
138

use std::{
    alloc::Layout,
    fmt,
    future::{self, Future},
    mem,
    pin::Pin,
    ptr,
    task::{Context, Poll},
};

use futures_util::future::{BoxFuture, LocalBoxFuture};
use scopeguard::defer;

// If the layouts of T and U are different, return Err(new_value)
fn try_reuse_store<T: ?Sized, U, F>(store: Pin<Box<T>>, new_val: U, cb: F) -> Result<(), U>
where
    F: FnOnce(Box<U>),
{
    let t_layout = Layout::for_value::<T>(&*store);
    let u_layout = Layout::new::<U>();
    if t_layout != u_layout {
        return Err(new_val);
    }
    // Consume the box to convert to a raw pointer, now we manage this pinned memory
    let raw: *mut T = Box::into_raw(unsafe { Pin::into_inner_unchecked(store) });
    // Regardless of whether dropping the old value succeeds, we always update, then
    // call the callback on the new value
    defer!({
        // SAFETY: T and U have the same layout, so `raw` is a valid pointer.
        let raw: *mut U = raw.cast::<U>();
        unsafe { raw.write(new_val) };
        let store = unsafe { Box::from_raw(raw) };
        cb(store)
    });
    unsafe { ptr::drop_in_place(raw) };
    Ok(())
}

pub mod sync {
    use super::*;
    pub struct BoxedFutureStore<'a, T> {
        store: BoxFuture<'a, T>,
    }

    impl<'a, T: 'a> BoxedFutureStore<'a, T> {
        pub fn new<F>(fut: F) -> Self
        where
            F: Future<Output = T> + Send + 'a,
        {
            Self {
                store: Box::pin(fut),
            }
        }

        pub fn set<F>(&mut self, fut: F)
        where
            F: Future<Output = T> + Send + 'a,
        {
            // Temporarily swap out the current boxed future so we can take ownership of it.
            let store = mem::replace(&mut self.store, Box::pin(future::pending()));
            // Try to reuse the box
            if let Err(fut) = try_reuse_store(store, fut, |b| self.store = Pin::from(b)) {
                // If reuse fails due to different memory layouts, fall back to reallocation.
                *self = Self::new(fut);
            }
        }

        pub fn poll(&mut self, cx: &mut Context<'_>) -> Poll<T> {
            self.store.as_mut().poll(cx)
        }
    }

    impl<T> Future for BoxedFutureStore<'_, T> {
        type Output = T;

        fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
            Pin::into_inner(self).store.as_mut().poll(cx)
        }
    }

    unsafe impl<T> Sync for BoxedFutureStore<'_, T> {}

    impl<T> fmt::Debug for BoxedFutureStore<'_, T> {
        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
            f.debug_struct("BoxedFutureStore").finish()
        }
    }
}

pub mod unsync {
    use super::*;

    pub struct LocalBoxedFutureStore<'a, T> {
        store: LocalBoxFuture<'a, T>,
    }

    impl<'a, T: 'a> LocalBoxedFutureStore<'a, T> {
        pub fn new<F>(fut: F) -> Self
        where
            F: Future<Output = T> + 'a,
        {
            Self {
                store: Box::pin(fut),
            }
        }

        pub fn set<F>(&mut self, fut: F)
        where
            F: Future<Output = T> + 'a,
        {
            // Temporarily swap out the current boxed future so we can take ownership of it.
            let store = mem::replace(&mut self.store, Box::pin(future::pending()));
            // Try to reuse the box
            if let Err(fut) = try_reuse_store(store, fut, |b| self.store = Pin::from(b)) {
                // If reuse fails due to different memory layouts, fall back to reallocation.
                *self = Self::new(fut);
            }
        }

        pub fn poll(&mut self, cx: &mut Context<'_>) -> Poll<T> {
            self.store.as_mut().poll(cx)
        }
    }

    impl<T> Future for LocalBoxedFutureStore<'_, T> {
        type Output = T;

        fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
            Pin::into_inner(self).store.as_mut().poll(cx)
        }
    }

    impl<T> fmt::Debug for LocalBoxedFutureStore<'_, T> {
        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
            f.debug_struct("LocalBoxedFutureStore").finish()
        }
    }
}