package message import ( "context" "git.asio.asia/nixevol/NixMsg/internal/app/port" ) type pushWorker struct { endpointID string connID port.ConnID wake chan struct{} stop chan struct{} done chan struct{} } func (a *App) startPushWorker(endpointID string, connID port.ConnID) { a.stopPushWorker(endpointID, "") w := &pushWorker{ endpointID: endpointID, connID: connID, wake: make(chan struct{}, 1), stop: make(chan struct{}), done: make(chan struct{}), } a.mu.Lock() a.workers[endpointID] = w a.mu.Unlock() go a.runPushWorker(w) } func (a *App) stopPushWorker(endpointID string, connID port.ConnID) { a.mu.Lock() w, ok := a.workers[endpointID] if !ok || (connID != "" && w.connID != connID) { a.mu.Unlock() return } delete(a.workers, endpointID) a.mu.Unlock() close(w.stop) <-w.done } func (a *App) runPushWorker(w *pushWorker) { defer close(w.done) for { select { case <-w.stop: return case <-w.wake: ctx, cancel := context.WithTimeout(context.Background(), pushOpTimeout) _ = a.PushPending(ctx, w.endpointID, w.connID) a.flushRevokes(ctx) cancel() } } } func (a *App) signalWorker(endpointID string) { a.mu.Lock() w := a.workers[endpointID] a.mu.Unlock() if w == nil { return } select { case w.wake <- struct{}{}: default: } } func (a *App) markHandshook(endpointID string, conn LiveConn) { conn.Ready = true a.mu.Lock() a.handshook[endpointID] = conn.ConnID a.mu.Unlock() if setter, ok := a.conns.(interface { Set(string, LiveConn) }); ok { setter.Set(endpointID, conn) } } func (a *App) clearHandshook(endpointID string, connID port.ConnID) { a.mu.Lock() defer a.mu.Unlock() if cur, ok := a.handshook[endpointID]; ok && (connID == "" || cur == connID) { delete(a.handshook, endpointID) } } func (a *App) canPush(endpointID string, connID port.ConnID) (LiveConn, port.ConnID, bool) { live, ok := a.lookupConn(endpointID) if !ok { return LiveConn{}, "", false } if connID != "" && live.ConnID != connID { return LiveConn{}, "", false } if connID == "" { connID = live.ConnID } if live.Ready { return live, connID, true } a.mu.Lock() hs, marked := a.handshook[endpointID] a.mu.Unlock() if marked && hs == live.ConnID { return live, connID, true } return LiveConn{}, "", false } func (a *App) isReadyEndpoint(endpointID string) bool { _, _, ok := a.canPush(endpointID, "") return ok } func shortWriteCtx() (context.Context, context.CancelFunc) { return context.WithTimeout(context.Background(), clearPushedTimeout) } func (a *App) NotifyDispatch() { if a.dispatchCh == nil { return } select { case a.dispatchCh <- struct{}{}: default: } }