Merge pull request #1097 from matrix-org/erikj/replication_typing_rest
Correctly handle typing stream id resetting
This commit is contained in:
commit
d1c217c823
|
@ -242,6 +242,9 @@ class SynchrotronTyping(object):
|
||||||
self._room_typing = {}
|
self._room_typing = {}
|
||||||
|
|
||||||
def stream_positions(self):
|
def stream_positions(self):
|
||||||
|
# We must update this typing token from the response of the previous
|
||||||
|
# sync. In particular, the stream id may "reset" back to zero/a low
|
||||||
|
# value which we *must* use for the next replication request.
|
||||||
return {"typing": self._latest_room_serial}
|
return {"typing": self._latest_room_serial}
|
||||||
|
|
||||||
def process_replication(self, result):
|
def process_replication(self, result):
|
||||||
|
|
|
@ -274,11 +274,18 @@ class ReplicationResource(Resource):
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def typing(self, writer, current_token, request_streams):
|
def typing(self, writer, current_token, request_streams):
|
||||||
current_position = current_token.presence
|
current_position = current_token.typing
|
||||||
|
|
||||||
request_typing = request_streams.get("typing")
|
request_typing = request_streams.get("typing")
|
||||||
|
|
||||||
if request_typing is not None:
|
if request_typing is not None:
|
||||||
|
# If they have a higher token than current max, we can assume that
|
||||||
|
# they had been talking to a previous instance of the master. Since
|
||||||
|
# we reset the token on restart, the best (but hacky) thing we can
|
||||||
|
# do is to simply resend down all the typing notifications.
|
||||||
|
if request_typing > current_position:
|
||||||
|
request_typing = 0
|
||||||
|
|
||||||
typing_rows = yield self.typing_handler.get_all_typing_updates(
|
typing_rows = yield self.typing_handler.get_all_typing_updates(
|
||||||
request_typing, current_position
|
request_typing, current_position
|
||||||
)
|
)
|
||||||
|
|
Loading…
Reference in New Issue