diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index 7026df90a2..9cff444779 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -32,10 +32,19 @@ logger = logging.getLogger(__name__)
class FederationHandler(BaseHandler):
"""Handles events that originated from federation."""
+ def __init__(self, hs):
+ super(FederationHandler, self).__init__(hs)
+
+ self.distributor.observe(
+ "user_joined_room",
+ self._on_user_joined
+ )
+
+ self.waiting_for_join_list = {}
@log_function
@defer.inlineCallbacks
- def on_receive(self, event, is_new_state):
+ def on_receive(self, event, is_new_state, backfilled):
if hasattr(event, "state_key") and not is_new_state:
logger.debug("Ignoring old state.")
return
@@ -70,6 +79,115 @@ class FederationHandler(BaseHandler):
else:
with (yield self.room_lock.lock(event.room_id)):
- store_id = yield self.store.persist_event(event)
+ store_id = 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:
+ federation = self.hs.get_federation()
+ yield federation.get_state_for_room(
+ event.origin, event.room_id
+ )
+
+ hosts = yield self.store.get_joined_hosts_for_room(
+ event.room_id
+ )
+ if self.hs.hostname in hosts:
+ try:
+ yield self.store.store_room(
+ event.room_id,
+ "",
+ is_public=False
+ )
+ except:
+ pass
+ except:
+ logger.exception(
+ "Failed to get current state for room %s",
+ event.room_id
+ )
+
+ if not backfilled:
+ yield self.notifier.on_new_room_event(event, store_id)
+
+ if event.type == RoomMemberEvent.TYPE:
+ if event.membership == Membership.JOIN:
+ user = self.hs.parse_userid(event.target_user_id)
+ self.distributor.fire(
+ "user_joined_room", user=user, room_id=event.room_id
+ )
+
+
+ @log_function
+ @defer.inlineCallbacks
+ def backfill(self, dest, room_id, limit):
+ events = yield self.hs.get_federation().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)
- yield self.notifier.on_new_room_event(event, store_id)
+ @log_function
+ @defer.inlineCallbacks
+ def do_invite_join(self, target_host, room_id, joinee, content):
+ federation = self.hs.get_federation()
+
+ hosts = yield self.store.get_joined_hosts_for_room(room_id)
+ if self.hs.hostname in hosts:
+ # We are already in the room.
+ logger.debug("We're already in the room apparently")
+ defer.returnValue(False)
+
+ # First get current state to see if we are already joined.
+ try:
+ yield federation.get_state_for_room(target_host, room_id)
+
+ hosts = yield self.store.get_joined_hosts_for_room(room_id)
+ if self.hs.hostname in hosts:
+ # Oh, we were actually in the room already.
+ logger.debug("We're already in the room apparently")
+ defer.returnValue(False)
+ except Exception:
+ logger.exception("Failed to get current state")
+
+ new_event = self.event_factory.create_event(
+ etype=InviteJoinEvent.TYPE,
+ target_host=target_host,
+ room_id=room_id,
+ user_id=joinee,
+ content=content
+ )
+
+ new_event.destinations = [target_host]
+
+ yield federation.handle_new_event(new_event)
+
+ # TODO (erikj): Time out here.
+ d = defer.Deferred()
+ self.waiting_for_join_list.setdefault((joinee, room_id), []).append(d)
+ yield d
+
+ try:
+ yield self.store.store_room(
+ event.room_id,
+ "",
+ is_public=False
+ )
+ except:
+ pass
+
+
+ defer.returnValue(True)
+
+
+ @log_function
+ def _on_user_joined(self, user, room_id):
+ waiters = self.waiting_for_join_list.get((user.to_string(), room_id), [])
+ while waiters:
+ waiters.pop().callback(None)
|
