Matrix-Mirrors/forked-synapse
1
0
Fork 0
mirror of https://mau.dev/maunium/synapse.git synced 2024-08-24 11:16:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add parse_row method to replication stream class

Browse Source 
This will allow individual stream classes to override how a row is parsed.
This commit is contained in:
Richard van der Hoff 2019-03-27 07:40:32 +00:00
parent 91c3513668
commit f570916a3e
3 changed files with 19 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
synapse/replication/tcp/client.py
View File

@ -105,13 +105,14 @@ class ReplicationClientHandler(object):
def on_rdata(self, stream_name, token, rows): def on_rdata(self, stream_name, token, rows):
"""Called to handle a batch of replication data with a given stream token. """Called to handle a batch of replication data with a given stream token.
By default this just pokes the slave store. Can be overriden in subclasses to By default this just pokes the slave store. Can be overridden in subclasses to
handle more. handle more.
Args: Args:
stream_name (str): name of the replication stream for this batch of rows stream_name (str): name of the replication stream for this batch of rows
token (int): stream token for this batch of rows token (int): stream token for this batch of rows
rows (list): a list of Stream.ROW_TYPE objects. rows (list): a list of Stream.ROW_TYPE objects as returned by
Stream.parse_row.
Returns: Returns:
Deferred|None Deferred|None

2
synapse/replication/tcp/protocol.py
View File

@ -605,7 +605,7 @@ class ClientReplicationStreamProtocol(BaseReplicationStreamProtocol):
inbound_rdata_count.labels(stream_name).inc() inbound_rdata_count.labels(stream_name).inc()
try: try:
row = STREAMS_MAP[stream_name].ROW_TYPE(*cmd.row) row = STREAMS_MAP[stream_name].parse_row(cmd.row)
except Exception: except Exception:
logger.exception( logger.exception(
"[%s] Failed to parse RDATA: %r %r", "[%s] Failed to parse RDATA: %r %r",

15
synapse/replication/tcp/streams/_base.py
View File

@ -115,6 +115,21 @@ class Stream(object):
ROW_TYPE = None # The type of the row ROW_TYPE = None # The type of the row
_LIMITED = True # Whether the update function takes a limit _LIMITED = True # Whether the update function takes a limit
@classmethod
def parse_row(cls, row):
"""Parse a row received over replication
By default, assumes that the row data is an array object and passes its contents
to the constructor of the ROW_TYPE for this stream.
Args:
row: row data from the incoming RDATA command, after json decoding
Returns:
ROW_TYPE object for this stream
"""
return cls.ROW_TYPE(*row)
def __init__(self, hs): def __init__(self, hs):
# The token from which we last asked for updates # The token from which we last asked for updates
self.last_token = self.current_token() self.last_token = self.current_token()