diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py
index 38ae360bcd..7868575a2e 100644
--- a/synapse/federation/replication.py
+++ b/synapse/federation/replication.py
@@ -541,7 +541,8 @@ class _TransactionQueue(object):
)
def eb(failure):
- deferred.errback(failure)
+ if not deferred.called:
+ deferred.errback(failure)
self._attempt_new_transaction(destination).addErrback(eb)
return deferred
diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py
index f141e92ce2..b37c8be964 100644
--- a/synapse/handlers/_base.py
+++ b/synapse/handlers/_base.py
@@ -35,7 +35,7 @@ class BaseRoomHandler(BaseHandler):
extra_users=[]):
snapshot.fill_out_prev_events(event)
- store_id = yield self.store.persist_event(event)
+ yield self.store.persist_event(event)
destinations = set(extra_destinations)
# Send a PDU to all hosts who have joined the room.
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index 9023c3d403..eac110419c 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -22,8 +22,6 @@ from synapse.api.constants import Membership
from synapse.util.logutils import log_function
from synapse.federation.pdu_codec import PduCodec
-from synapse.api.errors import AuthError
-
from twisted.internet import defer
import logging
@@ -87,12 +85,6 @@ class FederationHandler(BaseHandler):
yield self.replication_layer.send_pdu(pdu)
@log_function
- def get_state_for_room(self, destination, room_id):
- return self.replication_layer.get_state_for_context(
- destination, room_id
- )
-
- @log_function
@defer.inlineCallbacks
def on_receive_pdu(self, pdu, backfilled):
""" Called by the ReplicationLayer when we have a new pdu. We need to
@@ -141,19 +133,19 @@ class FederationHandler(BaseHandler):
yield self.hs.get_handlers().room_member_handler.change_membership(
new_event,
- True
+ do_auth=True
)
else:
with (yield self.room_lock.lock(event.room_id)):
- store_id = yield self.store.persist_event(event, backfilled)
+ yield self.store.persist_event(event, backfilled)
room = yield self.store.get_room(event.room_id)
if not room:
# Huh, let's try and get the current state
try:
- yield self.get_state_for_room(
+ yield self.replication_layer.get_state_for_context(
event.origin, event.room_id
)
@@ -163,9 +155,9 @@ class FederationHandler(BaseHandler):
if self.hs.hostname in hosts:
try:
yield self.store.store_room(
- event.room_id,
- "",
- is_public=False
+ room_id=event.room_id,
+ room_creator_user_id="",
+ is_public=False,
)
except:
pass
@@ -188,27 +180,14 @@ class FederationHandler(BaseHandler):
@log_function
@defer.inlineCallbacks
def backfill(self, dest, room_id, limit):
- events = yield self._backfill(dest, room_id, limit)
-
- for event in events:
- try:
- yield self.store.persist_event(event, backfilled=True)
- except:
- logger.exception("Failed to persist event: %s", event)
-
- defer.returnValue(events)
-
- @defer.inlineCallbacks
- def _backfill(self, dest, room_id, limit):
pdus = yield self.replication_layer.backfill(dest, room_id, limit)
- if not pdus:
- defer.returnValue([])
+ events = []
- events = [
- self.pdu_codec.event_from_pdu(pdu)
- for pdu in pdus
- ]
+ for pdu in pdus:
+ event = self.pdu_codec.event_from_pdu(pdu)
+ events.append(event)
+ yield self.store.persist_event(event, backfilled=True)
defer.returnValue(events)
@@ -224,7 +203,9 @@ class FederationHandler(BaseHandler):
# First get current state to see if we are already joined.
try:
- yield self.get_state_for_room(target_host, room_id)
+ yield self.replication_layer.get_state_for_context(
+ target_host, room_id
+ )
hosts = yield self.store.get_joined_hosts_for_room(room_id)
if self.hs.hostname in hosts:
@@ -254,8 +235,8 @@ class FederationHandler(BaseHandler):
try:
yield self.store.store_room(
- room_id,
- "",
+ room_id=room_id,
+ room_creator_user_id="",
is_public=False
)
except:
diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py
index c479908f61..bef1508892 100644
--- a/synapse/handlers/presence.py
+++ b/synapse/handlers/presence.py
@@ -141,10 +141,6 @@ class PresenceHandler(BaseHandler):
@defer.inlineCallbacks
def is_presence_visible(self, observer_user, observed_user):
- defer.returnValue(True)
- return
- # FIXME (erikj): This code path absolutely kills the database.
-
assert(observed_user.is_mine)
if observer_user == observed_user:
@@ -189,10 +185,6 @@ class PresenceHandler(BaseHandler):
@defer.inlineCallbacks
def set_state(self, target_user, auth_user, state):
- return
- # TODO (erikj): Turn this back on. Why did we end up sending EDUs
- # everywhere?
-
if not target_user.is_mine:
raise SynapseError(400, "User is not hosted on this Home Server")
@@ -445,16 +437,22 @@ class PresenceHandler(BaseHandler):
)
def _start_polling_remote(self, user, domain, remoteusers):
+ to_poll = set()
+
for u in remoteusers:
if u not in self._remote_recvmap:
self._remote_recvmap[u] = set()
+ to_poll.add(u)
self._remote_recvmap[u].add(user)
+ if not to_poll:
+ return defer.succeed(None)
+
return self.federation.send_edu(
destination=domain,
edu_type="m.presence",
- content={"poll": [u.to_string() for u in remoteusers]}
+ content={"poll": [u.to_string() for u in to_poll]}
)
def stop_polling_presence(self, user, target_user=None):
@@ -497,16 +495,22 @@ class PresenceHandler(BaseHandler):
del self._local_pushmap[localpart]
def _stop_polling_remote(self, user, domain, remoteusers):
+ to_unpoll = set()
+
for u in remoteusers:
self._remote_recvmap[u].remove(user)
if not self._remote_recvmap[u]:
del self._remote_recvmap[u]
+ to_unpoll.add(u)
+
+ if not to_unpoll:
+ return defer.succeed(None)
return self.federation.send_edu(
destination=domain,
edu_type="m.presence",
- content={"unpoll": [u.to_string() for u in remoteusers]}
+ content={"unpoll": [u.to_string() for u in to_unpoll]}
)
@defer.inlineCallbacks
diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index eb638fe50a..3e41d7a46b 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -325,7 +325,8 @@ class RoomMemberHandler(BaseRoomHandler):
)
snapshot = yield self.store.snapshot_room(
- room_id, joinee, RoomMemberEvent.TYPE, joinee
+ room_id, joinee.to_string(), RoomMemberEvent.TYPE,
+ joinee.to_string()
)
yield self._do_join(new_event, snapshot, room_host=host, do_auth=True)
diff --git a/synapse/handlers/typing.py b/synapse/handlers/typing.py
index 9d38a7336e..9fab0ff37c 100644
--- a/synapse/handlers/typing.py
+++ b/synapse/handlers/typing.py
@@ -17,11 +17,12 @@ from twisted.internet import defer
from ._base import BaseHandler
+from synapse.api.errors import SynapseError, AuthError
+
import logging
from collections import namedtuple
-
logger = logging.getLogger(__name__)
diff --git a/synapse/rest/login.py b/synapse/rest/login.py
index bcf63fd2ab..99e4f10aac 100644
--- a/synapse/rest/login.py
+++ b/synapse/rest/login.py
@@ -27,7 +27,7 @@ class LoginRestServlet(RestServlet):
PASS_TYPE = "m.login.password"
def on_GET(self, request):
- return (200, {"type": LoginRestServlet.PASS_TYPE})
+ return (200, {"flows": [{"type": LoginRestServlet.PASS_TYPE}]})
def on_OPTIONS(self, request):
return (200, {})
diff --git a/synapse/server.py b/synapse/server.py
index ade8dc6c15..3e72b2bcd5 100644
--- a/synapse/server.py
+++ b/synapse/server.py
@@ -126,11 +126,6 @@ class BaseHomeServer(object):
object."""
return UserID.from_string(s, hs=self)
- def parse_roomid(self, s):
- """Parse the string given by 's' as a Room ID and return a RoomID
- object."""
- return RoomID.from_string(s, hs=self)
-
def parse_roomalias(self, s):
"""Parse the string given by 's' as a Room Alias and return a RoomAlias
object."""
diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py
index 4f42afc015..4945b0796e 100644
--- a/synapse/storage/stream.py
+++ b/synapse/storage/stream.py
@@ -294,7 +294,7 @@ class StreamStore(SQLBaseStore):
logger.debug("get_room_events_max_id: %s", res)
if not res or not res[0] or not res[0]["m"]:
- return "s1"
+ return "s0"
key = res[0]["m"]
return "s%d" % (key,)
diff --git a/synapse/streams/events.py b/synapse/streams/events.py
index eaa397c650..c68cf1a59c 100644
--- a/synapse/streams/events.py
+++ b/synapse/streams/events.py
@@ -15,7 +15,6 @@
from twisted.internet import defer
-from synapse.api.constants import Membership
from synapse.types import StreamToken
@@ -32,7 +31,7 @@ class NullSource(object):
return defer.succeed(0)
def get_pagination_rows(self, user, pagination_config, key):
- return defer.succeed(([], from_token))
+ return defer.succeed(([], pagination_config.from_token))
class RoomEventSource(object):
diff --git a/synapse/util/logutils.py b/synapse/util/logutils.py
index 9270a1790b..021649071b 100644
--- a/synapse/util/logutils.py
+++ b/synapse/util/logutils.py
@@ -15,6 +15,7 @@
from inspect import getcallargs
+from functools import wraps
import logging
@@ -26,6 +27,7 @@ def log_function(f):
lineno = f.func_code.co_firstlineno
pathname = f.func_code.co_filename
+ @wraps(f)
def wrapped(*args, **kwargs):
name = f.__module__
logger = logging.getLogger(name)
|
