Merge pull request #982 from twmsh/master

update bytes(protobuf types) mapping, from Vec<u8> to bytes:Byte

Cargo.toml

@@ -55,6 +55,7 @@ rpassword = "7.0"
 base64 = "0.13"
 sysinfo = "0.24"
 num_cpus = "1.13"
+bytes = { version = "1.2", features = ["serde"] }
 default-net = "0.11.0"
 wol-rs = "0.9.1"
 

libs/hbb_common/Cargo.toml

@@ -11,7 +11,7 @@ protobuf = { version = "3.1", features = ["with-bytes"] }
 tokio = { version = "1.20", features = ["full"] }
 tokio-util = { version = "0.7", features = ["full"] }
 futures = "0.3"
-bytes = "1.1"
+bytes = { version = "1.2", features = ["serde"] }
 log = "0.4"
 env_logger = "0.9"
 socket2 = { version = "0.3", features = ["reuseport"] }

libs/hbb_common/build.rs

@@ -5,6 +5,10 @@ fn main() {
         .out_dir("src/protos")
         .inputs(&["protos/rendezvous.proto", "protos/message.proto"])
         .include("protos")
+        .customize(
+            protobuf_codegen::Customize::default()
+            .tokio_bytes(true)
+        )
         .run()
         .expect("Codegen failed.");
 }

libs/scrap/src/common/vpxcodec.rs

@@ -12,6 +12,7 @@ use crate::STRIDE_ALIGN;
 use super::vpx::{vp8e_enc_control_id::*, vpx_codec_err_t::*, *};
 use std::os::raw::{c_int, c_uint};
 use std::{ptr, slice};
+use hbb_common::bytes::Bytes;
 
 #[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
 pub enum VpxVideoCodecId {
@@ -291,7 +292,7 @@ impl VpxEncoder {
     #[inline]
     fn create_frame(frame: &EncodeFrame) -> EncodedVideoFrame {
         EncodedVideoFrame {
-            data: frame.data.to_vec(),
+            data: Bytes::from(frame.data.to_vec()),
             key: frame.key,
             pts: frame.pts,
             ..Default::default()

src/client.rs

@@ -216,7 +216,7 @@ impl Client {
                             } else {
                                 peer_nat_type = ph.nat_type();
                                 is_local = ph.is_local();
-                                signed_id_pk = ph.pk;
+                                signed_id_pk = ph.pk.into();
                                 relay_server = ph.relay_server;
                                 peer_addr = AddrMangle::decode(&ph.socket_addr);
                                 log::info!("Hole Punched {} = {}", peer, peer_addr);
@@ -408,8 +408,8 @@ impl Client {
                                 let sealed_key = box_::seal(&key.0, &nonce, &their_pk_b, &out_sk_b);
                                 let mut msg_out = Message::new();
                                 msg_out.set_public_key(PublicKey {
-                                    asymmetric_value: our_pk_b.0.into(),
-                                    symmetric_value: sealed_key,
+                                    asymmetric_value: Vec::from(our_pk_b.0).into(),
+                                    symmetric_value: sealed_key.into(),
                                     ..Default::default()
                                 });
                                 timeout(CONNECT_TIMEOUT, conn.send(&msg_out)).await??;
@@ -1139,7 +1139,7 @@ impl LoginConfigHandler {
         let my_id = Config::get_id();
         let mut lr = LoginRequest {
             username: self.id.clone(),
-            password,
+            password:password.into(),
             my_id,
             my_name: crate::username(),
             option: self.get_option_message(true).into(),

src/clipboard_file.rs

@@ -67,7 +67,7 @@ pub fn clip_2_msg(clip: ClipbaordFile) -> Message {
                     CliprdrServerFormatDataResponse {
                         conn_id,
                         msg_flags,
-                        format_data,
+                        format_data: format_data.into(),
                         ..Default::default()
                     },
                 )),
@@ -117,7 +117,7 @@ pub fn clip_2_msg(clip: ClipbaordFile) -> Message {
                         conn_id,
                         msg_flags,
                         stream_id,
-                        requested_data,
+                        requested_data: requested_data.into(),
                         ..Default::default()
                     },
                 )),
@@ -156,7 +156,7 @@ pub fn msg_2_clip(msg: Cliprdr) -> Option<ClipbaordFile> {
             Some(ClipbaordFile::ServerFormatDataResponse {
                 conn_id: data.conn_id,
                 msg_flags: data.msg_flags,
-                format_data: data.format_data,
+                format_data: data.format_data.into(),
             })
         }
         Some(cliprdr::Union::FileContentsRequest(data)) => {
@@ -177,7 +177,7 @@ pub fn msg_2_clip(msg: Cliprdr) -> Option<ClipbaordFile> {
                 conn_id: data.conn_id,
                 msg_flags: data.msg_flags,
                 stream_id: data.stream_id,
-                requested_data: data.requested_data,
+                requested_data: data.requested_data.into(),
             })
         }
         _ => None,

src/common.rs

@@ -48,7 +48,7 @@ pub fn create_clipboard_msg(content: String) -> Message {
     let mut msg = Message::new();
     msg.set_clipboard(Clipboard {
         compress,
-        content,
+        content:content.into(),
         ..Default::default()
     });
     msg
@@ -79,7 +79,7 @@ pub fn update_clipboard(clipboard: Clipboard, old: Option<&Arc<Mutex<String>>>)
     let content = if clipboard.compress {
         decompress(&clipboard.content)
     } else {
-        clipboard.content
+        clipboard.content.into()
     };
     if let Ok(content) = String::from_utf8(content) {
         if content.is_empty() {

src/ipc.rs

@@ -19,6 +19,7 @@ use parity_tokio_ipc::{
 };
 use serde_derive::{Deserialize, Serialize};
 use std::{collections::HashMap, sync::atomic::Ordering};
+use bytes::Bytes;
 #[cfg(not(windows))]
 use std::{fs::File, io::prelude::*};
 
@@ -75,7 +76,7 @@ pub enum FS {
     WriteBlock {
         id: i32,
         file_num: i32,
-        data: Vec<u8>,
+        data: Bytes,
         compressed: bool,
     },
     WriteDone {
@@ -562,8 +563,8 @@ where
         }
     }
 
-    pub async fn send_raw(&mut self, data: Vec<u8>) -> ResultType<()> {
-        self.inner.send(bytes::Bytes::from(data)).await?;
+    pub async fn send_raw(&mut self, data: Bytes) -> ResultType<()> {
+        self.inner.send(data).await?;
         Ok(())
     }
 

src/platform/linux.rs

@@ -111,7 +111,8 @@ pub fn get_cursor_data(hcursor: u64) -> ResultType<CursorData> {
                         cd.id = (*img).cursor_serial as _;
                         let pixels =
                             std::slice::from_raw_parts((*img).pixels, (cd.width * cd.height) as _);
-                        cd.colors.resize(pixels.len() * 4, 0);
+                        // cd.colors.resize(pixels.len() * 4, 0);
+                        let mut cd_colors = vec![0_u8; pixels.len() * 4];
                         for y in 0..cd.height {
                             for x in 0..cd.width {
                                 let pos = (y * cd.width + x) as usize;
@@ -124,12 +125,13 @@ pub fn get_cursor_data(hcursor: u64) -> ResultType<CursorData> {
                                     continue;
                                 }
                                 let pos = pos * 4;
-                                cd.colors[pos] = r as _;
-                                cd.colors[pos + 1] = g as _;
-                                cd.colors[pos + 2] = b as _;
-                                cd.colors[pos + 3] = a as _;
+                                cd_colors[pos] = r as _;
+                                cd_colors[pos + 1] = g as _;
+                                cd_colors[pos + 2] = b as _;
+                                cd_colors[pos + 3] = a as _;
                             }
                         }
+                        cd.colors = cd_colors.into();
                         res = Some(cd);
                     }
                     if !img.is_null() {

src/platform/macos.rs

@@ -342,7 +342,7 @@ pub fn get_cursor_data(hcursor: u64) -> ResultType<CursorData> {
         }
         Ok(CursorData {
             id: hcursor,
-            colors,
+            colors: colors.into(),
             hotx: hotspot.x as _,
             hoty: hotspot.y as _,
             width: size.width as _,

src/platform/windows.rs

@@ -164,7 +164,7 @@ pub fn get_cursor_data(hcursor: u64) -> ResultType<CursorData> {
 
         Ok(CursorData {
             id: hcursor,
-            colors: cbits,
+            colors: cbits.into(),
             hotx: ii.0.xHotspot as _,
             hoty: ii.0.yHotspot as _,
             width: width as _,

src/rendezvous_mediator.rs

@@ -266,7 +266,7 @@ impl RendezvousMediator {
 
     async fn handle_request_relay(&self, rr: RequestRelay, server: ServerPtr) -> ResultType<()> {
         self.create_relay(
-            rr.socket_addr,
+            rr.socket_addr.into(),
             rr.relay_server,
             rr.uuid,
             server,
@@ -303,7 +303,7 @@ impl RendezvousMediator {
 
         let mut msg_out = Message::new();
         let mut rr = RelayResponse {
-            socket_addr,
+            socket_addr: socket_addr.into(),
             version: crate::VERSION.to_owned(),
             ..Default::default()
         };
@@ -334,8 +334,8 @@ impl RendezvousMediator {
         let relay_server = self.get_relay_server(fla.relay_server);
         msg_out.set_local_addr(LocalAddr {
             id: Config::get_id(),
-            socket_addr: AddrMangle::encode(peer_addr),
-            local_addr: AddrMangle::encode(local_addr),
+            socket_addr: AddrMangle::encode(peer_addr).into(),
+            local_addr: AddrMangle::encode(local_addr).into(),
             relay_server,
             version: crate::VERSION.to_owned(),
             ..Default::default()
@@ -353,7 +353,7 @@ impl RendezvousMediator {
         {
             let uuid = Uuid::new_v4().to_string();
             return self
-                .create_relay(ph.socket_addr, relay_server, uuid, server, true, true)
+                .create_relay(ph.socket_addr.into(), relay_server, uuid, server, true, true)
                 .await;
         }
         let peer_addr = AddrMangle::decode(&ph.socket_addr);
@@ -394,8 +394,8 @@ impl RendezvousMediator {
         self.last_id_pk_registry = id.clone();
         msg_out.set_register_pk(RegisterPk {
             id,
-            uuid,
-            pk,
+            uuid: uuid.into(),
+            pk: pk.into(),
             ..Default::default()
         });
         socket.send(&msg_out, self.addr.to_owned()).await?;

src/server.rs

@@ -20,6 +20,8 @@ use std::{
     sync::{Arc, Mutex, RwLock, Weak},
     time::Duration,
 };
+use bytes::Bytes;
+
 pub mod audio_service;
 cfg_if::cfg_if! {
 if #[cfg(not(any(target_os = "android", target_os = "ios")))] {
@@ -130,13 +132,13 @@ pub async fn create_tcp_connection(
             id: sign::sign(
                 &IdPk {
                     id: Config::get_id(),
-                    pk: our_pk_b.0.to_vec(),
+                    pk: Bytes::from(our_pk_b.0.to_vec()),
                     ..Default::default()
                 }
                 .write_to_bytes()
                 .unwrap_or_default(),
                 &sk,
-            ),
+            ).into(),
             ..Default::default()
         });
         timeout(CONNECT_TIMEOUT, stream.send(&msg_out)).await??;

src/server/audio_service.rs

@@ -367,7 +367,7 @@ fn send_f32(data: &[f32], encoder: &mut Encoder, sp: &GenericService) {
         Ok(data) => {
             let mut msg_out = Message::new();
             msg_out.set_audio_frame(AudioFrame {
-                data,
+                data: data.into(),
                 timestamp: crate::common::get_time(),
                 ..Default::default()
             });

src/server/connection.rs

@@ -1459,7 +1459,7 @@ async fn start_ipc(
                             file_num,
                             data,
                             compressed}) = data {
-                                stream.send(&Data::FS(ipc::FS::WriteBlock{id, file_num, data: Vec::new(), compressed})).await?;
+                                stream.send(&Data::FS(ipc::FS::WriteBlock{id, file_num, data: Bytes::new(), compressed})).await?;
                                 stream.send_raw(data).await?;
                         } else {
                             stream.send(&data).await?;

src/server/input_service.rs

@@ -145,7 +145,7 @@ fn run_cursor(sp: MouseCursorService, state: &mut StateCursor) -> ResultType<()>
                 msg = cached.clone();
             } else {
                 let mut data = crate::get_cursor_data(hcursor)?;
-                data.colors = hbb_common::compress::compress(&data.colors[..], COMPRESS_LEVEL);
+                data.colors = hbb_common::compress::compress(&data.colors[..], COMPRESS_LEVEL).into();
                 let mut tmp = Message::new();
                 tmp.set_cursor_data(data);
                 msg = Arc::new(tmp);

src/ui/cm.rs

@@ -590,7 +590,7 @@ async fn start_pa() {
                                             } else {
                                                 buf.clone()
                                             };
-                                        if let Err(err) = stream.send_raw(out).await {
+                                        if let Err(err) = stream.send_raw(out.into()).await {
                                             log::error!("Failed to send audio data:{}", err);
                                             break;
                                         }