diff options
Diffstat (limited to 'synapse')
| -rw-r--r-- | synapse/__init__.py | 2 | ||||
| -rw-r--r-- | synapse/api/events/__init__.py | 1 | ||||
| -rw-r--r-- | synapse/api/events/factory.py | 7 | ||||
| -rwxr-xr-x | synapse/app/homeserver.py | 16 | ||||
| -rw-r--r-- | synapse/federation/replication.py | 4 | ||||
| -rw-r--r-- | synapse/handlers/_base.py | 1 | ||||
| -rw-r--r-- | synapse/handlers/federation.py | 29 | ||||
| -rw-r--r-- | synapse/handlers/presence.py | 8 | ||||
| -rw-r--r-- | synapse/handlers/room.py | 33 | ||||
| -rw-r--r-- | synapse/http/server.py | 28 | ||||
| -rw-r--r-- | synapse/rest/register.py | 4 | ||||
| -rw-r--r-- | synapse/server.py | 2 | ||||
| -rw-r--r-- | synapse/storage/__init__.py | 7 | ||||
| -rw-r--r-- | synapse/storage/_base.py | 5 | ||||
| -rw-r--r-- | synapse/storage/roommember.py | 2 | ||||
| -rw-r--r-- | synapse/storage/schema/im.sql | 24 | ||||
| -rw-r--r-- | synapse/storage/stream.py | 10 |
17 files changed, 153 insertions, 30 deletions
diff --git a/synapse/__init__.py b/synapse/__init__.py index 1e7b2ab272..47fc1b2ea4 100644 --- a/synapse/__init__.py +++ b/synapse/__init__.py @@ -15,3 +15,5 @@ """ This is a reference implementation of a synapse home server. """ + +__version__ = "0.0.1" diff --git a/synapse/api/events/__init__.py b/synapse/api/events/__init__.py index 921fd08832..aa04dbece7 100644 --- a/synapse/api/events/__init__.py +++ b/synapse/api/events/__init__.py @@ -51,6 +51,7 @@ class SynapseEvent(JsonEncodedObject): "depth", "destinations", "origin", + "outlier", ] required_keys = [ diff --git a/synapse/api/events/factory.py b/synapse/api/events/factory.py index b61dac7acd..c2cdcddf41 100644 --- a/synapse/api/events/factory.py +++ b/synapse/api/events/factory.py @@ -33,16 +33,21 @@ class EventFactory(object): RoomConfigEvent ] - def __init__(self): + def __init__(self, hs): self._event_list = {} # dict of TYPE to event class for event_class in EventFactory._event_classes: self._event_list[event_class.TYPE] = event_class + self.clock = hs.get_clock() + def create_event(self, etype=None, **kwargs): kwargs["type"] = etype if "event_id" not in kwargs: kwargs["event_id"] = random_string(10) + if "ts" not in kwargs: + kwargs["ts"] = int(self.clock.time_msec()) + if etype in self._event_list: handler = self._event_list[etype] else: diff --git a/synapse/app/homeserver.py b/synapse/app/homeserver.py index ca102236cf..40e3561ee5 100755 --- a/synapse/app/homeserver.py +++ b/synapse/app/homeserver.py @@ -37,6 +37,7 @@ import logging import logging.config import sqlite3 import os +import re logger = logging.getLogger(__name__) @@ -56,7 +57,7 @@ class SynapseHomeServer(HomeServer): return File("webclient") # TODO configurable? def build_resource_for_content_repo(self): - return ContentRepoResource("uploads", self.auth) + return ContentRepoResource(self, self.upload_dir, self.auth) def build_db_pool(self): """ Set up all the dbs. Since all the *.sql have IF NOT EXISTS, so we @@ -235,8 +236,8 @@ def setup(): parser.add_argument('--pid-file', dest="pid", help="When running as a " "daemon, the file to store the pid in", default="hs.pid") - parser.add_argument("-w", "--webclient", dest="webclient", - action="store_true", help="Host the web client.") + parser.add_argument("-W", "--webclient", dest="webclient", default=True, + action="store_false", help="Don't host a web client.") args = parser.parse_args() verbosity = int(args.verbose) if args.verbose else None @@ -255,9 +256,16 @@ def setup(): logger.info("Server hostname: %s", args.host) + if re.search(":[0-9]+$", args.host): + domain_with_port = args.host + else: + domain_with_port = "%s:%s" % (args.host, args.port) + hs = SynapseHomeServer( args.host, - db_name=db_name + domain_with_port=domain_with_port, + upload_dir=os.path.abspath("uploads"), + db_name=db_name, ) # This object doesn't need to be saved because it's set as the handler for diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py index 8030d0963f..cf634a64b2 100644 --- a/synapse/federation/replication.py +++ b/synapse/federation/replication.py @@ -509,10 +509,10 @@ class _TransactionQueue(object): # a transaction in progress. If we do, stick it in the pending_pdus # table and we'll get back to it later. - destinations = [ + destinations = set([ d for d in pdu.destinations if d != self.server_name - ] + ]) logger.debug("Sending to: %s", str(destinations)) diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py index c2f4685c92..3f07b5aa4a 100644 --- a/synapse/handlers/_base.py +++ b/synapse/handlers/_base.py @@ -24,4 +24,5 @@ class BaseHandler(object): self.notifier = hs.get_notifier() self.room_lock = hs.get_room_lock_manager() self.state_handler = hs.get_state_handler() + self.distributor = hs.get_distributor() self.hs = hs diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index 351bb3c084..16bac95331 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -32,6 +32,15 @@ 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 @@ -103,6 +112,13 @@ class FederationHandler(BaseHandler): 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 @@ -152,8 +168,10 @@ class FederationHandler(BaseHandler): yield federation.handle_new_event(new_event) - store_id = yield self.store.persist_event(new_event) - self.notifier.on_new_room_event(new_event, store_id) + # 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( @@ -166,3 +184,10 @@ class FederationHandler(BaseHandler): 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) diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py index 540e114b82..c88cc18788 100644 --- a/synapse/handlers/presence.py +++ b/synapse/handlers/presence.py @@ -142,6 +142,10 @@ 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: @@ -187,6 +191,10 @@ 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") diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py index 049b4884af..5489de841f 100644 --- a/synapse/handlers/room.py +++ b/synapse/handlers/room.py @@ -24,6 +24,7 @@ from synapse.api.events.room import ( RoomConfigEvent ) from synapse.api.streams.event import EventStream, EventsStreamData +from synapse.handlers.presence import PresenceStreamData from synapse.util import stringutils from ._base import BaseHandler @@ -257,21 +258,38 @@ class MessageHandler(BaseHandler): membership_list=[Membership.INVITE, Membership.JOIN] ) - ret = [] + rooms_ret = [] + + now_rooms_token = yield self.store.get_room_events_max_id() + + # FIXME (erikj): Fix this. + presence_stream = PresenceStreamData(self.hs) + now_presence_token = yield presence_stream.max_token() + presence = yield presence_stream.get_rows( + user_id, 0, now_presence_token, None, None + ) + + # FIXME (erikj): We need to not generate this token, + now_token = "%s_%s" % (now_rooms_token, now_presence_token) for event in room_list: d = { "room_id": event.room_id, "membership": event.membership, } - ret.append(d) + + if event.membership == Membership.INVITE: + d["inviter"] = event.user_id + + rooms_ret.append(d) if event.membership != Membership.JOIN: continue try: messages, token = yield self.store.get_recent_events_for_room( event.room_id, - limit=50, + limit=10, + end_token=now_rooms_token, ) d["messages"] = { @@ -279,10 +297,17 @@ class MessageHandler(BaseHandler): "start": token[0], "end": token[1], } + + current_state = yield self.store.get_current_state(event.room_id) + d["state"] = [c.get_dict() for c in current_state] except: logger.exception("Failed to get snapshot") - logger.debug("snapshot_all_rooms returning: %s", ret) + user = self.hs.parse_userid(user_id) + + ret = {"rooms": rooms_ret, "presence": presence[0], "end": now_token} + + # logger.debug("snapshot_all_rooms returning: %s", ret) defer.returnValue(ret) diff --git a/synapse/http/server.py b/synapse/http/server.py index c28d9a33f9..66f966fcaa 100644 --- a/synapse/http/server.py +++ b/synapse/http/server.py @@ -212,8 +212,9 @@ class ContentRepoResource(resource.Resource): """ isLeaf = True - def __init__(self, directory, auth): + def __init__(self, hs, directory, auth): resource.Resource.__init__(self) + self.hs = hs self.directory = directory self.auth = auth @@ -250,7 +251,8 @@ class ContentRepoResource(resource.Resource): file_ext = re.sub("[^a-z]", "", file_ext) suffix += "." + file_ext - file_path = os.path.join(self.directory, prefix + main_part + suffix) + file_name = prefix + main_part + suffix + file_path = os.path.join(self.directory, file_name) logger.info("User %s is uploading a file to path %s", auth_user.to_string(), file_path) @@ -259,8 +261,8 @@ class ContentRepoResource(resource.Resource): attempts = 0 while os.path.exists(file_path): main_part = random_string(24) - file_path = os.path.join(self.directory, - prefix + main_part + suffix) + file_name = prefix + main_part + suffix + file_path = os.path.join(self.directory, file_name) attempts += 1 if attempts > 25: # really? Really? raise SynapseError(500, "Unable to create file.") @@ -272,11 +274,14 @@ class ContentRepoResource(resource.Resource): # servers. # TODO: A little crude here, we could do this better. - filename = request.path.split(self.directory + "/")[1] + filename = request.path.split('/')[-1] # be paranoid filename = re.sub("[^0-9A-z.-_]", "", filename) file_path = self.directory + "/" + filename + + logger.debug("Searching for %s", file_path) + if os.path.isfile(file_path): # filename has the content type base64_contentype = filename.split(".")[1] @@ -304,6 +309,10 @@ class ContentRepoResource(resource.Resource): self._async_render(request) return server.NOT_DONE_YET + def render_OPTIONS(self, request): + respond_with_json_bytes(request, 200, {}, send_cors=True) + return server.NOT_DONE_YET + @defer.inlineCallbacks def _async_render(self, request): try: @@ -313,8 +322,15 @@ class ContentRepoResource(resource.Resource): with open(fname, "wb") as f: f.write(request.content.read()) + + # FIXME (erikj): These should use constants. + file_name = os.path.basename(fname) + url = "http://%s/matrix/content/%s" % ( + self.hs.domain_with_port, file_name + ) + respond_with_json_bytes(request, 200, - json.dumps({"content_token": fname}), + json.dumps({"content_token": url}), send_cors=True) except CodeMessageException as e: diff --git a/synapse/rest/register.py b/synapse/rest/register.py index eb457562b9..f17ec11cf4 100644 --- a/synapse/rest/register.py +++ b/synapse/rest/register.py @@ -33,10 +33,10 @@ class RegisterRestServlet(RestServlet): try: register_json = json.loads(request.content.read()) if "password" in register_json: - password = register_json["password"] + password = register_json["password"].encode("utf-8") if type(register_json["user_id"]) == unicode: - desired_user_id = register_json["user_id"] + desired_user_id = register_json["user_id"].encode("utf-8") if urllib.quote(desired_user_id) != desired_user_id: raise SynapseError( 400, diff --git a/synapse/server.py b/synapse/server.py index d4c2481483..c5b0a32757 100644 --- a/synapse/server.py +++ b/synapse/server.py @@ -159,7 +159,7 @@ class HomeServer(BaseHomeServer): return DataStore(self) def build_event_factory(self): - return EventFactory() + return EventFactory(self) def build_handlers(self): return Handlers(self) diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index 7732906927..d06033b980 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -105,6 +105,11 @@ class DataStore(RoomMemberStore, RoomStore, "processed": True, } + if hasattr(event, "outlier"): + vals["outlier"] = event.outlier + else: + vals["outlier"] = False + if backfilled: if not self.min_token_deferred.called: yield self.min_token_deferred @@ -123,7 +128,7 @@ class DataStore(RoomMemberStore, RoomStore, except: logger.exception( "Failed to persist, probably duplicate: %s", - event_id + event.event_id ) return diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 36cc57c1b8..75aab2d3b9 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -294,6 +294,11 @@ class SQLBaseStore(object): def _parse_event_from_row(self, row_dict): d = copy.deepcopy({k: v for k, v in row_dict.items() if v}) + + d.pop("stream_ordering", None) + d.pop("topological_ordering", None) + d.pop("processed", None) + d.update(json.loads(row_dict["unrecognized_keys"])) d["content"] = json.loads(d["content"]) del d["unrecognized_keys"] diff --git a/synapse/storage/roommember.py b/synapse/storage/roommember.py index 1f8984b6e8..a9a09e1425 100644 --- a/synapse/storage/roommember.py +++ b/synapse/storage/roommember.py @@ -146,7 +146,7 @@ class RoomMemberStore(SQLBaseStore): rows = yield self._execute_and_decode(sql, *where_values) - logger.debug("_get_members_query Got rows %s", rows) + # logger.debug("_get_members_query Got rows %s", rows) results = [self._parse_event_from_row(r) for r in rows] defer.returnValue(results) diff --git a/synapse/storage/schema/im.sql b/synapse/storage/schema/im.sql index ea04261ff0..e92f21ef3b 100644 --- a/synapse/storage/schema/im.sql +++ b/synapse/storage/schema/im.sql @@ -22,9 +22,15 @@ CREATE TABLE IF NOT EXISTS events( content TEXT NOT NULL, unrecognized_keys TEXT, processed BOOL NOT NULL, + outlier BOOL NOT NULL, CONSTRAINT ev_uniq UNIQUE (event_id) ); +CREATE INDEX IF NOT EXISTS events_event_id ON events (event_id); +CREATE INDEX IF NOT EXISTS events_stream_ordering ON events (stream_ordering); +CREATE INDEX IF NOT EXISTS events_topological_ordering ON events (topological_ordering); +CREATE INDEX IF NOT EXISTS events_room_id ON events (room_id); + CREATE TABLE IF NOT EXISTS state_events( event_id TEXT NOT NULL, room_id TEXT NOT NULL, @@ -33,6 +39,12 @@ CREATE TABLE IF NOT EXISTS state_events( prev_state TEXT ); +CREATE UNIQUE INDEX IF NOT EXISTS state_events_event_id ON state_events (event_id); +CREATE INDEX IF NOT EXISTS state_events_room_id ON state_events (room_id); +CREATE INDEX IF NOT EXISTS state_events_type ON state_events (type); +CREATE INDEX IF NOT EXISTS state_events_state_key ON state_events (state_key); + + CREATE TABLE IF NOT EXISTS current_state_events( event_id TEXT NOT NULL, room_id TEXT NOT NULL, @@ -41,6 +53,11 @@ CREATE TABLE IF NOT EXISTS current_state_events( CONSTRAINT curr_uniq UNIQUE (room_id, type, state_key) ON CONFLICT REPLACE ); +CREATE INDEX IF NOT EXISTS curr_events_event_id ON current_state_events (event_id); +CREATE INDEX IF NOT EXISTS current_state_events_room_id ON current_state_events (room_id); +CREATE INDEX IF NOT EXISTS current_state_events_type ON current_state_events (type); +CREATE INDEX IF NOT EXISTS current_state_events_state_key ON current_state_events (state_key); + CREATE TABLE IF NOT EXISTS room_memberships( event_id TEXT NOT NULL, user_id TEXT NOT NULL, @@ -49,6 +66,10 @@ CREATE TABLE IF NOT EXISTS room_memberships( membership TEXT NOT NULL ); +CREATE INDEX IF NOT EXISTS room_memberships_event_id ON room_memberships (event_id); +CREATE INDEX IF NOT EXISTS room_memberships_room_id ON room_memberships (room_id); +CREATE INDEX IF NOT EXISTS room_memberships_user_id ON room_memberships (user_id); + CREATE TABLE IF NOT EXISTS feedback( event_id TEXT NOT NULL, feedback_type TEXT, @@ -77,5 +98,6 @@ CREATE TABLE IF NOT EXISTS rooms( CREATE TABLE IF NOT EXISTS room_hosts( room_id TEXT NOT NULL, - host TEXT NOT NULL + host TEXT NOT NULL, + CONSTRAINT room_hosts_uniq UNIQUE (room_id, host) ON CONFLICT IGNORE ); diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py index e994017bf2..3a17a723fe 100644 --- a/synapse/storage/stream.py +++ b/synapse/storage/stream.py @@ -177,6 +177,7 @@ class StreamStore(SQLBaseStore): "((room_id IN (%(current)s)) OR " "(event_id IN (%(invites)s))) " "AND e.stream_ordering > ? AND e.stream_ordering < ? " + "AND e.outlier = 0 " "ORDER BY stream_ordering ASC LIMIT %(limit)d " ) % { "current": current_room_membership_sql, @@ -224,7 +225,7 @@ class StreamStore(SQLBaseStore): sql = ( "SELECT * FROM events " - "WHERE room_id = ? AND %(bounds)s " + "WHERE outlier = 0 AND room_id = ? AND %(bounds)s " "ORDER BY topological_ordering %(order)s, stream_ordering %(order)s %(limit)s " ) % {"bounds": bounds, "order": order, "limit": limit_str} @@ -249,15 +250,14 @@ class StreamStore(SQLBaseStore): ) @defer.inlineCallbacks - def get_recent_events_for_room(self, room_id, limit, with_feedback=False): + def get_recent_events_for_room(self, room_id, limit, end_token, + with_feedback=False): # TODO (erikj): Handle compressed feedback - end_token = yield self.get_room_events_max_id() - sql = ( "SELECT * FROM events " "WHERE room_id = ? AND stream_ordering <= ? " - "ORDER BY topological_ordering, stream_ordering DESC LIMIT ? " + "ORDER BY topological_ordering DESC, stream_ordering DESC LIMIT ? " ) rows = yield self._execute_and_decode( |