fluvio_socket/
stream_socket.rs

1use std::sync::Arc;
2
3use fluvio_protocol::api::{Request, RequestMessage};
4use crate::{
5    AsyncResponse, ClientConfig, SharedMultiplexerSocket, SocketError, VersionedSerialSocket,
6    Versions,
7};
8
9const DEFAULT_STREAM_QUEUE_SIZE: usize = 10;
10
11/// Stream Socket
12#[derive(Debug)]
13pub struct StreamSocket {
14    config: Arc<ClientConfig>,
15    socket: SharedMultiplexerSocket,
16    versions: Versions,
17}
18
19impl StreamSocket {
20    pub fn new(
21        config: Arc<ClientConfig>,
22        socket: SharedMultiplexerSocket,
23        versions: Versions,
24    ) -> Self {
25        Self {
26            config,
27            socket,
28            versions,
29        }
30    }
31
32    pub async fn create_serial_socket(&mut self) -> VersionedSerialSocket {
33        VersionedSerialSocket::new(
34            self.socket.clone(),
35            self.config.clone(),
36            self.versions.clone(),
37        )
38    }
39
40    pub fn is_stale(&self) -> bool {
41        self.socket.is_stale()
42    }
43
44    pub async fn create_stream_with_version<R: Request>(
45        &mut self,
46        request: R,
47        version: i16,
48    ) -> Result<AsyncResponse<R>, SocketError> {
49        let mut req_msg = RequestMessage::new_request(request);
50        req_msg.header.set_api_version(version);
51        req_msg
52            .header
53            .set_client_id(self.config.client_id().to_owned());
54        self.socket
55            .create_stream(req_msg, DEFAULT_STREAM_QUEUE_SIZE)
56            .await
57    }
58}
fluvio_socket/stream_socket.rs

fluvio_socket/
stream_socket.rs
