nmode's Git Repositories - signal-cli/blob - client/src/tcp.rs

   1 use jsonrpc_client_transports::{transports::duplex, RpcChannel, RpcError};
   2 use jsonrpc_core::futures_util::{SinkExt, StreamExt, TryStreamExt};
   3 use jsonrpc_server_utils::{codecs::StreamCodec, tokio_util::codec::Decoder};
   4 use tokio::net::{TcpStream, ToSocketAddrs};
   5
   6 /// Connect to a JSON-RPC TCP server.
   7 pub async fn connect<S: ToSocketAddrs, Client: From<RpcChannel>>(
   8     socket: S,
   9 ) -> Result<Client, RpcError> {
  10     let connection = TcpStream::connect(socket)
  11         .await
  12         .map_err(|e| RpcError::Other(Box::new(e)))?;
  13     let (sink, stream) = StreamCodec::stream_incoming().framed(connection).split();
  14     let sink = sink.sink_map_err(|e| RpcError::Other(Box::new(e)));
  15     let stream = stream.map_err(|e| log::error!("TCP stream error: {}", e));
  16
  17     let (client, sender) = duplex(
  18         Box::pin(sink),
  19         Box::pin(
  20             stream
  21                 .take_while(|x| std::future::ready(x.is_ok()))
  22                 .map(|x| x.expect("Stream is closed upon first error.")),
  23         ),
  24     );
  25
  26     tokio::spawn(client);
  27
  28     Ok(sender.into())
  29 }