From 83ba0490c0dcdae4b7db3dc03302e07c552f32fd Mon Sep 17 00:00:00 2001 From: Keagan McClelland Date: Thu, 5 Mar 2020 18:11:57 -0700 Subject: [PATCH] works modulo error handling --- Cargo.lock | 40 +++++++++ Cargo.toml | 13 +-- src/lightning_socket.rs | 45 ++++++++++ src/main.rs | 24 ++++- src/manifest.json | 1 + src/rpc.rs | 188 ++++++++++++++++++++++++++++++++++++++++ 6 files changed, 305 insertions(+), 6 deletions(-) create mode 100644 src/lightning_socket.rs create mode 100644 src/manifest.json create mode 100644 src/rpc.rs diff --git a/Cargo.lock b/Cargo.lock index ec1ea88..abd2ae0 100644 --- a/Cargo.lock +++ b/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" diff --git a/Cargo.toml b/Cargo.toml index 514bf47..6477dbb 100644 --- a/Cargo.toml +++ b/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" \ No newline at end of file +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"] } \ No newline at end of file diff --git a/src/lightning_socket.rs b/src/lightning_socket.rs new file mode 100644 index 0000000..1d4e11d --- /dev/null +++ b/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), + Resolved(Arc), +} + +#[derive(Clone, Debug)] +pub struct LightningSocketArc { + state: Arc>, +} + +impl LightningSocketArc { + pub fn new(r: Receiver) -> Self { + LightningSocketArc { + state: Arc::new(RwLock::new(LightningSocketState::Waiting(r))), + } + } + pub fn wait_for_path(self) -> BoxFuture<'static, Arc> { + 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() + } +} diff --git a/src/main.rs b/src/main.rs index d3cb2c2..556f28b 100644 --- a/src/main.rs +++ b/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>> + '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); diff --git a/src/manifest.json b/src/manifest.json new file mode 100644 index 0000000..9e26dfe --- /dev/null +++ b/src/manifest.json @@ -0,0 +1 @@ +{} \ No newline at end of file diff --git a/src/rpc.rs b/src/rpc.rs new file mode 100644 index 0000000..721cc84 --- /dev/null +++ b/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, 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(&self, serializer: S) -> Result { + serializer.serialize_str("2.0") + } +} +impl<'de> serde::Deserialize<'de> for JsonRpcV2 { + fn deserialize>(deserializer: D) -> Result { + 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, + #[serde(default)] + pub jsonrpc: JsonRpcV2, + pub method: String, + pub params: Vec, +} +impl AsRef 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 { + self.into() + } +} +impl From for Result { + 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, +} + +pub fn handle_stdio_rpc(send_side: Sender) { + // 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, mut conf: Vec) -> 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: D) -> Result { + #[derive(serde::Deserialize)] + struct Complete { + configuration: LightningConfig, + } + let complete = Complete::deserialize(d)?; + Ok(complete.configuration) + } +}