updates mayve:

stream/stream.go

@@ -74,6 +74,10 @@ func (s *Stream) Sub(sb Subscriber) string {
 	return o.id
 }
 
+func (s *Stream) Unsubscribe(id string) {
+	s.subs.Delete(id)
+}
+
 func (s *Stream) run() {
 	go s.run_update()
 	go s.run_send()

stream/websocket_client.go

@@ -54,7 +54,7 @@ type cmd struct {
 		Topic string           `json:"topic"`
 		Patch gjson.RawMessage `json:"patch"`
 		Seq   int              `json:"seq"`
-	}
+	} `json:"params"`
 }
 
 func (ws *WebsocketClient) connect() (err error) {
@@ -134,8 +134,13 @@ func (w *Receiver) Seq() int {
 	return w.seq
 }
 
+type sub_cmd struct {
+	Method string   `json:"method"`
+	Params []string `json:"params"`
+}
+
 func (w *Receiver) subscribe() error {
-	return w.c.c.WriteJSON(map[string]any{"method": "topic_subscribe", "params": []string{w.topic}})
+	return w.c.c.WriteJSON(sub_cmd{Method: "topic_subscribe", Params: []string{w.topic}})
 }
 
 func (w *Receiver) OnPatch(seq int, diffs []byte) error {
@@ -159,7 +164,7 @@ func (w *Receiver) OnPatchRaw(seq int, diffs gjson.RawMessage) error {
 }
 
 func (w *Receiver) Close() {
-	w.c.c.WriteJSON(map[string]any{"topic": "topic_unsubscribe", "params": []string{w.topic}})
+	w.c.c.WriteJSON(sub_cmd{Method: "topic_unsubscribe", Params: []string{w.topic}})
 	w.c.Lock()
 	defer w.c.Unlock()
 	delete(w.c.Streams, w.topic)

stream/websocket.go → stream/websocket_server.go

@@ -4,8 +4,8 @@ import (
 	"context"
 	"log"
 	"net/http"
-	"sync"
 
+	"gfx.cafe/open/goutil/generic"
 	"gfx.cafe/open/jsonpatch"
 	"gfx.cafe/open/jsonrpc2"
 	"gfx.cafe/open/jsonrpc2/external/websocket"
@@ -14,10 +14,8 @@ import (
 var DefaultUpgrader = websocket.Upgrader{} // use default options
 
 type WebsocketServer struct {
-	Streams map[string]*Stream
+	Streams generic.MapOf[string, *Stream]
 	u       websocket.Upgrader
-
-	mu sync.RWMutex
 }
 
 func NewWssServer(upgrader ...websocket.Upgrader) *WebsocketServer {
@@ -26,7 +24,6 @@ func NewWssServer(upgrader ...websocket.Upgrader) *WebsocketServer {
 		u = upgrader[0]
 	}
 	return &WebsocketServer{
-		Streams: make(map[string]*Stream),
 		u: u,
 	}
 }
@@ -43,6 +40,11 @@ func (ws *WebsocketServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	defer c.Close()
+	ws.handleClient(c, w, r)
+}
+
+func (ws *WebsocketServer) handleClient(c *websocket.Conn, w http.ResponseWriter, r *http.Request) {
+	routes := map[string]string{}
 	for {
 		_, message, err := c.ReadMessage()
 		if err != nil {
@@ -62,27 +64,31 @@ func (ws *WebsocketServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			continue
 		}
 		switch cmd.Method {
+		case "close":
+			break
 		case "topic_subscribe":
 			for _, v := range cmd.Params {
-				ws.mu.RLock()
-				if st, ok := ws.Streams[v]; ok {
-					ws.mu.RUnlock()
+				if st, ok := ws.Streams.Load(v); ok {
 					sub := NewWssSubscriber(r.Context(), v, c)
-					st.Sub(sub)
-				} else {
-					ws.mu.RUnlock()
+					xid_str := st.Sub(sub)
+					routes[v] = xid_str
 				}
 			}
 		case "topic_unsubscribe":
+			for _, v := range cmd.Params {
+				if id, ok := routes[v]; ok {
+					if st, ok := ws.Streams.Load(v); ok {
+						st.Unsubscribe(id)
+					}
+				}
+			}
 		}
 	}
 }
 
 func ServerRegister[T any](w *WebsocketServer, route string, v T) *TStream[T] {
 	o := New()
-	w.mu.Lock()
-	defer w.mu.Unlock()
-	w.Streams[route] = o
+	w.Streams.Store(route, o)
 	return &TStream[T]{
 		o,
 	}