works modulo error handling

Cargo.lock

@@ -50,11 +50,14 @@ checksum = "130aac562c0dd69c56b3b1cc8ffd2e17be31d0b6c25b61c96b76231aa23e39e1"
 name = "c-lightning-http-plugin"
 version = "0.1.0"
 dependencies = [
+ "crossbeam-channel",
  "failure",
  "futures",
  "hyper",
  "lazy_async_pool",
  "lazy_static",
+ "serde",
+ "serde_json",
  "tokio",
 ]
 
@@ -546,6 +549,43 @@ version = "0.1.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4c691c0e608126e00913e33f0ccf3727d5fc84573623b8d65b2df340b5201783"
 
+[[package]]
+name = "ryu"
+version = "1.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bfa8506c1de11c9c4e4c38863ccbe02a305c8188e85a05a784c9e11e1c3910c8"
+
+[[package]]
+name = "serde"
+version = "1.0.104"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "414115f25f818d7dfccec8ee535d76949ae78584fc4f79a6f45a904bf8ab4449"
+dependencies = [
+ "serde_derive",
+]
+
+[[package]]
+name = "serde_derive"
+version = "1.0.104"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "128f9e303a5a29922045a830221b8f78ec74a5f544944f3d5984f8ec3895ef64"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "serde_json"
+version = "1.0.48"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9371ade75d4c2d6cb154141b9752cf3781ec9c05e0e5cf35060e1e70ee7b9c25"
+dependencies = [
+ "itoa",
+ "ryu",
+ "serde",
+]
+
 [[package]]
 name = "signal-hook-registry"
 version = "1.2.0"

Cargo.toml

@@ -7,9 +7,12 @@ edition = "2018"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-hyper = { version = "0.13.2", features = ["stream"] }
-tokio = { version = "0.2.13", features = ["full"] }
-lazy_async_pool = "0.3.0"
-futures = { version = "0.3.4", features = ["async-await"] }
+crossbeam-channel = "0.4.2"
 failure = "0.1.6"
-lazy_static = "1.4.0"
+futures = { version = "0.3.4", features = ["async-await"] }
+hyper = { version = "0.13.2", features = ["stream"] }
+lazy_async_pool = "0.3.0"
+lazy_static = "1.4.0"
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
+tokio = { version = "0.2.13", features = ["full"] }

src/lightning_socket.rs

@@ -0,0 +1,45 @@
+use std::path::PathBuf;
+use std::sync::Arc;
+
+use crossbeam_channel::Receiver;
+use futures::future::BoxFuture;
+use futures::future::FutureExt;
+use tokio::sync::RwLock;
+
+#[derive(Clone, Debug)]
+pub enum LightningSocketState {
+    Waiting(Receiver<PathBuf>),
+    Resolved(Arc<PathBuf>),
+}
+
+#[derive(Clone, Debug)]
+pub struct LightningSocketArc {
+    state: Arc<RwLock<LightningSocketState>>,
+}
+
+impl LightningSocketArc {
+    pub fn new(r: Receiver<PathBuf>) -> Self {
+        LightningSocketArc {
+            state: Arc::new(RwLock::new(LightningSocketState::Waiting(r))),
+        }
+    }
+    pub fn wait_for_path(self) -> BoxFuture<'static, Arc<PathBuf>> {
+        async move {
+            let guard = self.state.read().await;
+            match &*guard {
+                LightningSocketState::Resolved(ref path) => path.clone(),
+                LightningSocketState::Waiting(receiver) => match receiver.try_recv() {
+                    Ok(pb) => {
+                        let arc_pb = Arc::new(pb);
+                        drop(guard);
+                        let mut guard = self.state.write().await;
+                        *guard = LightningSocketState::Resolved(arc_pb.clone());
+                        arc_pb
+                    }
+                    Err(_) => self.clone().wait_for_path().await,
+                },
+            }
+        }
+        .boxed()
+    }
+}

src/main.rs

@@ -17,6 +17,15 @@ use tokio::io::AsyncRead;
 use tokio::io::Result as TokioResult;
 use tokio::net::UnixStream;
 
+use crate::lightning_socket::LightningSocketArc;
+
+mod lightning_socket;
+mod rpc;
+
+// TODO: implement init
+// TODO: implement getmanifest
+//
+
 type BoxedByteStream = Box<
     dyn futures::Stream<Item = Result<Bytes, Box<dyn std::error::Error + 'static + Sync + Send>>>
         + 'static
@@ -144,7 +153,16 @@ async fn main() {
     // Construct our SocketAddr to listen on...
     let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
 
-    let pool = Pool::new(0, || UnixStream::connect("TODO").boxed());
+    let (send_side, recv_side) = crossbeam_channel::bounded(1);
+
+    // fork thread for stdio
+    let lightning_socket_fut = LightningSocketArc::new(recv_side);
+
+    let pool = Pool::new(0, move || {
+        let lightning_socket_fut = lightning_socket_fut.clone();
+        async move { UnixStream::connect(&*lightning_socket_fut.wait_for_path().await).await }
+            .boxed()
+    });
     // And a MakeService to handle each connection...
     let handler = move |req| handle((&pool).clone(), req);
     let make_service = make_service_fn(|_conn| {
@@ -155,6 +173,10 @@ async fn main() {
     // Then bind and serve...
     let server = Server::bind(&addr).serve(make_service);
 
+    std::thread::spawn(move || {
+        crate::rpc::handle_stdio_rpc(send_side);
+    });
+
     // And run forever...
     if let Err(e) = server.await {
         eprintln!("server error: {}", e);

src/manifest.json

@@ -0,0 +1 @@
+{}

src/rpc.rs

@@ -0,0 +1,188 @@
+use std::path::PathBuf;
+
+use crossbeam_channel::Sender;
+use serde_json::StreamDeserializer;
+use serde_json::Value;
+
+fn deserialize_some<'de, T, D>(deserializer: D) -> Result<Option<T>, D::Error>
+where
+    T: serde::Deserialize<'de>,
+    D: serde::Deserializer<'de>,
+{
+    serde::Deserialize::deserialize(deserializer).map(Some)
+}
+
+#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
+#[serde(untagged)]
+pub enum JsonRpcV2Id {
+    Num(serde_json::Number),
+    Str(String),
+    Null,
+}
+
+#[derive(Clone, Debug)]
+pub struct JsonRpcV2;
+impl Default for JsonRpcV2 {
+    fn default() -> Self {
+        JsonRpcV2
+    }
+}
+impl serde::Serialize for JsonRpcV2 {
+    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+        serializer.serialize_str("2.0")
+    }
+}
+impl<'de> serde::Deserialize<'de> for JsonRpcV2 {
+    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
+        let version: String = serde::Deserialize::deserialize(deserializer)?;
+        match version.as_str() {
+            "2.0" => (),
+            a => {
+                return Err(serde::de::Error::custom(format!(
+                    "invalid RPC version: {}",
+                    a
+                )))
+            }
+        }
+        Ok(JsonRpcV2)
+    }
+}
+
+#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
+pub struct RpcReq {
+    #[serde(default, deserialize_with = "deserialize_some")]
+    pub id: Option<JsonRpcV2Id>,
+    #[serde(default)]
+    pub jsonrpc: JsonRpcV2,
+    pub method: String,
+    pub params: Vec<Value>,
+}
+impl AsRef<RpcReq> for RpcReq {
+    fn as_ref(&self) -> &RpcReq {
+        &self
+    }
+}
+
+#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
+pub struct RpcRes {
+    pub id: JsonRpcV2Id,
+    pub jsonrpc: JsonRpcV2,
+    #[serde(flatten)]
+    pub result: RpcResult,
+}
+
+#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "lowercase")]
+pub enum RpcResult {
+    Result(Value),
+    Error(RpcError),
+}
+impl RpcResult {
+    pub fn res(self) -> Result<Value, RpcError> {
+        self.into()
+    }
+}
+impl From<RpcResult> for Result<Value, RpcError> {
+    fn from(r: RpcResult) -> Self {
+        match r {
+            RpcResult::Result(a) => Ok(a),
+            RpcResult::Error(e) => Err(e),
+        }
+    }
+}
+
+#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
+pub struct RpcError {
+    pub code: serde_json::Number,
+    pub message: String,
+    #[serde(
+        default,
+        deserialize_with = "deserialize_some",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub data: Option<Value>,
+}
+
+pub fn handle_stdio_rpc(send_side: Sender<PathBuf>) {
+    // create serde stream
+    let req_stream: StreamDeserializer<_, RpcReq> =
+        StreamDeserializer::new(serde_json::de::IoRead::new(std::io::stdin()));
+    // for request in stream
+    for e_req in req_stream {
+        let rpc_result = match e_req {
+            Ok(RpcReq {
+                id: Some(req_id),
+                jsonrpc: _,
+                method: method,
+                params: params,
+            }) => RpcRes {
+                id: req_id,
+                jsonrpc: Default::default(),
+                result: match &*method {
+                    "init" => match init(send_side.clone(), params) {
+                        Ok(_) => RpcResult::Result(serde_json::json!({})),
+                        Err(e) => RpcResult::Error(RpcError {
+                            code: todo!(),
+                            message: todo!(),
+                            data: Some(Value::String(format!("{}", e))),
+                        }),
+                    },
+                    "getmanifest" => {
+                        RpcResult::Result(serde_json::json!(include!("manifest.json")))
+                    }
+                    other => RpcResult::Error(RpcError {
+                        code: todo!(),
+                        message: todo!(),
+                        data: Some(Value::String(format!("{}", other.to_owned()))),
+                    }),
+                },
+            },
+            Ok(RpcReq {
+                id: None,
+                jsonrpc: _,
+                method: method,
+                params: params,
+            }) => {
+                continue;
+            }
+            Err(e) => RpcRes {
+                id: JsonRpcV2Id::Null,
+                jsonrpc: JsonRpcV2,
+                result: RpcResult::Error(RpcError {
+                    code: todo!(),
+                    message: todo!(),
+                    data: Some(Value::String(format!("{}", e))),
+                }),
+            },
+        };
+        serde_json::to_writer(std::io::stdout(), &rpc_result);
+        print!("\n\n");
+    }
+}
+
+fn init(send_side: Sender<PathBuf>, mut conf: Vec<Value>) -> Result<(), failure::Error> {
+    let arg0 = conf
+        .pop()
+        .ok_or(failure::format_err!("No arguments supplied"))?;
+    let conf: LightningConfig = serde_json::from_value(arg0)?;
+    send_side.send(conf.lightning_dir.join(conf.rpc_file));
+    Ok(())
+}
+
+#[derive(Clone, Debug)]
+pub struct LightningConfig {
+    lightning_dir: PathBuf,
+    rpc_file: String,
+    startup: bool,
+}
+
+impl<'de> serde::Deserialize<'de> for LightningConfig {
+    fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
+        #[derive(serde::Deserialize)]
+        struct Complete {
+            configuration: LightningConfig,
+        }
+        let complete = Complete::deserialize(d)?;
+        Ok(complete.configuration)
+    }
+}