From ceaa01e8b4aa0085788afcb2167da63f2b682a4a Mon Sep 17 00:00:00 2001 From: Greg Heartsfield Date: Sat, 12 Feb 2022 16:19:10 -0600 Subject: [PATCH] fix: removed manual nostr stream, so websocket pings work --- src/main.rs | 109 +++++++++++++++++++++++++++++------ src/protostream.rs | 141 --------------------------------------------- 2 files changed, 91 insertions(+), 159 deletions(-) delete mode 100644 src/protostream.rs diff --git a/src/main.rs b/src/main.rs index 66e1b0c..eee9b52 100644 --- a/src/main.rs +++ b/src/main.rs @@ -9,16 +9,17 @@ use hyper::{ }; use log::*; use nostr_rs_relay::close::Close; +use nostr_rs_relay::close::CloseCmd; use nostr_rs_relay::config; use nostr_rs_relay::conn; use nostr_rs_relay::db; use nostr_rs_relay::error::{Error, Result}; use nostr_rs_relay::event::Event; +use nostr_rs_relay::event::EventCmd; use nostr_rs_relay::info::RelayInfo; use nostr_rs_relay::nip05; -use nostr_rs_relay::protostream; -use nostr_rs_relay::protostream::NostrMessage::*; -use nostr_rs_relay::protostream::NostrResponse::*; +use nostr_rs_relay::subscription::Subscription; +use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::convert::Infallible; use std::env; @@ -29,7 +30,9 @@ use tokio::sync::broadcast::{self, Receiver, Sender}; use tokio::sync::mpsc; use tokio::sync::oneshot; use tokio_tungstenite::WebSocketStream; +use tungstenite::error::Error as WsError; use tungstenite::handshake; +use tungstenite::protocol::Message; use tungstenite::protocol::WebSocketConfig; /// Return a requested DB name from command line arguments. @@ -312,11 +315,57 @@ fn main() -> Result<(), Error> { Ok(()) } +/// Nostr protocol messages from a client +#[derive(Deserialize, Serialize, Clone, PartialEq, Debug)] +#[serde(untagged)] +pub enum NostrMessage { + /// An `EVENT` message + EventMsg(EventCmd), + /// A `REQ` message + SubMsg(Subscription), + /// A `CLOSE` message + CloseMsg(CloseCmd), +} + +/// Nostr protocol messages from a relay/server +#[derive(Deserialize, Serialize, Clone, PartialEq, Debug)] +pub enum NostrResponse { + /// A `NOTICE` response + NoticeRes(String), + /// An `EVENT` response, composed of the subscription identifier, + /// and serialized event JSON + EventRes(String, String), +} + +/// Convert Message to NostrMessage +fn convert_to_msg(msg: String) -> Result { + let config = config::SETTINGS.read().unwrap(); + let parsed_res: Result = serde_json::from_str(&msg).map_err(|e| e.into()); + match parsed_res { + Ok(m) => { + if let NostrMessage::EventMsg(_) = m { + if let Some(max_size) = config.limits.max_event_bytes { + // check length, ensure that some max size is set. + if msg.len() > max_size && max_size > 0 { + return Err(Error::EventMaxLengthError(msg.len())); + } + } + } + Ok(m) + } + Err(e) => { + debug!("proto parse error: {:?}", e); + debug!("parse error on message: {}", msg.trim()); + Err(Error::ProtoParseError) + } + } +} + /// Handle new client connections. This runs through an event loop /// for all client communication. async fn nostr_server( pool: db::SqlitePool, - ws_stream: WebSocketStream, + mut ws_stream: WebSocketStream, broadcast: Sender, event_tx: tokio::sync::mpsc::Sender, mut shutdown: Receiver<()>, @@ -327,7 +376,10 @@ async fn nostr_server( //let conn = tokio_tungstenite::accept_async_with_config(stream, Some(config)).await; //let ws_stream = conn.expect("websocket handshake error"); // wrap websocket into a stream & sink of Nostr protocol messages - let mut nostr_stream = protostream::wrap_ws_in_nostr(ws_stream); + + // don't wrap in a proto stream, because it broke pings. + //let mut nostr_stream = protostream::wrap_ws_in_nostr(ws_stream); + // Track internal client state let mut conn = conn::ClientConn::new(); let cid = conn.get_client_prefix(); @@ -351,9 +403,12 @@ async fn nostr_server( }, Some(query_result) = query_rx.recv() => { // database informed us of a query result we asked for - let res = EventRes(query_result.sub_id,query_result.event); + //let res = EventRes(query_result.sub_id,query_result.event); client_received_event_count += 1; - nostr_stream.send(res).await.ok(); + // send a result + let subesc = query_result.sub_id.replace("\"", ""); + let send_str = format!("[\"EVENT\",\"{}\",{}]", subesc, &query_result.event); + ws_stream.send(Message::Text(send_str)).await.ok(); }, // TODO: consider logging the LaggedRecv error Ok(global_event) = bcast_rx.recv() => { @@ -368,17 +423,34 @@ async fn nostr_server( cid, s, global_event.get_event_id_prefix()); // create an event response and send it - let res = EventRes(s.to_owned(),event_str); - nostr_stream.send(res).await.ok(); + let subesc = s.replace("\"", ""); + ws_stream.send(Message::Text(format!("[\"EVENT\",\"{}\",{}]", subesc, event_str))).await.ok(); + //nostr_stream.send(res).await.ok(); } else { warn!("could not serialize event {:?}", global_event.get_event_id_prefix()); } } }, // check if this client has a subscription - proto_next = nostr_stream.next() => { - match proto_next { - Some(Ok(EventMsg(ec))) => { + ws_next = ws_stream.next() => { + let protomsg = match ws_next { + Some(Ok(Message::Text(m))) => { + let msg_parse = convert_to_msg(m); + Some(msg_parse) + }, + None | Some(Ok(Message::Close(_))) | Some(Err(WsError::AlreadyClosed)) | Some(Err(WsError::ConnectionClosed)) => { + info!("Closing connection"); + None + }, + x => { + info!("message was: {:?} (ignoring)", x); + continue; + } + }; + + // convert ws_next into proto_next + match protomsg { + Some(Ok(NostrMessage::EventMsg(ec))) => { // An EventCmd needs to be validated to be converted into an Event // handle each type of message let parsed : Result = Result::::from(ec); @@ -398,11 +470,11 @@ async fn nostr_server( }, Err(_) => { info!("client {:?} sent an invalid event", cid); - nostr_stream.send(NoticeRes("event was invalid".to_owned())).await.ok(); + ws_stream.send(Message::Text(format!("[\"NOTICE\",\"{}\"]", "event was invalid"))).await.ok(); } } }, - Some(Ok(SubMsg(s))) => { + Some(Ok(NostrMessage::SubMsg(s))) => { debug!("client {} requesting a subscription", cid); // subscription handling consists of: // * registering the subscription so future events can be matched @@ -422,12 +494,12 @@ async fn nostr_server( }, Err(e) => { info!("Subscription error: {}", e); - nostr_stream.send(NoticeRes(format!("{}",e))).await.ok(); - + let s = e.to_string().replace("\"", ""); + ws_stream.send(Message::Text(format!("[\"NOTICE\",\"{}\"]", s))).await.ok(); } } }, - Some(Ok(CloseMsg(cc))) => { + Some(Ok(NostrMessage::CloseMsg(cc))) => { // closing a request simply removes the subscription. let parsed : Result = Result::::from(cc); match parsed { @@ -458,7 +530,8 @@ async fn nostr_server( } Some(Err(Error::EventMaxLengthError(s))) => { info!("client {:?} sent event larger ({} bytes) than max size", cid, s); - nostr_stream.send(NoticeRes("event exceeded max size".to_owned())).await.ok(); + //TODO + //nostr_stream.send(NoticeRes("event exceeded max size".to_owned())).await.ok(); }, Some(Err(e)) => { info!("got non-fatal error from client: {:?}, error: {:?}", cid, e); diff --git a/src/protostream.rs b/src/protostream.rs deleted file mode 100644 index 474662c..0000000 --- a/src/protostream.rs +++ /dev/null @@ -1,141 +0,0 @@ -//! Nostr protocol layered over WebSocket -use crate::close::CloseCmd; -use crate::config; -use crate::error::{Error, Result}; -use crate::event::EventCmd; -use crate::subscription::Subscription; -use core::pin::Pin; -use futures::sink::Sink; -use futures::stream::Stream; -use futures::task::Context; -use futures::task::Poll; -use hyper::upgrade::Upgraded; -use log::*; -use serde::{Deserialize, Serialize}; -use tokio_tungstenite::WebSocketStream; -use tungstenite::error::Error as WsError; -use tungstenite::protocol::Message; - -/// Nostr protocol messages from a client -#[derive(Deserialize, Serialize, Clone, PartialEq, Debug)] -#[serde(untagged)] -pub enum NostrMessage { - /// An `EVENT` message - EventMsg(EventCmd), - /// A `REQ` message - SubMsg(Subscription), - /// A `CLOSE` message - CloseMsg(CloseCmd), -} - -/// Nostr protocol messages from a relay/server -#[derive(Deserialize, Serialize, Clone, PartialEq, Debug)] -pub enum NostrResponse { - /// A `NOTICE` response - NoticeRes(String), - /// An `EVENT` response, composed of the subscription identifier, - /// and serialized event JSON - EventRes(String, String), -} - -/// A Nostr protocol stream is layered on top of a Websocket stream. -pub struct NostrStream { - ws_stream: WebSocketStream, -} - -/// Given a websocket, return a protocol stream wrapper. -pub fn wrap_ws_in_nostr(ws: WebSocketStream) -> NostrStream { - NostrStream { ws_stream: ws } -} - -/// Implement the [`Stream`] interface to produce Nostr messages. -impl Stream for NostrStream { - type Item = Result; - fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - // get the configuration - /// Convert Message to NostrMessage - fn convert(msg: String) -> Result { - let config = config::SETTINGS.read().unwrap(); - let parsed_res: Result = serde_json::from_str(&msg).map_err(|e| e.into()); - match parsed_res { - Ok(m) => { - if let NostrMessage::EventMsg(_) = m { - if let Some(max_size) = config.limits.max_event_bytes { - // check length, ensure that some max size is set. - if msg.len() > max_size && max_size > 0 { - return Err(Error::EventMaxLengthError(msg.len())); - } - } - } - Ok(m) - } - Err(e) => { - debug!("proto parse error: {:?}", e); - debug!("parse error on message: {}", msg.trim()); - Err(Error::ProtoParseError) - } - } - } - match Pin::new(&mut self.ws_stream).poll_next(cx) { - Poll::Pending => Poll::Pending, - Poll::Ready(None) => Poll::Ready(None), - Poll::Ready(Some(v)) => match v { - Ok(Message::Text(vs)) => Poll::Ready(Some(convert(vs))), - Ok(Message::Ping(x)) => { - debug!("client ping ({:?})", x); - //Pin::new(&mut self.ws_stream).start_send(Message::Pong(x)); - // TODO: restructure this so that Pongs work - //Pin::new(&mut self.ws_stream).write_pending(); - //info!("sent pong"); - Poll::Pending - } - Ok(Message::Binary(_)) => Poll::Ready(Some(Err(Error::ProtoParseError))), - Ok(Message::Pong(_)) => Poll::Pending, - Ok(Message::Close(_)) => Poll::Ready(None), - Err(WsError::AlreadyClosed) | Err(WsError::ConnectionClosed) => Poll::Ready(None), - Err(_) => Poll::Ready(Some(Err(Error::ConnError))), - }, - } - } -} - -/// Implement the [`Sink`] interface to produce Nostr responses. -impl Sink for NostrStream { - type Error = Error; - - fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - // map the error type - match Pin::new(&mut self.ws_stream).poll_ready(cx) { - Poll::Ready(Ok(())) => Poll::Ready(Ok(())), - Poll::Ready(Err(_)) => Poll::Ready(Err(Error::ConnWriteError)), - Poll::Pending => Poll::Pending, - } - } - - fn start_send(mut self: Pin<&mut Self>, item: NostrResponse) -> Result<(), Self::Error> { - // TODO: do real escaping for these - at least on NOTICE, - // which surely has some problems if arbitrary text is sent. - let send_str = match item { - NostrResponse::NoticeRes(msg) => { - let s = msg.replace("\"", ""); - format!("[\"NOTICE\",\"{}\"]", s) - } - NostrResponse::EventRes(sub, eventstr) => { - let subesc = sub.replace("\"", ""); - format!("[\"EVENT\",\"{}\",{}]", subesc, eventstr) - } - }; - match Pin::new(&mut self.ws_stream).start_send(Message::Text(send_str)) { - Ok(()) => Ok(()), - Err(_) => Err(Error::ConnWriteError), - } - } - - fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll> { - Poll::Ready(Ok(())) - } - - fn poll_close(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll> { - Poll::Ready(Ok(())) - } -}