Merge pull request #352 from matrix-org/dmr/close-v2-server

tests-integration/connection_test.go

@@ -253,6 +253,7 @@ func TestConnectionTimeoutNotReset(t *testing.T) {
 	}
 	req.SetTimeoutMSecs(1000) // 1s
 	// inject 4 events 500ms apart - if we reset the timeout each time then we will return late
+	done := make(chan struct{})
 	go func() {
 		time.Sleep(10 * time.Millisecond)
 		ticker := time.NewTicker(500 * time.Millisecond)
@@ -277,6 +278,7 @@ func TestConnectionTimeoutNotReset(t *testing.T) {
 			})
 			i++
 		}
+		done <- struct{}{}
 	}()
 	startTime := time.Now()
 	res = v3.mustDoV3RequestWithPos(t, aliceToken, res.Pos, req)
@@ -286,6 +288,8 @@ func TestConnectionTimeoutNotReset(t *testing.T) {
 	}
 	m.MatchResponse(t, res, m.MatchList("a", m.MatchV3Count(2)), m.MatchNoV3Ops())
 
+	// Wait for all the responses before closing the v2 server.
+	<-done
 }
 
 // Test that the txn_id is echoed back

tests-integration/v3_test.go

@@ -207,7 +207,13 @@ func (s *testV2Server) nextResponse(userID, token string) *sync2.SyncResponse {
 		cond.Broadcast()
 	}
 	select {
-	case data := <-ch:
+	case data, stillOpen := <-ch:
+		if !stillOpen {
+			if !testutils.Quiet {
+				log.Printf("testV2Server: closing, returning null to %s %s", userID, token)
+			}
+			return nil
+		}
 		if !testutils.Quiet {
 			log.Printf(
 				"testV2Server: nextResponse %s %s returning data: [invite=%d,join=%d,leave=%d]",
@@ -228,6 +234,11 @@ func (s *testV2Server) url() string {
 }
 
 func (s *testV2Server) close() {
+	s.mu.Lock()
+	for _, ch := range s.queues {
+		close(ch)
+	}
+	s.mu.Unlock()
 	s.srv.Close()
 }