refact: replace request response protocol with stream protocol

Cargo.toml

@@ -1,20 +1,22 @@
 [package]
 name = "dephy-pproxy"
-version = "0.3.2"
+version = "0.4.0"
 edition = "2021"
 
 [dependencies]
 async-trait = "0.1.81"
+bytes = "1.7.2"
 clap = "4.5.4"
 futures = "0.3.30"
 hex = "0.4.3"
-libp2p = { version = "0.54.1", features = ["tokio", "macros", "yamux", "noise", "tcp", "request-response", "relay", "secp256k1"] }
+libp2p = { version = "0.54.1", features = ["tokio", "macros", "yamux", "noise", "tcp", "relay", "secp256k1"] }
+libp2p-stream = "0.2.0-alpha"
 prost = "0.13.1"
 reqwest = { version = "0.12.5", features = ["json", "rustls-tls"], default-features = false }
 serde = { version = "1.0.207", features = ["derive"] }
 thiserror = "1.0.60"
-tokio = { version = "1.37.0", features = ["rt-multi-thread"] }
-tokio-util = "0.7.11"
+tokio = { version = "1.37.0", features = ["io-util", "rt-multi-thread"] }
+tokio-util = { version = "0.7.11", features = ["compat"] }
 tonic = "0.12.1"
 tonic-web = "0.12.1"
 tracing = "0.1.40"
@@ -26,6 +28,3 @@ tonic-build = "0.12.1"
 [[bin]]
 name = "pproxy"
 path = "src/main.rs"
-
-[dev-dependencies]
-futures_ringbuf = "0.4.0"

build.rs

@@ -1,5 +1,5 @@
 fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let protos = ["proto/command_v1.proto", "proto/tunnel_v1.proto"];
+    let protos = ["proto/command_v1.proto"];
     tonic_build::configure()
         .protoc_arg("--experimental_allow_proto3_optional")
         .compile(&protos, &["proto"])?;

src/error.rs

@@ -1,8 +1,12 @@
 use std::io::ErrorKind as IOErrorKind;
 
+use crate::types::TunnelReply;
+
 #[derive(Debug, thiserror::Error)]
 #[non_exhaustive]
 pub enum Error {
+    #[error("Io error: {0}")]
+    IoError(#[from] std::io::Error),
     #[error("Multiaddr parse error: {0}")]
     MultiaddrParseError(String),
     #[error("SocketAddr parse error: {0}")]
@@ -17,8 +21,12 @@ pub enum Error {
     EssentialTaskClosed,
     #[error("Libp2p swarm create error: {0}")]
     Libp2pSwarmCreateError(String),
+    #[error("Libp2p dial error: {0}")]
+    Libp2pDialError(#[from] libp2p::swarm::DialError),
     #[error("Libp2p transport error: {0}")]
     Libp2pTransportError(#[from] libp2p::core::transport::TransportError<std::io::Error>),
+    #[error("Libp2p open stream error")]
+    Libp2pOpenStreamError(#[from] libp2p_stream::OpenStreamError),
     #[error("Reqwest error: {0}")]
     ReqwestError(#[from] reqwest::Error),
     #[error("Protocol not support: {0}")]
@@ -32,7 +40,9 @@ pub enum Error {
     #[error("Tunnel context not found: {0}")]
     TunnelContextNotFound(String),
     #[error("Tunnel error: {0:?}")]
-    Tunnel(TunnelError),
+    TunnelError(TunnelError),
+    #[error("Tunnel protocol error: {0}")]
+    TunnelProtocolError(#[from] TunnelProtocolError),
     #[error("Protobuf decode error: {0}")]
     ProtobufDecode(#[from] prost::DecodeError),
     #[error("Access denied, peer: {0}")]
@@ -67,6 +77,33 @@ pub enum TunnelError {
     Unknown = u8::MAX,
 }
 
+#[derive(Debug, thiserror::Error)]
+pub enum TunnelProtocolError {
+    #[error("{0}")]
+    IoError(#[from] std::io::Error),
+    #[error("unsupported tunnel version {0:#x}")]
+    UnsupportedTunnelVersion(u8),
+    #[error("unsupported command {0:#x}")]
+    UnsupportedCommand(u8),
+    #[error("{0}")]
+    TunnelReply(TunnelReply),
+}
+
+impl TunnelProtocolError {
+    /// Convert to `TunnelReply` for responding
+    pub fn as_reply(&self) -> TunnelReply {
+        match *self {
+            TunnelProtocolError::IoError(ref err) => match err.kind() {
+                std::io::ErrorKind::ConnectionRefused => TunnelReply::ConnectionRefused,
+                _ => TunnelReply::GeneralFailure,
+            },
+            TunnelProtocolError::UnsupportedTunnelVersion(..) => TunnelReply::GeneralFailure,
+            TunnelProtocolError::UnsupportedCommand(..) => TunnelReply::CommandNotSupported,
+            TunnelProtocolError::TunnelReply(r) => r,
+        }
+    }
+}
+
 impl<T> From<tokio::sync::mpsc::error::SendError<T>> for Error {
     fn from(_: tokio::sync::mpsc::error::SendError<T>) -> Self {
         Error::EssentialTaskClosed
@@ -81,7 +118,7 @@ impl From<futures::channel::oneshot::Canceled> for Error {
 
 impl From<TunnelError> for Error {
     fn from(error: TunnelError) -> Self {
-        Error::Tunnel(error)
+        Error::TunnelError(error)
     }
 }