Add http request api

Author: Ma233

.pre-commit-config.yaml

@@ -13,6 +13,18 @@ repos:
       - id: trailing-whitespace
   - repo: local
     hooks:
+      - id: buf-format
+        name: protobuf format
+        description: Format protobuf files with buf.
+        entry: bash -c 'buf format --write'
+        language: system
+        files: \.proto$
+      - id: buf-lint
+        name: protobuf lint
+        description: Lint protobuf files for errors.
+        entry: bash -c 'buf lint'
+        language: system
+        files: \.proto$
       - id: taplo
         name: taplo format
         description: Format Cargo.toml files with taplo.

buf.yaml

@@ -0,0 +1,4 @@
+version: v1
+lint:
+  except:
+    - PACKAGE_DIRECTORY_MATCH

build.rs

@@ -1,4 +1,4 @@
 fn main() -> Result<(), Box<dyn std::error::Error>> {
-    tonic_build::compile_protos("proto/command.proto")?;
+    tonic_build::compile_protos("proto/command_v1.proto")?;
     Ok(())
 }

proto/command.proto

@@ -0,0 +1,22 @@
+syntax = "proto3";
+package command.v1;
+
+message ConnectNodeRequest {
+  string address = 1;
+}
+
+message ConnectNodeResponse {}
+
+message RequestHttpServerRequest {
+  string address = 1;
+  bytes data = 2;
+}
+
+message RequestHttpServerResponse {
+  bytes data = 1;
+}
+
+service CommandService {
+  rpc ConnectNode(ConnectNodeRequest) returns (ConnectNodeResponse);
+  rpc RequestHttpServer(RequestHttpServerRequest) returns (RequestHttpServerResponse);
+}

proto/command_v1.proto

@@ -2,12 +2,11 @@ use tonic::Request;
 use tonic::Response;
 use tonic::Status;
 use tracing::trace;
-use tracing::warn;
 
 use crate::PProxyHandle;
 
 pub mod proto {
-    tonic::include_proto!("command");
+    tonic::include_proto!("command.v1");
 }
 
 pub struct PProxyCommander {
@@ -19,24 +18,36 @@ impl PProxyCommander {
         Self { handle }
     }
 
-    pub fn into_tonic_service(self) -> proto::command_server::CommandServer<Self> {
-        proto::command_server::CommandServer::new(self)
+    pub fn into_tonic_service(self) -> proto::command_service_server::CommandServiceServer<Self> {
+        proto::command_service_server::CommandServiceServer::new(self)
     }
 }
 
 #[tonic::async_trait]
-impl proto::command_server::Command for PProxyCommander {
+impl proto::command_service_server::CommandService for PProxyCommander {
     async fn connect_node(
         &self,
         request: Request<proto::ConnectNodeRequest>,
-    ) -> Result<Response<proto::ConnectNodeReply>, Status> {
+    ) -> Result<Response<proto::ConnectNodeResponse>, Status> {
         trace!("handle request: {:?}", request);
 
-        if let Err(e) = self.handle.connect_node(request.into_inner()).await {
-            warn!("failed to handle request: {:?}", e);
-        }
+        self.handle
+            .connect_node(request.into_inner())
+            .await
+            .map(Response::new)
+            .map_err(|e| tonic::Status::internal(format!("{:?}", e)))
+    }
+
+    async fn request_http_server(
+        &self,
+        request: tonic::Request<proto::RequestHttpServerRequest>,
+    ) -> std::result::Result<tonic::Response<proto::RequestHttpServerResponse>, tonic::Status> {
+        trace!("handle request: {:?}", request);
 
-        let reply = proto::ConnectNodeReply {};
-        Ok(Response::new(reply))
+        self.handle
+            .request_http_server(request.into_inner())
+            .await
+            .map(Response::new)
+            .map_err(|e| tonic::Status::internal(format!("{:?}", e)))
     }
 }

src/command.rs

@@ -13,3 +13,9 @@ impl<T> From<tokio::sync::mpsc::error::SendError<T>> for Error {
         Error::EssentialTaskClosed
     }
 }
+
+impl From<tokio::sync::oneshot::error::RecvError> for Error {
+    fn from(_: tokio::sync::oneshot::error::RecvError) -> Self {
+        Error::EssentialTaskClosed
+    }
+}

src/error.rs

@@ -1,13 +1,14 @@
 use std::net::SocketAddr;
 
-use tokio::sync::mpsc::channel;
-use tokio::sync::mpsc::Receiver;
-use tokio::sync::mpsc::Sender;
+use tokio::sync::mpsc;
+use tokio::sync::oneshot;
 use tracing::trace;
 use tracing::warn;
 
-use crate::command::proto::ConnectNodeReply;
 use crate::command::proto::ConnectNodeRequest;
+use crate::command::proto::ConnectNodeResponse;
+use crate::command::proto::RequestHttpServerRequest;
+use crate::command::proto::RequestHttpServerResponse;
 pub use crate::error::Error;
 use crate::server::*;
 
@@ -26,31 +27,35 @@ pub type Result<T> = std::result::Result<T, error::Error>;
 
 #[derive(Debug)]
 pub enum PProxyEvent {
-    ConnectNodeReply(ConnectNodeReply),
+    ConnectNodeResponse(ConnectNodeResponse),
 }
 
 #[derive(Debug)]
 pub enum PProxyCommand {
-    ConnectNodeRequest(ConnectNodeRequest),
+    ConnectNode(ConnectNodeRequest),
+    RequestHttpServer(
+        RequestHttpServerRequest,
+        oneshot::Sender<RequestHttpServerResponse>,
+    ),
 }
 
 pub struct PProxy {
     #[allow(dead_code)]
-    event_tx: Sender<PProxyEvent>,
-    command_rx: Receiver<PProxyCommand>,
+    event_tx: mpsc::Sender<PProxyEvent>,
+    command_rx: mpsc::Receiver<PProxyCommand>,
     p2p_server: P2pServer,
 }
 
 pub struct PProxyHandle {
     #[allow(dead_code)]
-    event_rx: Receiver<PProxyEvent>,
-    command_tx: Sender<PProxyCommand>,
+    event_rx: mpsc::Receiver<PProxyEvent>,
+    command_tx: mpsc::Sender<PProxyCommand>,
 }
 
 impl PProxy {
     pub fn new(server_addr: SocketAddr) -> (Self, PProxyHandle) {
-        let (event_tx, event_rx) = channel(DEFAULT_CHANNEL_SIZE);
-        let (command_tx, command_rx) = channel(DEFAULT_CHANNEL_SIZE);
+        let (event_tx, event_rx) = mpsc::channel(DEFAULT_CHANNEL_SIZE);
+        let (command_tx, command_rx) = mpsc::channel(DEFAULT_CHANNEL_SIZE);
 
         (
             Self {
@@ -82,11 +87,16 @@ impl PProxy {
                 command = self.command_rx.recv() => match command {
                     None => return,
                     Some(ref command) => match command {
-                        PProxyCommand::ConnectNodeRequest(request) => {
+                        PProxyCommand::ConnectNode(request) => {
                             if let Err(error) = self.on_connect_node(request).await {
                                 warn!("failed to handle command {:?}: {:?}", command, error);
                             }
                         }
+                        PProxyCommand::RequestHttpServer(request, _rx) => {
+                            if let Err(error) = self.on_request_http_server(request).await {
+                                warn!("failed to handle command {:?}: {:?}", command, error);
+                            }
+                        }
                     }
                 }
             }
@@ -109,13 +119,32 @@ impl PProxy {
             .await
             .map_err(From::from)
     }
+
+    async fn on_request_http_server(&mut self, _request: &RequestHttpServerRequest) -> Result<()> {
+        Ok(())
+    }
 }
 
 impl PProxyHandle {
-    pub async fn connect_node(&self, request: ConnectNodeRequest) -> Result<()> {
+    pub async fn connect_node(&self, request: ConnectNodeRequest) -> Result<ConnectNodeResponse> {
         self.command_tx
-            .send(PProxyCommand::ConnectNodeRequest(request))
-            .await
-            .map_err(From::from)
+            .send(PProxyCommand::ConnectNode(request))
+            .await?;
+
+        Ok(ConnectNodeResponse {})
+    }
+
+    pub async fn request_http_server(
+        &self,
+        request: RequestHttpServerRequest,
+    ) -> Result<RequestHttpServerResponse> {
+        let (tx, rx) = oneshot::channel();
+
+        self.command_tx
+            .send(PProxyCommand::RequestHttpServer(request, tx))
+            .await?;
+
+        let response = rx.await?;
+        Ok(response)
     }
 }

src/lib.rs

@@ -1,8 +1,8 @@
 use clap::Arg;
 use clap::ArgAction;
 use clap::Command;
-use pproxy::command::PProxyCommander;
-use pproxy::PProxy;
+use dephy_pproxy::command::PProxyCommander;
+use dephy_pproxy::PProxy;
 use tonic::transport::Server;
 
 fn parse_args() -> Command {
@@ -13,7 +13,6 @@ fn parse_args() -> Command {
     app = app
         .arg(
             Arg::new("SERVER_ADDR")
-                .short('s')
                 .long("server-addr")
                 .num_args(1)
                 .default_value("127.0.0.1:10010")
@@ -27,6 +26,13 @@ fn parse_args() -> Command {
                 .default_value("127.0.0.1:10086")
                 .action(ArgAction::Set)
                 .help("Commander server address"),
+        )
+        .arg(
+            Arg::new("PROXY_ADDR")
+                .long("proxy-addr")
+                .num_args(1)
+                .action(ArgAction::Set)
+                .help("Will proxy to this address if set"),
         );
 
     app