rtc_config

crates/livekit-core/src/lib.rs

@@ -4,5 +4,7 @@ pub mod proto {
 
 mod rtc_engine;
 mod signal_client;
+mod lk_runtime;
+mod pc_transport;
 
 pub mod room;

crates/livekit-core/src/lk_runtime.rs

@@ -0,0 +1,25 @@
+use log::trace;
+
+use livekit_webrtc::peer_connection_factory::PeerConnectionFactory;
+use livekit_webrtc::webrtc::RTCRuntime;
+
+pub struct LKRuntime {
+    pub rtc_runtime: RTCRuntime,
+    pub pc_factory: PeerConnectionFactory,
+}
+
+impl LKRuntime {
+    pub fn new() -> Self {
+        trace!("LKRuntime::new()");
+        Self {
+            rtc_runtime: RTCRuntime::new(),
+            pc_factory: PeerConnectionFactory::new(),
+        }
+    }
+}
+
+impl Drop for LKRuntime {
+    fn drop(&mut self) {
+        trace!("LKRuntime::drop()");
+    }
+}

crates/livekit-core/src/pc_transport.rs

@@ -0,0 +1,105 @@
+use std::sync::Arc;
+use std::time::Duration;
+
+use log::{error, trace};
+
+use livekit_webrtc::jsep::{IceCandidate, SessionDescription};
+use livekit_webrtc::peer_connection::{PeerConnection, RTCOfferAnswerOptions, SdpError, SignalingState};
+use livekit_webrtc::peer_connection_factory::RTCConfiguration;
+use livekit_webrtc::rtc_error::RTCError;
+
+use crate::lk_runtime::LKRuntime;
+
+const NEGOTIATION_FREQUENCY: Duration = Duration::from_millis(150); // TODO(theomonnom)
+
+pub type OnOfferHandler = Box<dyn FnMut(SessionDescription)>;
+
+pub struct PCTransport {
+    peer_connection: PeerConnection,
+    pending_candidates: Vec<IceCandidate>,
+    on_offer_handler: Option<OnOfferHandler>,
+    restarting_ice: bool,
+    renegotiate: bool,
+}
+
+impl PCTransport {
+    pub fn new(lk_runtime: Arc<LKRuntime>, cfg: RTCConfiguration) -> Result<Self, RTCError> {
+        let peer_connection = lk_runtime.pc_factory.create_peer_connection(cfg)?;
+
+        Ok(Self {
+            peer_connection,
+            pending_candidates: Vec::default(),
+            on_offer_handler: None,
+            restarting_ice: false,
+            renegotiate: false,
+        })
+    }
+
+    pub fn peer_connection(&mut self) -> &mut PeerConnection {
+        &mut self.peer_connection
+    }
+
+    pub fn on_offer(&mut self, handler: OnOfferHandler) {
+        self.on_offer_handler = Some(handler);
+    }
+
+    pub fn add_ice_candidate(&mut self, ice_candidate: IceCandidate) {
+        if self.peer_connection.remote_description().is_none() {
+            self.pending_candidates.push(ice_candidate);
+            return;
+        }
+
+        self.peer_connection.add_ice_candidate(ice_candidate);
+    }
+
+    pub async fn set_remote_description(&mut self, remote_description: SessionDescription) -> Result<(), SdpError> {
+        self.peer_connection.set_remote_description(remote_description).await?;
+
+        for ic in self.pending_candidates.drain(..) {
+            self.peer_connection.add_ice_candidate(ic);
+        }
+        self.restarting_ice = false;
+
+        if self.renegotiate {
+            self.renegotiate = false;
+            self.create_and_send_offer(RTCOfferAnswerOptions::default()).await?;
+        }
+
+        Ok(())
+    }
+
+    pub async fn negotiate(&mut self) -> Result<(), SdpError> {
+        // TODO(theomonnom) Debounce here with NEGOTIATION_FREQUENCY
+        self.create_and_send_offer(RTCOfferAnswerOptions::default()).await
+    }
+
+    async fn create_and_send_offer(&mut self, options: RTCOfferAnswerOptions) -> Result<(), SdpError> {
+        if self.on_offer_handler.is_none() {
+            return Ok(());
+        }
+
+        if options.ice_restart {
+            trace!("restarting ICE");
+            self.restarting_ice = true;
+        }
+
+        if self.peer_connection.signaling_state() == SignalingState::HaveLocalOffer {
+            if options.ice_restart {
+                if let Some(remote_description) = self.peer_connection.remote_description() {
+                    self.peer_connection.set_remote_description(remote_description).await?;
+                } else {
+                    error!("trying to ice restart when the pc doesn't have remote description");
+                }
+            } else {
+                self.renegotiate = true;
+                return Ok(());
+            }
+        }
+
+        let offer = self.peer_connection.create_offer(options).await?;
+        trace!("created offer {:?}", offer);
+        self.peer_connection.set_local_description(offer.clone()).await?;
+        self.on_offer_handler.as_mut().unwrap()(offer);
+        Ok(())
+    }
+}

crates/livekit-core/src/rtc_engine.rs

@@ -1,49 +1,264 @@
-use crate::proto::signal_response;
-use crate::signal_client::{SignalClient, SignalClientError};
-use log::error;
-use tokio::sync::broadcast;
+use std::sync::{Arc, Mutex, Weak};
 
-pub struct RTCEngine {
-    signal_client: SignalClient,
+use lazy_static::lazy_static;
+use log::{error, trace};
+use prost::Message as ProstMessage;
+use thiserror::Error;
+use tokio::sync::mpsc;
+
+use livekit_webrtc::data_channel::{DataChannel, DataChannelInit};
+use livekit_webrtc::jsep::{IceCandidate, SessionDescription};
+use livekit_webrtc::peer_connection::PeerConnectionState;
+use livekit_webrtc::peer_connection_factory::{
+    ContinualGatheringPolicy, ICEServer, IceTransportsType, RTCConfiguration,
+};
+use livekit_webrtc::rtc_error::RTCError;
+
+use crate::lk_runtime::LKRuntime;
+use crate::pc_transport::PCTransport;
+use crate::proto::{DataPacket, JoinResponse, signal_request, SignalTarget, TrickleRequest};
+use crate::proto::signal_response::Message;
+use crate::signal_client;
+use crate::signal_client::{SignalClient, SignalError};
+
+const LOSSY_DC_LABEL: &str = "_lossy";
+const RELIABLE_DC_LABEL: &str = "_reliable";
+
+#[derive(Error, Debug)]
+pub enum EngineError {
+    #[error("signal failure")]
+    SignalError(#[from] SignalError),
+    #[error("internal webrtc failure")]
+    RTCError(#[from] RTCError),
 }
 
-impl RTCEngine {
-    pub fn new() -> RTCEngine {
-        Self {
-            signal_client: SignalClient::new(),
-        }
-    }
+lazy_static! {
+    // Share one LKRuntime across all RTCEngine instances
+    static ref LK_RUNTIME: Mutex<Weak<LKRuntime>> = Mutex::new(Weak::new());
+}
 
-    pub async fn connect(&mut self, url: &str, token: &str) -> Result<(), SignalClientError> {
-        self.signal_client.connect(url, token).await?;
+enum EngineMessage {}
 
-        tokio::spawn(Self::handle_rtc(
-            self.signal_client.response_rx.resubscribe(),
-        ));
+struct PeerInternal {
+    publisher_pc: PCTransport,
+    subscriber_pc: PCTransport,
 
-        Ok(())
-    }
+    lossy_dc: DataChannel,
+    reliable_dc: DataChannel,
 
-    pub fn update(&self) {}
+    pub_ice_rx: mpsc::Receiver<IceCandidate>,
+    sub_ice_rx: mpsc::Receiver<IceCandidate>,
 
-    async fn handle_rtc(mut signal_receiver: broadcast::Receiver<signal_response::Message>) {
-        loop {
-            let msg = match signal_receiver.recv().await {
-                Ok(msg) => msg,
-                Err(error) => {
-                    error!("Failed to receive SignalResponse: {:?}", error);
-                    continue;
-                }
-            };
+    pub_offer_rx: mpsc::Receiver<SessionDescription>,
 
-            match msg {
-                signal_response::Message::Join(join) => {}
-                signal_response::Message::Trickle(trickle) => {}
-                signal_response::Message::Answer(answer) => {}
-                signal_response::Message::Offer(offer) => {}
-                _ => {}
+    primary_connection_state_rx: mpsc::Receiver<PeerConnectionState>,
+    secondary_connection_state_rx: mpsc::Receiver<PeerConnectionState>,
+
+    lossy_data_rx: mpsc::Receiver<DataPacket>,
+    reliable_data_rx: mpsc::Receiver<DataPacket>,
+}
+
+struct RTCInternal {
+    lk_runtime: Arc<LKRuntime>,
+    signal_client: Arc<SignalClient>,
+    pc_internal: PeerInternal,
+}
+
+impl RTCInternal {
+    async fn connect(
+        url: &str,
+        token: &str,
+    ) -> Result<Self, EngineError> {
+        let mut lk_runtime = None;
+        {
+            // Acquire an existing/a new LKRuntime
+            let mut lk_runtime_ref = LK_RUNTIME.lock().unwrap();
+            lk_runtime = lk_runtime_ref.upgrade();
+
+            if lk_runtime.is_none() {
+                let new_runtime = Arc::new(LKRuntime::new());
+                *lk_runtime_ref = Arc::downgrade(&new_runtime);
+                lk_runtime = Some(new_runtime);
             }
         }
+        let lk_runtime = lk_runtime.unwrap();
+        let signal_client = Arc::new(signal_client::connect(url, token).await?);
+
+        if let Message::Join(join) = signal_client.recv().await? {
+            let pc_internal = Self::configure(lk_runtime.clone(), join)?;
+
+            Ok(Self {
+                lk_runtime,
+                signal_client,
+                pc_internal,
+            })
+        } else {
+            panic!("the first received message isn't a JoinResponse");
+        }
+    }
+
+    async fn run(&mut self) {
+        loop {
+            tokio::select! {
+                Ok(signal) = self.signal_client.recv() => {
+
+                },
+                Some(ice_candidate) = self.pc_internal.pub_ice_rx.recv() => {
+                    tokio::spawn({
+                        let sc = self.signal_client.clone();
+
+                        async move {
+                            let _ = sc.send(signal_request::Message::Trickle(TrickleRequest {
+                                candidate_init: ice_candidate.to_string(),
+                                target: SignalTarget::Publisher as i32
+                            })).await;
+                        }
+                    });
+                },
+                Some(ice_candidate) = self.pc_internal.sub_ice_rx.recv() => {
+                    tokio::spawn({
+                        let sc = self.signal_client.clone();
+
+                        async move {
+                            let _ = sc.send(signal_request::Message::Trickle(TrickleRequest {
+                                candidate_init: ice_candidate.to_string(),
+                                target: SignalTarget::Subscriber as i32
+                            })).await;
+                        }
+                    });
+                },
+                Some(sdp) = self.pc_internal.pub_offer_rx.recv() => {
+
+                },
+                Some(state) = self.pc_internal.primary_connection_state_rx.recv() => {
+
+                },
+                Some(state) = self.pc_internal.secondary_connection_state_rx.recv() => {
+
+                },
+                Some(data) = self.pc_internal.lossy_data_rx.recv() => {
+
+                },
+                Some(data) = self.pc_internal.reliable_data_rx.recv() => {
+
+                }
+            }
+        }
+    }
+
+    fn configure(lk_runtime: Arc<LKRuntime>, join: JoinResponse) -> Result<PeerInternal, EngineError> {
+        let cfg = RTCConfiguration {
+            ice_servers: {
+                let mut servers = vec![];
+                for is in join.ice_servers {
+                    servers.push(ICEServer {
+                        urls: is.urls,
+                        username: is.username,
+                        password: is.credential,
+                    })
+                }
+                servers
+            },
+            continual_gathering_policy: ContinualGatheringPolicy::GatherContinually,
+            ice_transport_type: IceTransportsType::All,
+        };
+
+        // Create the PeerConnections
+        let mut publisher_pc = PCTransport::new(lk_runtime.clone(), cfg.clone())?;
+        let mut subscriber_pc = PCTransport::new(lk_runtime, cfg)?;
+
+        let (pub_ice_tx, pub_ice_rx) = mpsc::channel(8);
+        let (sub_ice_tx, sub_ice_rx) = mpsc::channel(8);
+        let (pub_offer_tx, pub_offer_rx) = mpsc::channel(8);
+        let (primary_connection_state_tx, primary_connection_state_rx) = mpsc::channel(8);
+        let (secondary_connection_state_tx, secondary_connection_state_rx) = mpsc::channel(8);
+        let (lossy_data_tx, lossy_data_rx) = mpsc::channel(8);
+        let (reliable_data_tx, reliable_data_rx) = mpsc::channel(8);
+
+        publisher_pc.peer_connection().on_ice_candidate(Box::new(move |ice_candidate| {
+            trace!("publisher - on_ice_candidate: {:?}", ice_candidate);
+            let _ = pub_ice_tx.blocking_send(ice_candidate);
+        }));
+
+        subscriber_pc.peer_connection().on_ice_candidate(Box::new(move |ice_candidate| {
+            trace!("subscriber - on_ice_candidate: {:?}", ice_candidate);
+            let _ = sub_ice_tx.blocking_send(ice_candidate);
+        }));
+
+        publisher_pc.on_offer(Box::new(move |offer| {
+            trace!("publisher - on_offer: {:?}", offer);
+            let _ = pub_offer_tx.blocking_send(offer); // TODO(theomonnom) Don't use blocking_send here
+        }));
+
+        let mut primary_pc = &publisher_pc;
+        let mut secondary_pc = &subscriber_pc;
+        if join.subscriber_primary {
+            primary_pc = &subscriber_pc;
+            secondary_pc = &publisher_pc;
+        }
+
+        primary_pc.peer_connection().on_connection_change(Box::new(move |state| {
+            let _ = primary_connection_state_tx.blocking_send(state);
+        }));
+
+        secondary_pc.peer_connection().on_connection_change(Box::new(move |state| {
+            let _ = secondary_connection_state_tx.blocking_send(state);
+        }));
+
+        let mut lossy_dc = publisher_pc.peer_connection().create_data_channel(LOSSY_DC_LABEL, {
+            let mut dc_init = DataChannelInit::default();
+            dc_init.ordered = true;
+            dc_init.max_retransmits = Some(0);
+            dc_init
+        })?;
+
+        let mut reliable_dc = publisher_pc.peer_connection().create_data_channel(RELIABLE_DC_LABEL, {
+            let mut dc_init = DataChannelInit::default();
+            dc_init.ordered = true;
+            dc_init
+        })?;
+
+        lossy_dc.on_message(Box::new(|data, binary| {
+            if let Ok(data) = DataPacket::decode(data) {
+                let _ = lossy_data_tx.blocking_send(data);
+            } else {
+                trace!("lossy_dc - failed to decode DataPacket");
+            }
+        }));
+
+        reliable_dc.on_message(Box::new(|data, binary| {
+            if let Ok(data) = DataPacket::decode(data) {
+                let _ = reliable_data_tx.blocking_send(data);
+            } else {
+                trace!("reliable_dc - failed to decode DataPacket");
+            }
+        }));
+
+        Ok(PeerInternal {
+            publisher_pc,
+            subscriber_pc,
+            lossy_dc,
+            reliable_dc,
+            pub_ice_rx,
+            sub_ice_rx,
+            pub_offer_rx,
+            primary_connection_state_rx,
+            secondary_connection_state_rx,
+            lossy_data_rx,
+            reliable_data_rx,
+        })
+    }
+}
+
+pub struct RTCEngine {}
+
+/// Initialize the SignalClient & the PeerConnections
+//pub async fn connect(url: &str, token: &str) -> Result<RTCEngine, EngineError> {
+//}
+
+impl RTCEngine {
+    async fn rtc_handle() {
+        loop {}
     }
 }
 
@@ -51,9 +266,25 @@ impl RTCEngine {
 async fn test_test() {
     env_logger::init();
 
-    let mut engine = RTCEngine::new();
-    engine.connect("ws://localhost:7880", "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE2NjQ1OTY4MDYsImlzcyI6IkFQSUNrSG04M01oZ2hQeCIsIm5hbWUiOiJ1c2VyMSIsIm5iZiI6MTY2MDk5NjgwNiwic3ViIjoidXNlcjEiLCJ2aWRlbyI6eyJyb29tIjoibXktZmlyc3Qtcm9vbSIsInJvb21Kb2luIjp0cnVlfX0.SWU_LETMK6ZmFOf38pYjVhpur0o7jJc6u61h8BH7g20").await.unwrap();
-
-    // Wait before exiting the program
-    tokio::time::sleep(core::time::Duration::from_millis(1000 * 25)).await;
+    //engine.connect("ws://localhost:7880", "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE2NjQ1OTY4MDYsImlzcyI6IkFQSUNrSG04M01oZ2hQeCIsIm5hbWUiOiJ1c2VyMSIsIm5iZiI6MTY2MDk5NjgwNiwic3ViIjoidXNlcjEiLCJ2aWRlbyI6eyJyb29tIjoibXktZmlyc3Qtcm9vbSIsInJvb21Kb2luIjp0cnVlfX0.SWU_LETMK6ZmFOf38pYjVhpur0o7jJc6u61h8BH7g20").await.unwrap();
 }
+
+/*sync fn handle_rtc(mut signal_receiver: broadcast::Receiver<Message>) {
+    loop {
+        let msg = match signal_receiver.recv().await {
+            Ok(msg) => msg,
+            Err(error) => {
+                error!("Failed to receive SignalResponse: {:?}", error);
+                continue;
+            }
+        };
+
+        match msg {
+            Message::Join(join) => {}
+            Message::Trickle(trickle) => {}
+            Message::Answer(answer) => {}
+            Message::Offer(offer) => {}
+            _ => {}
+        }
+    }
+}*/

crates/livekit-core/src/signal_client.rs

@@ -1,100 +1,186 @@
-use futures_util::SinkExt;
-use futures_util::StreamExt;
+use futures_util::{SinkExt, StreamExt};
+use futures_util::stream::{SplitSink, SplitStream};
 use log::{error, info};
-use prost::Message as ProtoMessage;
-use std::borrow::Borrow;
+use prost::Message as ProstMessage;
 use thiserror::Error;
 use tokio::net::TcpStream;
-use tokio::sync::broadcast;
+use tokio::sync::{mpsc, oneshot};
 use tokio::task::JoinHandle;
-use tokio_tungstenite::tungstenite::Message;
 use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream};
+use tokio_tungstenite::tungstenite::{
+    Error as WsError,
+    Message, protocol::frame::{CloseFrame, coding::CloseCode},
+};
 
-use crate::{proto, proto::signal_response};
+use crate::proto::{signal_request, signal_response, SignalRequest, SignalResponse};
+
+pub const PROTOCOL_VERSION: u32 = 8;
 
 #[derive(Error, Debug)]
-pub enum SignalClientError {
+pub enum SignalError {
     #[error("websocket failure")]
-    WebSocket(#[from] tokio_tungstenite::tungstenite::Error),
+    WsError(#[from] WsError),
     #[error("failed to parse the url")]
     UrlParse(#[from] url::ParseError),
-    #[error("failed to parse messages from server")]
+    #[error("failed to decode messages from server")]
     ProtoParse(#[from] prost::DecodeError),
 }
 
+type SignalResult<T> = Result<T, SignalError>;
+type WebSocket = WebSocketStream<MaybeTlsStream<TcpStream>>;
+
+#[derive(Debug)]
+struct RecvMessage {
+    response_chn: oneshot::Sender<SignalResult<signal_response::Message>>,
+}
+
+#[derive(Debug)]
+struct SendMessage {
+    signal: signal_request::Message,
+    response_chn: oneshot::Sender<SignalResult<()>>,
+}
+
 pub struct SignalClient {
-    ws_handle: Option<JoinHandle<Result<(), SignalClientError>>>,
-    response_tx: broadcast::Sender<signal_response::Message>,
-    pub response_rx: broadcast::Receiver<signal_response::Message>,
+    read_sender: mpsc::Sender<RecvMessage>,
+    write_sender: mpsc::Sender<SendMessage>,
+    write_shutdown_sender: oneshot::Sender<()>,
+    read_shutdown_sender: oneshot::Sender<()>,
+    read_handle: JoinHandle<()>,
+    write_handle: JoinHandle<()>,
+}
+
+pub async fn connect(url: &str, token: &str) -> SignalResult<SignalClient> {
+    let mut lk_url = url::Url::parse(url)?;
+    lk_url.set_path("/rtc");
+    lk_url
+        .query_pairs_mut()
+        .append_pair("access_token", token)
+        .append_pair("protocol", PROTOCOL_VERSION.to_string().as_str());
+
+    let (ws_stream, _) = connect_async(lk_url).await?;
+    let (ws_writer, ws_reader) = ws_stream.split();
+
+    let (read_tx, read_rx) = mpsc::channel::<RecvMessage>(1);
+    let (write_tx, write_rx) = mpsc::channel::<SendMessage>(1);
+    let (read_shutdown_tx, read_shutdown_rx) = oneshot::channel();
+    let (write_shutdown_tx, write_shutdown_rx) = oneshot::channel();
+
+    let read_handle = tokio::spawn(SignalClient::ws_read(read_rx, ws_reader, read_shutdown_rx));
+    let write_handle = tokio::spawn(SignalClient::ws_write(
+        write_rx,
+        ws_writer,
+        write_shutdown_rx,
+    ));
+
+    Ok(SignalClient {
+        read_sender: read_tx,
+        write_sender: write_tx,
+        write_shutdown_sender: write_shutdown_tx,
+        read_shutdown_sender: read_shutdown_tx,
+        read_handle,
+        write_handle,
+    })
 }
 
 impl SignalClient {
-    pub fn new() -> Self {
-        let (tx, rx) = broadcast::channel(16);
-
-        Self {
-            response_tx: tx,
-            response_rx: rx,
-            ws_handle: None,
-        }
+    pub async fn close(self) {
+        let _ = self.write_shutdown_sender.send(());
+        let _ = self.write_handle.await;
+        let _ = self.read_shutdown_sender.send(());
+        let _ = self.read_handle.await;
     }
 
-    pub async fn connect(&mut self, url: &str, token: &str) -> Result<(), SignalClientError> {
-        let mut lk_url = url::Url::parse(url)?;
-        lk_url.set_path("/rtc");
-        lk_url
-            .query_pairs_mut()
-            .append_pair("access_token", token)
-            .append_pair("protocol", "8");
-
-        info!("Connecting to {}", lk_url);
-        let (ws, _) = connect_async(&lk_url).await?;
-
-        self.ws_handle = Some(tokio::spawn(Self::handle_ws(ws, self.response_tx.clone())));
-        Ok(())
+    pub async fn recv(&self) -> SignalResult<signal_response::Message> {
+        let (send, recv) = oneshot::channel();
+        let msg = RecvMessage { response_chn: send };
+        let _ = self.read_sender.send(msg).await;
+        recv.await.expect("channel closed")
     }
 
-    pub async fn disconnect() {
-        unimplemented!()
+    pub async fn send(&self, signal: signal_request::Message) -> SignalResult<()> {
+        let (send, recv) = oneshot::channel();
+        let msg = SendMessage {
+            signal,
+            response_chn: send,
+        };
+        let _ = self.write_sender.send(msg).await;
+        recv.await.expect("channel closed")
     }
 
-    async fn handle_ws(
-        mut ws: WebSocketStream<MaybeTlsStream<TcpStream>>,
-        response_tx: broadcast::Sender<signal_response::Message>,
-    ) -> Result<(), SignalClientError> {
+    async fn ws_write(
+        mut write_receiver: mpsc::Receiver<SendMessage>,
+        mut ws_writer: SplitSink<WebSocket, Message>,
+        mut shutdown_receiver: oneshot::Receiver<()>,
+    ) {
         loop {
             tokio::select! {
-                next_msg = ws.next() => {
-                    let ws_msg = match next_msg {
-                        Some(msg) => msg?,
-                        None => break,
+                Some(msg) = write_receiver.recv() => {
+                    let req = SignalRequest {
+                        message: Some(msg.signal),
                     };
 
-                    let data = match ws_msg {
-                        Message::Binary(data) => data,
-                        Message::Ping(data) => {
-                            ws.send(Message::Pong(data)).await?;
-                            continue
-                        },
-                        Message::Close(_frame) => break,
-                        _ => continue,
-                    };
-
-                    let proto_msg = proto::SignalResponse::decode(data.borrow())?;
-                    let signal_response = proto_msg.message.unwrap();
-
-                    match signal_response {
-                        signal_response::Message::Pong(ts) => {
-
-                        },
-                        _ => {
-                            response_tx.send(signal_response).unwrap();
-                        }
+                    let write_res = ws_writer.send(Message::Binary(req.encode_to_vec())).await;
+                    if let Err(err) = write_res {
+                        error!("failed to send message to ws: {:?}", err);
+                        let _ = msg.response_chn.send(Err(err.into()));
+                        break;
                     }
+
+                    let _ = msg.response_chn.send(Ok(()));
+                },
+                _ = (&mut shutdown_receiver) => {
+                    let _ = ws_writer.send(Message::Close(Some(CloseFrame {
+                        code: CloseCode::Normal,
+                        reason: "disconnected by client".into()
+                    }))).await;
+                    let _ = ws_writer.flush().await;
+                    break;
                 }
             }
         }
+    }
 
-        Ok(())
+    async fn ws_read(
+        mut write_receiver: mpsc::Receiver<RecvMessage>,
+        mut ws_reader: SplitStream<WebSocket>,
+        mut shutdown_receiver: oneshot::Receiver<()>,
+    ) {
+        loop {
+            tokio::select! {
+                Some(msg) = write_receiver.recv() => {
+                    let read = ws_reader.next().await;
+                    if read.is_none() {
+                        let _ = msg.response_chn.send(Err(SignalError::WsError(WsError::ConnectionClosed)));
+                        break;
+                    }
+                    let read = read.unwrap();
+                    match read {
+                        Ok(Message::Binary(data)) => {
+                            let res = SignalResponse::decode(data.as_slice()).expect("failed to decode incoming SignalResponse");
+
+                            // TODO(theomonnon) Handle Message::Pong
+                            let res = res.message.unwrap();
+                            let _ = msg.response_chn.send(Ok(res));
+                        }
+                        _ => {
+                            error!("unhandled websocket message: {:?}", read);
+                            let _ = msg.response_chn.send(Err(SignalError::WsError(WsError::ConnectionClosed)));
+                            break;
+                        }
+                    }
+                },
+                _ = (&mut shutdown_receiver) => break
+            }
+        }
     }
 }
+
+#[tokio::test]
+async fn test_test() {
+    env_logger::init();
+    let client = connect("ws://localhost:7880", "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE2NzEyMzk4NjAsImlzcyI6IkFQSXpLYkFTaUNWYWtnSiIsIm5hbWUiOiJ0ZXN0IiwibmJmIjoxNjY0MDM5ODYwLCJzdWIiOiJ0ZXN0IiwidmlkZW8iOnsicm9vbUFkbWluIjp0cnVlLCJyb29tQ3JlYXRlIjp0cnVlLCJyb29tSm9pbiI6dHJ1ZX19.0Bee2jI2cSZveAbZ8MLc-ADoMYQ4l8IRxcAxpXAS6a8").await.unwrap();
+    let msg = client.recv().await.unwrap();
+
+    client.close().await;
+    info!("Received message {:?}", msg);
+}