wireframe 0.3.0

Simplify building servers and clients for custom binary protocols.
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

//! Logical-session handle API for fair pooled acquisition.

use std::sync::Arc;

use super::{client_pool::ClientPoolInner, lease::PooledClientLease};
use crate::{
    client::ClientError,
    message::{DecodeWith, EncodeWith},
    serializer::Serializer,
};

/// Stable logical-session identity for fair pooled acquisition.
///
/// A `PoolHandle` lets one caller participate in the pool's fairness policy
/// over time. It is the preferred API when a logical session repeatedly
/// acquires pooled leases and you want the scheduler to account for that
/// identity explicitly.
///
/// # Examples
///
/// ```no_run
/// # use std::net::SocketAddr;
/// use wireframe::client::{ClientPoolConfig, PoolHandle, WireframeClient};
///
/// # async fn demo(addr: SocketAddr) -> Result<(), wireframe::client::ClientError> {
/// let pool = WireframeClient::builder()
///     .connect_pool(addr, ClientPoolConfig::default())
///     .await?;
/// let mut session: PoolHandle<_, _, _> = pool.handle();
/// let lease = session.acquire().await?;
/// drop(lease);
/// # Ok(())
/// # }
/// ```
pub struct PoolHandle<S, P, C>
where
    S: Serializer + Clone + Send + Sync + 'static,
    P: bincode::Encode + Clone + Send + Sync + 'static,
    C: Send + 'static,
{
    inner: Arc<ClientPoolInner<S, P, C>>,
    handle_id: u64,
}

impl<S, P, C> PoolHandle<S, P, C>
where
    S: Serializer + Clone + Send + Sync + 'static,
    P: bincode::Encode + Clone + Send + Sync + 'static,
    C: Send + 'static,
{
    pub(crate) fn new(inner: Arc<ClientPoolInner<S, P, C>>, handle_id: u64) -> Self {
        Self { inner, handle_id }
    }

    /// Acquire one pooled-client lease under the configured fairness policy.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use std::net::SocketAddr;
    /// use wireframe::client::{ClientPoolConfig, PoolHandle, WireframeClient};
    ///
    /// # async fn demo(addr: SocketAddr) -> Result<(), wireframe::client::ClientError> {
    /// let pool = WireframeClient::builder()
    ///     .connect_pool(addr, ClientPoolConfig::default())
    ///     .await?;
    /// let mut session: PoolHandle<_, _, _> = pool.handle();
    /// let lease = session.acquire().await?;
    /// drop(lease);
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`ClientError`] if permit acquisition or pooled checkout fails.
    pub async fn acquire(&mut self) -> Result<PooledClientLease<S, P, C>, ClientError> {
        self.inner
            .scheduler
            .acquire_for_handle(Arc::clone(&self.inner), self.handle_id)
            .await
    }

    /// Perform a complete request-response round trip through a fair lease.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use std::net::SocketAddr;
    /// use wireframe::client::{ClientPoolConfig, PoolHandle, WireframeClient};
    ///
    /// #[derive(bincode::Encode, bincode::Decode)]
    /// struct Ping(u8);
    ///
    /// #[derive(bincode::Encode, bincode::Decode)]
    /// struct Pong(u8);
    ///
    /// # async fn demo(addr: SocketAddr) -> Result<(), wireframe::client::ClientError> {
    /// let pool = WireframeClient::builder()
    ///     .connect_pool(addr, ClientPoolConfig::default())
    ///     .await?;
    /// let mut session: PoolHandle<_, _, _> = pool.handle();
    /// let _reply: Pong = session.call(&Ping(1)).await?;
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`ClientError`] if lease acquisition, serialization, decode, or
    /// transport I/O fails.
    pub async fn call<Req, Resp>(&mut self, request: &Req) -> Result<Resp, ClientError>
    where
        Req: EncodeWith<S>,
        Resp: DecodeWith<S>,
    {
        let lease = self.acquire().await?;
        lease.call(request).await
    }
}

impl<S, P, C> Drop for PoolHandle<S, P, C>
where
    S: Serializer + Clone + Send + Sync + 'static,
    P: bincode::Encode + Clone + Send + Sync + 'static,
    C: Send + 'static,
{
    fn drop(&mut self) { self.inner.scheduler.deregister_handle(self.handle_id); }
}