Matrix-Mirrors/anonymousland-synapse
1
0
Fork 0
mirror of https://git.anonymousland.org/anonymousland/synapse.git synced 2025-08-12 03:00:05 -04:00
Code Issues Projects Releases Packages Wiki Activity

Move client command handling out of TCP protocol (#7185)

Browse source 
The aim here is to move the command handling out of the TCP protocol classes and to also merge the client and server command handling (so that we can reuse them for redis protocol). This PR simply moves the client paths to the new `ReplicationCommandHandler`, a future PR will move the server paths too.
This commit is contained in:
Erik Johnston 2020-04-06 09:58:42 +01:00 • committed by GitHub
parent d73bf18d13
commit 5016b162fc
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
12 changed files with 379 additions and 372 deletions
Download patch file Download diff file Expand all files Collapse all files

38
tests/replication/tcp/streams/_base.py
View file

@ -15,7 +15,7 @@
from mock import Mock
from synapse.replication.tcp.commands import ReplicateCommand
from synapse.replication.tcp.handler import ReplicationCommandHandler
from synapse.replication.tcp.protocol import ClientReplicationStreamProtocol
from synapse.replication.tcp.resource import ReplicationStreamProtocolFactory
@ -26,15 +26,20 @@ from tests.server import FakeTransport
class BaseStreamTestCase(unittest.HomeserverTestCase):
"""Base class for tests of the replication streams"""
def make_homeserver(self, reactor, clock):
self.test_handler = Mock(wraps=TestReplicationDataHandler())
return self.setup_test_homeserver(replication_data_handler=self.test_handler)
def prepare(self, reactor, clock, hs):
# build a replication server
server_factory = ReplicationStreamProtocolFactory(self.hs)
self.streamer = server_factory.streamer
server_factory = ReplicationStreamProtocolFactory(hs)
self.streamer = hs.get_replication_streamer()
self.server = server_factory.buildProtocol(None)
self.test_handler = Mock(wraps=TestReplicationClientHandler())
repl_handler = ReplicationCommandHandler(hs)
repl_handler.handler = self.test_handler
self.client = ClientReplicationStreamProtocol(
hs, "client", "test", clock, self.test_handler,
hs, "client", "test", clock, repl_handler,
)
self._client_transport = None
@ -69,13 +74,9 @@ class BaseStreamTestCase(unittest.HomeserverTestCase):
self.streamer.on_notifier_poke()
self.pump(0.1)
def replicate_stream(self):
"""Make the client end a REPLICATE command to set up a subscription to a stream"""
self.client.send_command(ReplicateCommand())
class TestReplicationClientHandler(object):
"""Drop-in for ReplicationClientHandler which just collects RDATA rows"""
class TestReplicationDataHandler:
"""Drop-in for ReplicationDataHandler which just collects RDATA rows"""
def __init__(self):
self.streams = set()
@ -88,18 +89,9 @@ class TestReplicationClientHandler(object):
positions[stream] = max(token, positions.get(stream, 0))
return positions
def get_currently_syncing_users(self):
return []
def update_connection(self, connection):
pass
def finished_connecting(self):
pass
async def on_position(self, stream_name, token):
"""Called when we get new position data."""
async def on_rdata(self, stream_name, token, rows):
for r in rows:
self._received_rdata_rows.append((stream_name, token, r))
async def on_position(self, stream_name, token):
pass