Initial commit

Change-Id: I2b916ff0acd2a88aeef709cf4f900503e823d44d
diff --git a/client/src/transports/codec.rs b/client/src/transports/codec.rs
new file mode 100644
index 0000000..6e69571
--- /dev/null
+++ b/client/src/transports/codec.rs
@@ -0,0 +1,37 @@
+use bytes::{BufMut, BytesMut};
+use serde_json::Value;
+use std::io;
+use tokio_util::codec::{Decoder, Encoder};
+
+pub struct JsonCodec;
+
+impl Encoder<BytesMut> for JsonCodec {
+    type Error = io::Error;
+
+    fn encode(&mut self, data: BytesMut, buf: &mut BytesMut) -> Result<(), io::Error> {
+        buf.reserve(data.len());
+        buf.put(data);
+        Ok(())
+    }
+}
+
+impl Decoder for JsonCodec {
+    type Item = Value;
+    type Error = io::Error;
+
+    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Value>, io::Error> {
+        if src.is_empty() {
+            return Ok(None);
+        }
+
+        match serde_json::from_slice::<Value>(src) {
+            Ok(val) => {
+                src.clear();
+
+                Ok(Some(val))
+            }
+            Err(ref e) if e.is_eof() => Ok(None),
+            Err(e) => Err(e.into()),
+        }
+    }
+}
diff --git a/client/src/transports/ipc.rs b/client/src/transports/ipc.rs
new file mode 100644
index 0000000..09798a7
--- /dev/null
+++ b/client/src/transports/ipc.rs
@@ -0,0 +1,18 @@
+use crate::transports::{Receiver, Sender, codec::JsonCodec};
+use futures_util::stream::StreamExt;
+use jsonrpsee::core::client::{TransportReceiverT, TransportSenderT};
+use std::{io::Error, path::Path};
+use tokio::net::UnixStream;
+use tokio_util::codec::Framed;
+
+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) = Framed::new(connection, JsonCodec).split();
+
+    let sender = Sender { inner: sink };
+    let receiver = Receiver { inner: stream };
+
+    Ok((sender, receiver))
+}
diff --git a/client/src/transports/mod.rs b/client/src/transports/mod.rs
new file mode 100644
index 0000000..7456643
--- /dev/null
+++ b/client/src/transports/mod.rs
@@ -0,0 +1,115 @@
+mod codec;
+pub mod ipc;
+pub mod tcp;
+
+use bytes::BytesMut;
+use futures_util::{Sink, SinkExt, Stream, stream::StreamExt};
+use jsonrpsee::core::{
+    async_trait,
+    client::{ReceivedMessage, TransportReceiverT, TransportSenderT},
+};
+use serde_json::{Value, json};
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+enum TransportError {
+    #[error("Connection closed.")]
+    ConnectionClosed,
+
+    #[error("IO error: {0}")]
+    Io(#[from] std::io::Error),
+
+    #[error("Unkown error: {0}")]
+    Unknown(String),
+}
+
+struct Sender<T: Send + Sink<BytesMut>> {
+    inner: T,
+}
+
+#[async_trait]
+impl<T: Send + Sink<BytesMut, Error = impl std::error::Error> + Unpin + 'static> TransportSenderT
+    for Sender<T>
+{
+    type Error = TransportError;
+
+    async fn send(&mut self, body: String) -> Result<(), Self::Error> {
+        let mut message: Value =
+            serde_json::from_str(&body).map_err(|e| TransportError::Unknown(e.to_string()))?;
+
+        // NOTE(mnaser): In order to be able to use the subscription client, we need to
+        //               drop the subscription message for the "update" method, as the
+        //               remote doesn't support JSON-RPC 2.0.
+        if message["method"] == json!("update") {
+            return Ok(());
+        }
+
+        // NOTE(mnaser): jsonrpsee runs using JSON-RPC 2.0 only which the remote doesn't
+        //               support, so we intercept the message, remove "jsonrpc" and then
+        //               send the message.
+        message.as_object_mut().map(|obj| obj.remove("jsonrpc"));
+
+        // NOTE(mnaser): OVSDB expects all requests to have a "params" key, so we add an
+        //               empty array if it doesn't exist.
+        if !message.as_object().unwrap().contains_key("params") {
+            message["params"] = json!([]);
+        }
+
+        self.inner
+            .send(BytesMut::from(message.to_string().as_str()))
+            .await
+            .map_err(|e| TransportError::Unknown(e.to_string()))?;
+
+        Ok(())
+    }
+
+    async fn close(&mut self) -> Result<(), Self::Error> {
+        self.inner
+            .close()
+            .await
+            .map_err(|e| TransportError::Unknown(e.to_string()))?;
+
+        Ok(())
+    }
+}
+
+struct Receiver<T: Send + Stream> {
+    inner: T,
+}
+
+#[async_trait]
+impl<T: Send + Stream<Item = Result<Value, std::io::Error>> + Unpin + 'static> TransportReceiverT
+    for Receiver<T>
+{
+    type Error = TransportError;
+
+    async fn receive(&mut self) -> Result<ReceivedMessage, Self::Error> {
+        match self.inner.next().await {
+            None => Err(TransportError::ConnectionClosed),
+            Some(Ok(mut message)) => {
+                // NOTE(mnaser): jsonrpsee runs using JSON-RPC 2.0 only which the remote doesn't
+                //               support, so we intercept the message, add "jsonrpc" and then
+                //               send the message.
+                message
+                    .as_object_mut()
+                    .map(|obj| obj.insert("jsonrpc".to_string(), json!("2.0")));
+
+                // NOTE(mnaser): jsonrpsee expects no error field if there is a result, due to the
+                //               remote not supporting JSON-RPC 2.0, we need to remove the "error"
+                //               field if there is a "result" field.
+                if message.as_object().unwrap().contains_key("result") {
+                    message.as_object_mut().map(|obj| obj.remove("error"));
+                }
+
+                // NOTE(mnaser): If a message comes in with it's "id" field set to null, then
+                //               we remove it.
+                if message.as_object().unwrap().contains_key("id") && message["id"] == json!(null) {
+                    message.as_object_mut().map(|obj| obj.remove("id"));
+                }
+
+                Ok(ReceivedMessage::Bytes(message.to_string().into_bytes()))
+            }
+            Some(Err(e)) => Err(TransportError::Io(e)),
+        }
+    }
+}
diff --git a/client/src/transports/tcp.rs b/client/src/transports/tcp.rs
new file mode 100644
index 0000000..2599b64
--- /dev/null
+++ b/client/src/transports/tcp.rs
@@ -0,0 +1,18 @@
+use crate::transports::{Receiver, Sender, codec::JsonCodec};
+use futures_util::stream::StreamExt;
+use jsonrpsee::core::client::{TransportReceiverT, TransportSenderT};
+use std::io::Error;
+use tokio::net::{TcpStream, ToSocketAddrs};
+use tokio_util::codec::Framed;
+
+pub async fn connect(
+    socket: impl ToSocketAddrs,
+) -> Result<(impl TransportSenderT + Send, impl TransportReceiverT + Send), Error> {
+    let connection = TcpStream::connect(socket).await?;
+    let (sink, stream) = Framed::new(connection, JsonCodec).split();
+
+    let sender = Sender { inner: sink };
+    let receiver = Receiver { inner: stream };
+
+    Ok((sender, receiver))
+}