Switch to jsonrpsee

Fixes #1275

client/Cargo.toml

@@ -8,18 +8,16 @@ edition = "2021"
 [dependencies]
 anyhow = "1"
 clap = { version = "4", features = ["cargo", "derive", "wrap_help"] }
-jsonrpc-core = "18"
-jsonrpc-core-client = "18"
-jsonrpc-client-transports = { version = "18", default-features = false, features = [
-    "ipc",
-] }
-jsonrpc-derive = "18"
-jsonrpc-server-utils = "18"
 log = "0.4"
 serde = "1"
 serde_json = "1"
-tokio = { version = "1", features = ["rt", "macros", "net"] }
+tokio = { version = "1", features = ["rt", "macros", "net", "rt-multi-thread"] }
-
+jsonrpsee = { version = "0.19.0", features = [
-[patch.crates-io]
+    "macros",
-jsonrpc-client-transports = { git = "https://github.com/AsamK/jsonrpc", branch = "client_subscribe_named_params" }
+    "async-client",
-jsonrpc-derive = { git = "https://github.com/AsamK/jsonrpc", branch = "client_subscribe_named_params" }
+    "http-client",
+] }
+bytes = "1"
+tokio-util = "0.7"
+futures-util = "0.3"
+thiserror = "1"

client/src/cli.rs

@@ -10,13 +10,17 @@ pub struct Cli {
     pub account: Option<String>,
 
     /// TCP host and port of signal-cli daemon
-    #[arg(long)]
+    #[arg(long, conflicts_with = "json_rpc_http")]
     pub json_rpc_tcp: Option<Option<SocketAddr>>,
 
     /// UNIX socket address and port of signal-cli daemon
-    #[arg(long)]
+    #[arg(long, conflicts_with = "json_rpc_tcp")]
     pub json_rpc_socket: Option<Option<OsString>>,
 
+    /// HTTP URL of signal-cli daemon
+    #[arg(long, conflicts_with = "json_rpc_socket")]
+    pub json_rpc_http: Option<Option<String>>,
+
     #[arg(value_enum, long, default_value_t = OutputTypes::Json)]
     pub output: OutputTypes,
 

client/src/jsonrpc.rs

@@ -1,49 +1,59 @@
 use std::path::Path;
 
-use jsonrpc_client_transports::{transports::ipc, RpcError};
+use jsonrpsee::async_client::ClientBuilder;
-use jsonrpc_core::serde::Deserialize;
+use jsonrpsee::core::client::SubscriptionClientT;
-use jsonrpc_derive::rpc;
+use jsonrpsee::core::Error;
+use jsonrpsee::http_client::HttpClientBuilder;
+use jsonrpsee::proc_macros::rpc;
+use serde::Deserialize;
+use serde_json::Value;
 use tokio::net::ToSocketAddrs;
 
-pub type SignalCliClient = gen_client::Client;
+#[rpc(client)]
-
-#[rpc(client, params = "named")]
 pub trait Rpc {
-    #[rpc(name = "addDevice", params = "named")]
+    #[method(name = "addDevice", param_kind = map)]
-    fn add_device(&self, account: Option<String>, uri: String) -> Result<Value>;
+    async fn add_device(
+        &self,
+        account: Option<String>,
+        uri: String,
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "block", params = "named")]
+    #[method(name = "block", param_kind = map)]
     fn block(
         &self,
         account: Option<String>,
         recipients: Vec<String>,
         #[allow(non_snake_case)] groupIds: Vec<String>,
-    ) -> Result<Value>;
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "deleteLocalAccountData", params = "named")]
+    #[method(name = "deleteLocalAccountData", param_kind = map)]
     fn delete_local_account_data(
         &self,
         account: Option<String>,
         #[allow(non_snake_case)] ignoreRegistered: Option<bool>,
-    ) -> Result<Value>;
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "getUserStatus", params = "named")]
+    #[method(name = "getUserStatus", param_kind = map)]
-    fn get_user_status(&self, account: Option<String>, recipients: Vec<String>) -> Result<Value>;
+    fn get_user_status(
+        &self,
+        account: Option<String>,
+        recipients: Vec<String>,
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "joinGroup", params = "named")]
+    #[method(name = "joinGroup", param_kind = map)]
-    fn join_group(&self, account: Option<String>, uri: String) -> Result<Value>;
+    fn join_group(&self, account: Option<String>, uri: String) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "finishLink", params = "named")]
+    #[method(name = "finishLink", param_kind = map)]
     fn finish_link(
         &self,
         #[allow(non_snake_case)] deviceLinkUri: String,
         #[allow(non_snake_case)] deviceName: String,
-    ) -> Result<Value>;
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "listAccounts", params = "named")]
+    #[method(name = "listAccounts", param_kind = map)]
-    fn list_accounts(&self) -> Result<Value>;
+    fn list_accounts(&self) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "listContacts", params = "named")]
+    #[method(name = "listContacts", param_kind = map)]
     fn list_contacts(
         &self,
         account: Option<String>,
@@ -51,60 +61,64 @@ pub trait Rpc {
         #[allow(non_snake_case)] allRecipients: bool,
         blocked: Option<bool>,
         name: Option<String>,
-    ) -> Result<Value>;
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "listDevices", params = "named")]
+    #[method(name = "listDevices", param_kind = map)]
-    fn list_devices(&self, account: Option<String>) -> Result<Value>;
+    fn list_devices(&self, account: Option<String>) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "listGroups", params = "named")]
+    #[method(name = "listGroups", param_kind = map)]
     fn list_groups(
         &self,
         account: Option<String>,
         #[allow(non_snake_case)] groupIds: Vec<String>,
-    ) -> Result<Value>;
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "listIdentities", params = "named")]
+    #[method(name = "listIdentities", param_kind = map)]
-    fn list_identities(&self, account: Option<String>, number: Option<String>) -> Result<Value>;
+    fn list_identities(
+        &self,
+        account: Option<String>,
+        number: Option<String>,
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "listStickerPacks", params = "named")]
+    #[method(name = "listStickerPacks", param_kind = map)]
-    fn list_sticker_packs(&self, account: Option<String>) -> Result<Value>;
+    fn list_sticker_packs(&self, account: Option<String>) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "quitGroup", params = "named")]
+    #[method(name = "quitGroup", param_kind = map)]
     fn quit_group(
         &self,
         account: Option<String>,
         #[allow(non_snake_case)] groupId: String,
         delete: bool,
         admins: Vec<String>,
-    ) -> Result<Value>;
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "register", params = "named")]
+    #[method(name = "register", param_kind = map)]
     fn register(
         &self,
         account: Option<String>,
         voice: bool,
         captcha: Option<String>,
-    ) -> Result<Value>;
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "removeContact", params = "named")]
+    #[method(name = "removeContact", param_kind = map)]
     fn remove_contact(
         &self,
         account: Option<String>,
         recipient: String,
         forget: bool,
-    ) -> Result<Value>;
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "removeDevice", params = "named")]
+    #[method(name = "removeDevice", param_kind = map)]
     fn remove_device(
         &self,
         account: Option<String>,
         #[allow(non_snake_case)] deviceId: u32,
-    ) -> Result<Value>;
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "removePin", params = "named")]
+    #[method(name = "removePin", param_kind = map)]
-    fn remove_pin(&self, account: Option<String>) -> Result<Value>;
+    fn remove_pin(&self, account: Option<String>) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "remoteDelete", params = "named")]
+    #[method(name = "remoteDelete", param_kind = map)]
     fn remote_delete(
         &self,
         account: Option<String>,
@@ -112,9 +126,9 @@ pub trait Rpc {
         recipients: Vec<String>,
         #[allow(non_snake_case)] groupIds: Vec<String>,
         #[allow(non_snake_case)] noteToSelf: bool,
-    ) -> Result<Value>;
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "send", params = "named")]
+    #[method(name = "send", param_kind = map)]
     fn send(
         &self,
         account: Option<String>,
@@ -132,21 +146,21 @@ pub trait Rpc {
         sticker: Option<String>,
         #[allow(non_snake_case)] storyTimestamp: Option<u64>,
         #[allow(non_snake_case)] storyAuthor: Option<String>,
-    ) -> Result<Value>;
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "sendContacts", params = "named")]
+    #[method(name = "sendContacts", param_kind = map)]
-    fn send_contacts(&self, account: Option<String>) -> Result<Value>;
+    fn send_contacts(&self, account: Option<String>) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "sendPaymentNotification", params = "named")]
+    #[method(name = "sendPaymentNotification", param_kind = map)]
     fn send_payment_notification(
         &self,
         account: Option<String>,
         recipient: String,
         receipt: String,
         note: String,
-    ) -> Result<Value>;
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "sendReaction", params = "named")]
+    #[method(name = "sendReaction", param_kind = map)]
     fn send_reaction(
         &self,
         account: Option<String>,
@@ -158,75 +172,75 @@ pub trait Rpc {
         #[allow(non_snake_case)] targetTimestamp: u64,
         remove: bool,
         story: bool,
-    ) -> Result<Value>;
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "sendReceipt", params = "named")]
+    #[method(name = "sendReceipt", param_kind = map)]
     fn send_receipt(
         &self,
         account: Option<String>,
         recipient: String,
         #[allow(non_snake_case)] targetTimestamps: Vec<u64>,
         r#type: String,
-    ) -> Result<Value>;
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "sendSyncRequest", params = "named")]
+    #[method(name = "sendSyncRequest", param_kind = map)]
-    fn send_sync_request(&self, account: Option<String>) -> Result<Value>;
+    fn send_sync_request(&self, account: Option<String>) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "sendTyping", params = "named")]
+    #[method(name = "sendTyping", param_kind = map)]
     fn send_typing(
         &self,
         account: Option<String>,
         recipients: Vec<String>,
         #[allow(non_snake_case)] groupIds: Vec<String>,
         stop: bool,
-    ) -> Result<Value>;
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "setPin", params = "named")]
+    #[method(name = "setPin", param_kind = map)]
-    fn set_pin(&self, account: Option<String>, pin: String) -> Result<Value>;
+    fn set_pin(&self, account: Option<String>, pin: String) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "submitRateLimitChallenge", params = "named")]
+    #[method(name = "submitRateLimitChallenge", param_kind = map)]
     fn submit_rate_limit_challenge(
         &self,
         account: Option<String>,
         challenge: String,
         captcha: String,
-    ) -> Result<Value>;
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "startLink", params = "named")]
+    #[method(name = "startLink", param_kind = map)]
-    fn start_link(&self, account: Option<String>) -> Result<JsonLink>;
+    fn start_link(&self, account: Option<String>) -> Result<JsonLink, ErrorObjectOwned>;
 
-    #[rpc(name = "trust", params = "named")]
+    #[method(name = "trust", param_kind = map)]
     fn trust(
         &self,
         account: Option<String>,
         recipient: String,
         #[allow(non_snake_case)] trustAllKnownKeys: bool,
         #[allow(non_snake_case)] verifiedSafetyNumber: Option<String>,
-    ) -> Result<Value>;
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "unblock", params = "named")]
+    #[method(name = "unblock", param_kind = map)]
     fn unblock(
         &self,
         account: Option<String>,
         recipients: Vec<String>,
         #[allow(non_snake_case)] groupIds: Vec<String>,
-    ) -> Result<Value>;
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "unregister", params = "named")]
+    #[method(name = "unregister", param_kind = map)]
     fn unregister(
         &self,
         account: Option<String>,
         #[allow(non_snake_case)] deleteAccount: bool,
-    ) -> Result<Value>;
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "updateAccount", params = "named")]
+    #[method(name = "updateAccount", param_kind = map)]
     fn update_account(
         &self,
         account: Option<String>,
         #[allow(non_snake_case)] deviceName: Option<String>,
-    ) -> Result<Value>;
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "updateConfiguration", params = "named")]
+    #[method(name = "updateConfiguration", param_kind = map)]
     fn update_configuration(
         &self,
         account: Option<String>,
@@ -234,18 +248,18 @@ pub trait Rpc {
         #[allow(non_snake_case)] unidentifiedDeliveryIndicators: Option<bool>,
         #[allow(non_snake_case)] typingIndicators: Option<bool>,
         #[allow(non_snake_case)] linkPreviews: Option<bool>,
-    ) -> Result<Value>;
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "updateContact", params = "named")]
+    #[method(name = "updateContact", param_kind = map)]
     fn update_contact(
         &self,
         account: Option<String>,
         recipient: String,
         name: Option<String>,
         expiration: Option<u32>,
-    ) -> Result<Value>;
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "updateGroup", params = "named")]
+    #[method(name = "updateGroup", param_kind = map)]
     fn update_group(
         &self,
         account: Option<String>,
@@ -265,9 +279,9 @@ pub trait Rpc {
         #[allow(non_snake_case)] setPermissionEditDetails: Option<String>,
         #[allow(non_snake_case)] setPermissionSendMessages: Option<String>,
         expiration: Option<u32>,
-    ) -> Result<Value>;
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "updateProfile", params = "named")]
+    #[method(name = "updateProfile", param_kind = map)]
     fn update_profile(
         &self,
         account: Option<String>,
@@ -278,32 +292,33 @@ pub trait Rpc {
         #[allow(non_snake_case)] mobileCoinAddress: Option<String>,
         avatar: Option<String>,
         #[allow(non_snake_case)] removeAvatar: bool,
-    ) -> Result<Value>;
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "uploadStickerPack", params = "named")]
+    #[method(name = "uploadStickerPack", param_kind = map)]
-    fn upload_sticker_pack(&self, account: Option<String>, path: String) -> Result<Value>;
+    fn upload_sticker_pack(
+        &self,
+        account: Option<String>,
+        path: String,
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[rpc(name = "verify", params = "named")]
+    #[method(name = "verify", param_kind = map)]
     fn verify(
         &self,
         account: Option<String>,
         #[allow(non_snake_case)] verificationCode: String,
         pin: Option<String>,
-    ) -> Result<Value>;
+    ) -> Result<Value, ErrorObjectOwned>;
 
-    #[pubsub(
+    #[subscription(
-        subscription = "receive",
+        name = "subscribeReceive" => "receive",
-        subscribe,
+        unsubscribe = "unsubscribeReceive",
-        name = "subscribeReceive",
+        item = Value,
-        params = "named"
+        param_kind = map
     )]
-    fn subscribe_receive(&self, _: Self::Metadata, _: Subscriber<Value>, account: Option<String>);
+    async fn subscribe_receive(&self, account: Option<String>) -> SubscriptionResult;
 
-    #[pubsub(subscription = "receive", unsubscribe, name = "unsubscribeReceive")]
+    #[method(name = "version")]
-    fn unsubscribe_receive(&self, _: Option<Self::Metadata>, _: SubscriptionId) -> Result<bool>;
+    fn version(&self) -> Result<Value, ErrorObjectOwned>;
-
-    #[rpc(name = "version")]
-    fn version(&self) -> Result<Value>;
 }
 
 #[derive(Deserialize)]
@@ -312,10 +327,20 @@ pub struct JsonLink {
     pub device_link_uri: String,
 }
 
-pub async fn connect_tcp(tcp: impl ToSocketAddrs) -> Result<SignalCliClient, RpcError> {
+pub async fn connect_tcp(tcp: impl ToSocketAddrs) -> Result<impl SubscriptionClientT, Error> {
-    super::tcp::connect::<_, SignalCliClient>(tcp).await
+    let (sender, receiver) = super::transports::tcp::connect(tcp).await?;
+
+    Ok(ClientBuilder::default().build_with_tokio(sender, receiver))
 }
 
-pub async fn connect_unix(socket_path: impl AsRef<Path>) -> Result<SignalCliClient, RpcError> {
+pub async fn connect_unix(
-    ipc::connect::<_, SignalCliClient>(socket_path).await
+    socket_path: impl AsRef<Path>,
+) -> Result<impl SubscriptionClientT, Error> {
+    let (sender, receiver) = super::transports::ipc::connect(socket_path).await?;
+
+    Ok(ClientBuilder::default().build_with_tokio(sender, receiver))
+}
+
+pub async fn connect_http(uri: &str) -> Result<impl SubscriptionClientT, Error> {
+    HttpClientBuilder::default().build(uri)
 }

client/src/main.rs

@@ -1,40 +1,53 @@
-use clap::Parser;
-use jsonrpc_client_transports::{RpcError, TypedSubscriptionStream};
-use jsonrpc_core::{futures_util::StreamExt, Value};
 use std::{path::PathBuf, time::Duration};
+
+use clap::Parser;
+use cli::Cli;
+use jsonrpsee::core::client::{Subscription, SubscriptionClientT};
+use jsonrpsee::core::Error as RpcError;
+use serde_json::Value;
 use tokio::{select, time::sleep};
 
 use crate::cli::{GroupPermission, LinkState};
+use crate::jsonrpc::RpcClient;
 
 mod cli;
-#[allow(clippy::too_many_arguments)]
+#[allow(non_snake_case, clippy::too_many_arguments)]
 mod jsonrpc;
-mod tcp;
+mod transports;
 
 const DEFAULT_TCP: &str = "127.0.0.1:7583";
 const DEFAULT_SOCKET_SUFFIX: &str = "signal-cli/socket";
+const DEFAULT_HTTP: &str = "http://localhost:8080/api/v1/rpc";
 
 #[tokio::main]
 async fn main() -> Result<(), anyhow::Error> {
     let cli = cli::Cli::parse();
 
-    let client = connect(&cli)
+    let result = connect(cli).await;
-        .await
-        .map_err(|e| anyhow::anyhow!("Failed to connect to socket: {e}"))?;
 
-    let result = match cli.command {
+    match result {
+        Ok(Value::Null) => {}
+        Ok(v) => println!("{v}"),
+        Err(e) => return Err(anyhow::anyhow!("JSON-RPC command failed: {e:?}")),
+    }
+    Ok(())
+}
+
+async fn handle_command(
+    cli: Cli,
+    client: impl SubscriptionClientT + Sync,
+) -> Result<Value, RpcError> {
+    match cli.command {
         cli::CliCommands::Receive { timeout } => {
-            let mut stream = client
+            let mut stream = client.subscribe_receive(cli.account).await?;
-                .subscribe_receive(cli.account)
-                .map_err(|e| anyhow::anyhow!("JSON-RPC command failed: {:?}", e))?;
 
             {
                 while let Some(v) = stream_next(timeout, &mut stream).await {
-                    let v = v.map_err(|e| anyhow::anyhow!("JSON-RPC command failed: {:?}", e))?;
+                    let v = v?;
                     println!("{v}");
                 }
             }
-            return Ok(());
+            Ok(Value::Null)
         }
         cli::CliCommands::AddDevice { uri } => client.add_device(cli.account, uri).await,
         cli::CliCommands::Block {
@@ -54,7 +67,7 @@ async fn main() -> Result<(), anyhow::Error> {
             let url = client
                 .start_link(cli.account)
                 .await
-                .map_err(|e| anyhow::anyhow!("JSON-RPC command startLink failed: {e:?}",))?
+                .map_err(|e| RpcError::Custom(format!("JSON-RPC command startLink failed: {e:?}")))?
                 .device_link_uri;
             println!("{}", url);
             client.finish_link(url, name).await
@@ -349,18 +362,28 @@ async fn main() -> Result<(), anyhow::Error> {
             pin,
         } => client.verify(cli.account, verification_code, pin).await,
         cli::CliCommands::Version => client.version().await,
-    };
+    }
-
-    result
-        .map(|v| println!("{v}"))
-        .map_err(|e| anyhow::anyhow!("JSON-RPC command failed: {e:?}",))?;
-    Ok(())
 }
 
-async fn connect(cli: &cli::Cli) -> Result<jsonrpc::SignalCliClient, RpcError> {
+async fn connect(cli: Cli) -> Result<Value, RpcError> {
-    if let Some(tcp) = cli.json_rpc_tcp {
+    if let Some(http) = &cli.json_rpc_http {
+        let uri = if let Some(uri) = http {
+            uri
+        } else {
+            DEFAULT_HTTP
+        };
+        let client = jsonrpc::connect_http(uri)
+            .await
+            .map_err(|e| RpcError::Custom(format!("Failed to connect to socket: {e}")))?;
+
+        handle_command(cli, client).await
+    } else if let Some(tcp) = cli.json_rpc_tcp {
         let socket_addr = tcp.unwrap_or_else(|| DEFAULT_TCP.parse().unwrap());
-        jsonrpc::connect_tcp(socket_addr).await
+        let client = jsonrpc::connect_tcp(socket_addr)
+            .await
+            .map_err(|e| RpcError::Custom(format!("Failed to connect to socket: {e}")))?;
+
+        handle_command(cli, client).await
     } else {
         let socket_path = cli
             .json_rpc_socket
@@ -374,13 +397,17 @@ async fn connect(cli: &cli::Cli) -> Result<jsonrpc::SignalCliClient, RpcError> {
                 })
             })
             .unwrap_or_else(|| ("/run".to_owned() + DEFAULT_SOCKET_SUFFIX).into());
-        jsonrpc::connect_unix(socket_path).await
+        let client = jsonrpc::connect_unix(socket_path)
+            .await
+            .map_err(|e| RpcError::Custom(format!("Failed to connect to socket: {e}")))?;
+
+        handle_command(cli, client).await
     }
 }
 
 async fn stream_next(
     timeout: f64,
-    stream: &mut TypedSubscriptionStream<Value>,
+    stream: &mut Subscription<Value>,
 ) -> Option<Result<Value, RpcError>> {
     if timeout < 0.0 {
         stream.next().await

client/src/tcp.rs

@@ -1,29 +0,0 @@
-use jsonrpc_client_transports::{transports::duplex, RpcChannel, RpcError};
-use jsonrpc_core::futures_util::{SinkExt, StreamExt, TryStreamExt};
-use jsonrpc_server_utils::{codecs::StreamCodec, tokio_util::codec::Decoder};
-use tokio::net::{TcpStream, ToSocketAddrs};
-
-/// Connect to a JSON-RPC TCP server.
-pub async fn connect<S: ToSocketAddrs, Client: From<RpcChannel>>(
-    socket: S,
-) -> Result<Client, RpcError> {
-    let connection = TcpStream::connect(socket)
-        .await
-        .map_err(|e| RpcError::Other(Box::new(e)))?;
-    let (sink, stream) = StreamCodec::stream_incoming().framed(connection).split();
-    let sink = sink.sink_map_err(|e| RpcError::Other(Box::new(e)));
-    let stream = stream.map_err(|e| log::error!("TCP stream error: {}", e));
-
-    let (client, sender) = duplex(
-        Box::pin(sink),
-        Box::pin(
-            stream
-                .take_while(|x| std::future::ready(x.is_ok()))
-                .map(|x| x.expect("Stream is closed upon first error.")),
-        ),
-    );
-
-    tokio::spawn(client);
-
-    Ok(sender.into())
-}

client/src/transports/ipc.rs

@@ -0,0 +1,23 @@
+use std::path::Path;
+
+use futures_util::stream::StreamExt;
+use jsonrpsee::core::client::{TransportReceiverT, TransportSenderT};
+use jsonrpsee::core::Error;
+use tokio::net::UnixStream;
+use tokio_util::codec::Decoder;
+
+use super::stream_codec::StreamCodec;
+use super::{Receiver, Sender};
+
+/// Connect to a JSON-RPC Unix Socket server.
+pub async fn connect(
+    socket: impl AsRef<Path>,
+) -> Result<(impl TransportSenderT + Send, impl TransportReceiverT + Send), Error> {
+    let connection = UnixStream::connect(socket).await?;
+    let (sink, stream) = StreamCodec::stream_incoming().framed(connection).split();
+
+    let sender = Sender { inner: sink };
+    let receiver = Receiver { inner: stream };
+
+    Ok((sender, receiver))
+}

client/src/transports/mod.rs

@@ -0,0 +1,64 @@
+use futures_util::{stream::StreamExt, Sink, SinkExt, Stream};
+use jsonrpsee::core::{
+    async_trait,
+    client::{ReceivedMessage, TransportReceiverT, TransportSenderT},
+};
+use thiserror::Error;
+
+pub mod ipc;
+mod stream_codec;
+pub mod tcp;
+
+#[derive(Debug, Error)]
+enum Errors {
+    #[error("Other: {0}")]
+    Other(String),
+    #[error("Closed")]
+    Closed,
+}
+
+struct Sender<T: Send + Sink<String>> {
+    inner: T,
+}
+
+#[async_trait]
+impl<T: Send + Sink<String, Error = impl std::error::Error> + Unpin + 'static> TransportSenderT
+    for Sender<T>
+{
+    type Error = Errors;
+
+    async fn send(&mut self, body: String) -> Result<(), Self::Error> {
+        self.inner
+            .send(body)
+            .await
+            .map_err(|e| Errors::Other(format!("{:?}", e)))?;
+        Ok(())
+    }
+
+    async fn close(&mut self) -> Result<(), Self::Error> {
+        self.inner
+            .close()
+            .await
+            .map_err(|e| Errors::Other(format!("{:?}", e)))?;
+        Ok(())
+    }
+}
+
+struct Receiver<T: Send + Stream> {
+    inner: T,
+}
+
+#[async_trait]
+impl<T: Send + Stream<Item = Result<String, std::io::Error>> + Unpin + 'static> TransportReceiverT
+    for Receiver<T>
+{
+    type Error = Errors;
+
+    async fn receive(&mut self) -> Result<ReceivedMessage, Self::Error> {
+        match self.inner.next().await {
+            None => Err(Errors::Closed),
+            Some(Ok(msg)) => Ok(ReceivedMessage::Text(msg)),
+            Some(Err(e)) => Err(Errors::Other(format!("{:?}", e))),
+        }
+    }
+}

client/src/transports/stream_codec.rs

@@ -0,0 +1,61 @@
+use bytes::BytesMut;
+use std::{io, str};
+use tokio_util::codec::{Decoder, Encoder};
+
+type Separator = u8;
+
+/// Stream codec for streaming protocols (ipc, tcp)
+#[derive(Debug, Default)]
+pub struct StreamCodec {
+    incoming_separator: Separator,
+    outgoing_separator: Separator,
+}
+
+impl StreamCodec {
+    /// Default codec with streaming input data. Input can be both enveloped and not.
+    pub fn stream_incoming() -> Self {
+        StreamCodec::new(b'\n', b'\n')
+    }
+
+    /// New custom stream codec
+    pub fn new(incoming_separator: Separator, outgoing_separator: Separator) -> Self {
+        StreamCodec {
+            incoming_separator,
+            outgoing_separator,
+        }
+    }
+}
+
+impl Decoder for StreamCodec {
+    type Item = String;
+    type Error = io::Error;
+
+    fn decode(&mut self, buf: &mut BytesMut) -> io::Result<Option<Self::Item>> {
+        if let Some(i) = buf
+            .as_ref()
+            .iter()
+            .position(|&b| b == self.incoming_separator)
+        {
+            let line = buf.split_to(i);
+            let _ = buf.split_to(1);
+
+            match str::from_utf8(line.as_ref()) {
+                Ok(s) => Ok(Some(s.to_string())),
+                Err(_) => Err(io::Error::new(io::ErrorKind::Other, "invalid UTF-8")),
+            }
+        } else {
+            Ok(None)
+        }
+    }
+}
+
+impl Encoder<String> for StreamCodec {
+    type Error = io::Error;
+
+    fn encode(&mut self, msg: String, buf: &mut BytesMut) -> io::Result<()> {
+        let mut payload = msg.into_bytes();
+        payload.push(self.outgoing_separator);
+        buf.extend_from_slice(&payload);
+        Ok(())
+    }
+}

client/src/transports/tcp.rs

@@ -0,0 +1,21 @@
+use futures_util::stream::StreamExt;
+use jsonrpsee::core::client::{TransportReceiverT, TransportSenderT};
+use jsonrpsee::core::Error;
+use tokio::net::{TcpStream, ToSocketAddrs};
+use tokio_util::codec::Decoder;
+
+use super::stream_codec::StreamCodec;
+use super::{Receiver, Sender};
+
+/// Connect to a JSON-RPC TCP server.
+pub async fn connect(
+    socket: impl ToSocketAddrs,
+) -> Result<(impl TransportSenderT + Send, impl TransportReceiverT + Send), Error> {
+    let connection = TcpStream::connect(socket).await?;
+    let (sink, stream) = StreamCodec::stream_incoming().framed(connection).split();
+
+    let sender = Sender { inner: sink };
+    let receiver = Receiver { inner: stream };
+
+    Ok((sender, receiver))
+}