Fix a bug where only the initial buffer would get sent.

Author: clokep

synapse/logging/_remote.py

@@ -61,9 +61,6 @@ def stopProducing(self):
 
     def resumeProducing(self):
         # If we're already producing, nothing to do.
-        if not self._paused:
-            return
-
         self._paused = False
 
         # Loop until paused.

tests/logging/test_remote_handler.py

@@ -14,17 +14,14 @@
 # limitations under the License.
 import logging
 
+from twisted.test.proto_helpers import AccumulatingProtocol
+
 from synapse.logging import RemoteHandler
 
-from tests.server import connect_client
+from tests.server import FakeTransport
 from tests.unittest import DEBUG, HomeserverTestCase
 
 
-class FakeBeginner:
-    def beginLoggingTo(self, observers, **kwargs):
-        self.observers = observers
-
-
 class StructuredLoggingTestBase:
     """
     Test base that registers a cleanup handler to reset the stdlib log handler
@@ -38,6 +35,20 @@ def _cleanup():
         self.addCleanup(_cleanup)
 
 
+def connect_logging_client(reactor, client_id):
+    # This is essentially tests.server.connect_client, but disabling autoflush on
+    # the client transport. This is necessary to avoid an infinite loop due to
+    # sending of data via the logging transport causing additional logs to be
+    # written.
+    factory = reactor.tcpClients.pop(client_id)[2]
+    client = factory.buildProtocol(None)
+    server = AccumulatingProtocol()
+    server.makeConnection(FakeTransport(client, reactor))
+    client.makeConnection(FakeTransport(server, reactor, autoflush=False))
+
+    return client, server
+
+
 class RemoteHandlerTestCase(StructuredLoggingTestBase, HomeserverTestCase):
     @DEBUG
     def test_log_output(self):
@@ -51,12 +62,10 @@ def test_log_output(self):
         logger.info("Hello there, %s!", "wally")
 
         # Trigger the connection
-        self.pump()
-
-        _, server = connect_client(self.reactor, 0)
+        client, server = connect_logging_client(self.reactor, 0)
 
         # Trigger data being sent
-        self.pump()
+        client.transport.flush()
 
         # One log message, with a single trailing newline
         logs = server.data.decode("utf8").splitlines()
@@ -89,8 +98,8 @@ def test_log_backpressure_debug(self):
         logger.debug("too much debug")
 
         # Allow the reconnection
-        _, server = connect_client(self.reactor, 0)
-        self.pump()
+        client, server = connect_logging_client(self.reactor, 0)
+        client.transport.flush()
 
         # Only the 7 infos made it through, the debugs were elided
         logs = server.data.splitlines()
@@ -124,8 +133,8 @@ def test_log_backpressure_info(self):
         logger.debug("too much debug")
 
         # Allow the reconnection
-        _, server = connect_client(self.reactor, 0)
-        self.pump()
+        client, server = connect_logging_client(self.reactor, 0)
+        client.transport.flush()
 
         # The 10 warnings made it through, the debugs and infos were elided
         logs = server.data.splitlines()
@@ -150,8 +159,8 @@ def test_log_backpressure_cut_middle(self):
             logger.warning("warn %s" % (i,))
 
         # Allow the reconnection
-        _, server = connect_client(self.reactor, 0)
-        self.pump()
+        client, server = connect_logging_client(self.reactor, 0)
+        client.transport.flush()
 
         # The first five and last five warnings made it through, the debugs and
         # infos were elided

tests/server.py

@@ -571,12 +571,10 @@ def connect_client(reactor: IReactorTCP, client_id: int) -> AccumulatingProtocol
         reactor
         factory: The connecting factory to build.
     """
-    factory = reactor.tcpClients[client_id][2]
+    factory = reactor.tcpClients.pop(client_id)[2]
     client = factory.buildProtocol(None)
     server = AccumulatingProtocol()
     server.makeConnection(FakeTransport(client, reactor))
     client.makeConnection(FakeTransport(server, reactor))
 
-    reactor.tcpClients.pop(client_id)
-
     return client, server