bridgev2: split stopping matrix connector

Also fix stopping the websocket in the default Matrix connector

Author: tulir

bridgev2/bridge.go

@@ -319,7 +319,7 @@ func (br *Bridge) Stop() {
 func (br *Bridge) stop(isRunOnce bool) {
 	br.Log.Info().Msg("Shutting down bridge")
 	br.stopBackfillQueue.Set()
-	br.Matrix.Stop()
+	br.Matrix.PreStop()
 	if !isRunOnce {
 		br.cacheLock.Lock()
 		var wg sync.WaitGroup
@@ -330,6 +330,7 @@ func (br *Bridge) stop(isRunOnce bool) {
 		br.cacheLock.Unlock()
 		wg.Wait()
 	}
+	br.Matrix.Stop()
 	if stopNet, ok := br.Network.(StoppableNetwork); ok {
 		stopNet.Stop()
 	}

bridgev2/matrix/connector.go

@@ -233,13 +233,37 @@ func (br *Connector) GetCapabilities() *bridgev2.MatrixCapabilities {
 	return br.Capabilities
 }
 
-func (br *Connector) Stop() {
+func sendStopSignal(ch chan struct{}) {
+	if ch != nil {
+		select {
+		case ch <- struct{}{}:
+		default:
+		}
+	}
+}
+
+func (br *Connector) PreStop() {
 	br.stopping = true
 	br.AS.Stop()
+	if stopWebsocket := br.AS.StopWebsocket; stopWebsocket != nil {
+		stopWebsocket(appservice.ErrWebsocketManualStop)
+	}
+	sendStopSignal(br.wsStopPinger)
+	sendStopSignal(br.wsShortCircuitReconnectBackoff)
+}
+
+func (br *Connector) Stop() {
 	br.EventProcessor.Stop()
 	if br.Crypto != nil {
 		br.Crypto.Stop()
 	}
+	if wsStopChan := br.wsStopped; wsStopChan != nil {
+		select {
+		case <-wsStopChan:
+		case <-time.After(4 * time.Second):
+			br.Log.Warn().Msg("Timed out waiting for websocket to close")
+		}
+	}
 }
 
 var MinSpecVersion = mautrix.SpecV14

bridgev2/matrixinterface.go

@@ -31,6 +31,7 @@ type MatrixCapabilities struct {
 type MatrixConnector interface {
 	Init(*Bridge)
 	Start(ctx context.Context) error
+	PreStop()
 	Stop()
 
 	GetCapabilities() *MatrixCapabilities