jsonrpsee_server/
future.rs

1// Copyright 2019-2021 Parity Technologies (UK) Ltd.
2//
3// Permission is hereby granted, free of charge, to any
4// person obtaining a copy of this software and associated
5// documentation files (the "Software"), to deal in the
6// Software without restriction, including without
7// limitation the rights to use, copy, modify, merge,
8// publish, distribute, sublicense, and/or sell copies of
9// the Software, and to permit persons to whom the Software
10// is furnished to do so, subject to the following
11// conditions:
12//
13// The above copyright notice and this permission notice
14// shall be included in all copies or substantial portions
15// of the Software.
16//
17// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF
18// ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
19// TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
20// PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
21// SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
22// CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
23// OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR
24// IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
25// DEALINGS IN THE SOFTWARE.
26
27//! Utilities for handling async code.
28
29use std::sync::Arc;
30
31use tokio::sync::{watch, OwnedSemaphorePermit, Semaphore, TryAcquireError};
32
33/// Create channel to determine whether
34/// the server shall continue to run or not.
35pub fn stop_channel() -> (StopHandle, ServerHandle) {
36    let (tx, rx) = tokio::sync::watch::channel(());
37    (StopHandle::new(rx), ServerHandle::new(tx))
38}
39
40/// Represent a stop handle which is a wrapper over a `multi-consumer receiver`
41/// and cloning [`StopHandle`] will get a separate instance of the underlying receiver.
42#[derive(Debug, Clone)]
43pub struct StopHandle(watch::Receiver<()>);
44
45impl StopHandle {
46    /// Create a new stop handle.
47    pub(crate) fn new(rx: watch::Receiver<()>) -> Self {
48        Self(rx)
49    }
50
51    /// A future that resolves when server has been stopped
52    /// it consumes the stop handle.
53    pub async fn shutdown(mut self) {
54        let _ = self.0.changed().await;
55    }
56}
57
58/// Error when the server has already been stopped.
59#[derive(Debug, Copy, Clone, thiserror::Error)]
60#[error("The server is already stopped")]
61pub struct AlreadyStoppedError;
62
63/// Server handle.
64///
65/// When all [`StopHandle`]'s have been `dropped` or `stop` has been called
66/// the server will be stopped.
67#[derive(Debug, Clone)]
68pub struct ServerHandle(Arc<watch::Sender<()>>);
69
70impl ServerHandle {
71    /// Create a new server handle.
72    pub(crate) fn new(tx: watch::Sender<()>) -> Self {
73        Self(Arc::new(tx))
74    }
75
76    /// Tell the server to stop without waiting for the server to stop.
77    pub fn stop(&self) -> Result<(), AlreadyStoppedError> {
78        self.0.send(()).map_err(|_| AlreadyStoppedError)
79    }
80
81    /// Wait for the server to stop.
82    pub async fn stopped(self) {
83        self.0.closed().await
84    }
85}
86
87/// Limits the number of connections.
88#[derive(Clone, Debug)]
89pub struct ConnectionGuard {
90    inner: Arc<Semaphore>,
91    max: usize,
92}
93
94impl ConnectionGuard {
95    /// Create a new connection guard.
96    pub fn new(limit: usize) -> Self {
97        Self { inner: Arc::new(Semaphore::new(limit)), max: limit }
98    }
99
100    /// Acquire a connection permit.
101    pub fn try_acquire(&self) -> Option<ConnectionPermit> {
102        match self.inner.clone().try_acquire_owned() {
103            Ok(guard) => Some(guard),
104            Err(TryAcquireError::Closed) => {
105                unreachable!("Semaphore::Close is never called and can't be closed; qed")
106            }
107            Err(TryAcquireError::NoPermits) => None,
108        }
109    }
110
111    /// Get the number of available connection slots.
112    pub fn available_connections(&self) -> usize {
113        self.inner.available_permits()
114    }
115
116    /// Get the maximum number of connections.
117    pub fn max_connections(&self) -> usize {
118        self.max
119    }
120}
121
122/// Connection permit.
123pub type ConnectionPermit = OwnedSemaphorePermit;
jsonrpsee_server/future.rs

jsonrpsee_server/
future.rs
