Method observeSession

internal/bridge/watcher.go:13–54

observeSession subscribes to a session's observe channel and forwards events to C2. Re-subscribes automatically if the channel closes while the bridge is still running.

(sessionID string)

// observeSession subscribes to a session's observe channel and forwards events to C2.
// Re-subscribes automatically if the channel closes while the bridge is still running.
func (b *Bridge) observeSession(sessionID string) {
	subID := "bridge-observe-" + sessionID
	for {
		mgr := sessions.Global()
		if mgr == nil {
			return
		}
		ch := mgr.SubscribeObserve(sessionID, subID)
		if ch == nil {
			log.Warnf("[bridge] failed to subscribe observe for session %s", sessionID)
			return
		}
		log.Infof("[bridge] observeSession started for %s", sessionID)

		done := false
		for !done {
			select {
			case event, ok := <-ch:
				if !ok {
					done = true
				} else {
					b.forwardObserveEvent(event)
				}
			case <-b.ctx.Done():
				if mgr := sessions.Global(); mgr != nil {
					mgr.UnsubscribeObserve(sessionID, subID)
				}
				return
			}
		}

		if mgr := sessions.Global(); mgr != nil {
			mgr.UnsubscribeObserve(sessionID, subID)
		}

		if b.ctx.Err() != nil {
			return
		}
		log.Infof("[bridge] observe channel closed for %s, re-subscribing", sessionID)
		time.Sleep(1 * time.Second)
	}
}

// checkinSession sends a single checkin ping for a session.
func (b *Bridge) checkinSession(sessionID string) error {