Perform subscribe/unsubscribe from websocket

src/main.rs

@@ -72,7 +72,7 @@ async fn nostr_server(stream: TcpStream) {
 // Handles valid clients who have upgraded to WebSockets
 async fn process_client(stream: WebSocketStream<TcpStream>) {
     // get a protocol helper;
-    let proto = Proto::new();
+    let mut proto = Proto::new();
     let (mut write, mut read) = stream.split();
     // TODO: select on a timeout to kill non-responsive clients
 
@@ -91,7 +91,7 @@ async fn process_client(stream: WebSocketStream<TcpStream>) {
                     .await
                     .expect("send failed");
                 // Handle this request. Everything else below is basically websocket error handling.
-                proto.process_message(cmd);
+                proto.process_message(cmd).ok();
             }
             Ok(Message::Binary(_)) => {
                 info!("Ignoring Binary message");

src/proto.rs

@@ -27,13 +27,20 @@ impl Proto {
         p
     }
 
-    pub fn process_message(self: &Self, cmd: String) {
+    // TODO: figure out NOTICE handling for errors here
+    pub fn process_message(&mut self, cmd: String) -> Result<()> {
         info!(
             "Processing message in proto for client: {:?}",
             self.client_id
         );
-        // check what kind of message
-        info!("Parse result: {:?}", parse_type(cmd));
+        let message = parse_cmd(cmd)?;
+        info!("Parsed message: {:?}", message);
+        match message {
+            NostrRequest::EvReq(_) => {}
+            NostrRequest::SubReq(sub) => self.subscribe(sub),
+            NostrRequest::CloseReq(close) => self.unsubscribe(close),
+        };
+        Ok(())
     }
 
     pub fn subscribe(&mut self, s: Subscription) {
@@ -61,11 +68,18 @@ impl Proto {
         }
         // add subscription
         self.subscriptions.insert(k, s);
+        info!(
+            "Registered new subscription, currently have {} active subs",
+            self.subscriptions.len()
+        );
     }
 
     pub fn unsubscribe(&mut self, c: Close) {
         self.subscriptions.remove(&c.get_id());
-        unimplemented!();
+        info!(
+            "Removed subscription, currently have {} active subs",
+            self.subscriptions.len()
+        );
     }
 }
 
@@ -99,7 +113,7 @@ fn msg_type_wrapper(msg: String) -> Result<NostrRawMessage> {
     }
 }
 
-pub fn parse_type(msg: String) -> Result<NostrRequest> {
+pub fn parse_cmd(msg: String) -> Result<NostrRequest> {
     // turn this raw string into a parsed request
     let typ = msg_type_wrapper(msg)?;
     match typ {

src/subscription.rs

@@ -86,6 +86,9 @@ impl Subscription {
     pub fn get_id(&self) -> String {
         self.id.clone()
     }
+    pub fn get_filter_count(&self) -> usize {
+        self.filters.len()
+    }
 }
 
 #[cfg(test)]