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(
|
