You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: client.rs
    path of file: ./repos/ownserver/ownserver_server/src
    
    the code of the file until where you have to start completion: use std::sync::Arc;

use bytes::BytesMut;
use futures::{stream::SplitSink, StreamExt, SinkExt};
use ownserver_lib::{ClientId, Endpoints, ControlPacketV2Codec, ControlPacketV2};
use tokio_util::{sync::CancellationToken, codec::{Encoder, Decoder}};
use tracing::Instrument;
use warp::ws::{Message, WebSocket};

use crate::{Store, remote::stream::StreamMessage, ClientStreamError};


#[derive(Debug)]
pub struct Client {
    pub client_id: ClientId,
    endpoints: Endpoints,

    ws_tx: SplitSi
    
    pub fn new(store: Arc<Store>, client_id: ClientId, endpoints: Endpoints, ws: WebSocket) -> Self {
        let (sink, mut stream) = ws.split();
        let token = CancellationToken::new();

        let ct = token.clone();
        let store_ = store.clone();
        tokio::spawn(async move {
            loop {
                tokio::select! {
                    _ = ct.cancelled() => {
                        break
                    }
                    result = stream.next() => {
                        let message = match result {
                            // handle protocol message
                            Some(Ok(msg)) if (msg.is_binary() || msg.is_text()) && !msg.as_bytes().is_empty() => {
                                msg.into_bytes()
                            }
                            // handle close with reason
                            Some(Ok(msg)) if msg.is_close() && !msg.as_bytes().is_empty() => {
                                tracing::info!(cid = %client_id, close_reason=?msg, "client got close");
                                break
                            }
                            _ => {
                                tracing::info!(cid = %client_id, "goodbye client");
                                break
                            }
                        };
                
                        let mut bytes = BytesMut::from(&message[..]);
                        let packet = match ControlPacketV2Codec::new().decode(&mut bytes) {
                            Ok(Some(packet)) => packet,
                            Ok(None) => {
                                // TODO: should handle None case
                                tracing::warn!(cid = %client_id, "failed to parse partial client message");
                                continue;
                            }
                            Err(e) => {
                                tracing::warn!(cid = %client_id, error = ?e, "failed to parse client message");
                                continue;
                            }
                        };


                
                        tracing::trace!(cid = %client_id, ?packet, "got control packet from client");

                        let (stream_id, message) = match packet {
                            ControlPacketV2::Data(stream_id, data) => {
                                tracing::trace!(cid = %client_id, sid = %stream_id, "forwarding to stream: {}", data.len());
                                (stream_id, StreamMessage::Data(data))
                            }
                            ControlPacketV2::Refused(stream_id) => {
                                tracing::debug!(cid = %client_id, sid = %stream_id, "tunnel says: refused");
                                (stream_id, StreamMessage::TunnelRefused)
                            }
                            ControlPacketV2::Ping => {
                                tracing::trace!(cid = %client_id, "pong");
                                continue;
                            }
                            ControlPacketV2::Init(stream_id, endpoint_id) => {
                                tracing::error!(cid = %client_id, sid = %stream_id, eid = %endpoint_id, "invalid protocol ControlPacketV2::Init");
                                continue;
                            }
                            ControlPacketV2::End(stream_id) => {
                                tracing::error!(cid = %client_id, sid = %stream_id, "invalid protocol ControlPacketV2::End");
                                continue;
                            }
                        };

                        tracing::trace!(cid = %client_id, sid = %stream_id, "forward message to remote stream");
                        if let Err(e) = store_.send_to_remote(stream_id, message).await {
                            tracing::debug!(cid = %client_id, sid = %stream_id, error = ?e, "Failed to send to remote stream");

                            store_.disable_remote(stream_id).await;
                        }
                    }
                }
            }
            store_.disable_client(client_id).await;
        }.instrument(tracing::info_span!("client_read_loop")));

        Self { client_id, endpoints, ws_tx: sink, store, ct: token, disabled: false }
    }