package server import ( "forge.lightcrystal.systems/lightcrystal/underbbs/models" "hacklab.nilfm.cc/quartzgun/auth" "hacklab.nilfm.cc/quartzgun/renderer" "net/http" "nhooyr.io/websocket" ) type Subscriber struct { msgs chan []byte closeSlow func() } type BBSServer struct { subscribeMessageBuffer int publishLimiter *rate.Limiter logf func(f string, v ...interface{}) serveMux http.ServeMux subscribersLock sync.Mutex subscribers map[*Subscriber][]models.Adapter } func New() *BBSServer { srvr := &BBSServer{ subscribeMessageBuffer: 16, logf: log.Printf, subscribers: make(map[*Subscriber][]model.Adapter), } // frontend is here srvr.serveMux.Handle("/app/", http.StripPrefix("/app/", renderer.Subtree("./dist"))) // websocket srvr.serveMux.HandleFunc("/subscribe", srvr.subscribeHandler) srvr.serveMux.HandleFunc("/publish", srvr.publishHandler) return srvr } func (self *GameTableServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { self.serveMux.ServeHTTP(w, r) } func (self *GameTableServer) subscribeHandler(w http.ResponseWriter, r *http.Request) { c, err := websocket.Accept(w, r, &websocket.AcceptOptions{ Subprotocols: [], }) if err != nil { self.logf("%v", err) return } defer c.Close(websocket.StatusInternalError, "") err = self.subscribe(r, c) if errors.Is(err, context.Canceled) { return } if websocket.CloseStatus(err) == websocket.StatusNormalClosure || websocket.CloseStatus(err) == websocket.StatusGoingAway { return } if err != nil { self.logf("%v", err) return } } func (self *GameTableServer) subscribe(r *http.Request, c *websocket.Conn) error { ctx := r.Context() ctx = c.CloseRead(ctx) s := &Subscriber{ msgs: make(chan []byte, self.subscribeMessageBuffer), closeSlow: func() { c.Close(websocket.StatusPolicyViolation, "connection too slow to keep up with messages") }, } // validate the subprotocol self.addSubscriber(s, tableKey) defer self.deleteSubscriber(s) select { case s.msgs <- self.getCurrentState(tableKey): default: go s.closeSlow() } for { select { case msg := <-s.msgs: err := writeTimeout(ctx, time.Second*5, c, msg) if err != nil { return err } case <-ctx.Done(): return ctx.Err() } } } func (self *GameTableServer) publishHandler(w http.ResponseWriter, r *http.Request) { if r.Method != "POST" { http.Error(w, http.StatusText(http.StatusMethodNotAllowed), http.StatusMethodNotAllowed) return } body := http.MaxBytesReader(w, r.Body, 8192) msg, err := ioutil.ReadAll(body) if err != nil { http.Error(w, http.StatusText(http.StatusRequestEntityTooLarge), http.StatusRequestEntityTooLarge) return } self.publish(msg) w.WriteHeader(http.StatusAccepted) } func (self *GameTableServer) publish(msg []byte) { self.subscribersLock.Lock() defer self.subscribersLock.Unlock() // send messages to our adapter(s) self.publishLimiter.Wait(context.Background()) // send any response from the adapter(s) back to the client for s, k := range self.subscribers { // whatever logic to select which subscriber to send back to } } func (self *GameTableServer) addSubscriber(s *Subscriber, k models.TableKey) { self.subscribersLock.Lock() self.subscribers[s] = k self.subscribersLock.Unlock() } func (self *GameTableServer) deleteSubscriber(s *Subscriber) { self.subscribersLock.Lock() delete(self.subscribers, s) self.subscribersLock.Unlock() } func writeTimeout(ctx context.Context, timeout time.Duration, c *websocket.Conn, msg []byte) error { ctx, cancel := context.WithTimeout(ctx, timeout) defer cancel() return c.Write(ctx, websocket.MessageText, msg) }