fix: removed manual nostr stream, so websocket pings work

This commit is contained in:
Greg Heartsfield 2022-02-12 16:19:10 -06:00
parent bc68cd0c74
commit ceaa01e8b4
2 changed files with 91 additions and 159 deletions

View File

@ -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<NostrMessage> {
let config = config::SETTINGS.read().unwrap();
let parsed_res: Result<NostrMessage> = 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<Upgraded>,
mut ws_stream: WebSocketStream<Upgraded>,
broadcast: Sender<Event>,
event_tx: tokio::sync::mpsc::Sender<Event>,
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<Event> = Result::<Event>::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<Close> = Result::<Close>::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);

View File

@ -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<Upgraded>,
}
/// Given a websocket, return a protocol stream wrapper.
pub fn wrap_ws_in_nostr(ws: WebSocketStream<Upgraded>) -> NostrStream {
NostrStream { ws_stream: ws }
}
/// Implement the [`Stream`] interface to produce Nostr messages.
impl Stream for NostrStream {
type Item = Result<NostrMessage>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
// get the configuration
/// Convert Message to NostrMessage
fn convert(msg: String) -> Result<NostrMessage> {
let config = config::SETTINGS.read().unwrap();
let parsed_res: Result<NostrMessage> = 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<NostrResponse> for NostrStream {
type Error = Error;
fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
// 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<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn poll_close(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
}