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..495149466c 100755
--- a/synapse/app/homeserver.py
+++ b/synapse/app/homeserver.py
@@ -56,7 +56,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 +235,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
@@ -257,7 +257,8 @@ def setup():
hs = SynapseHomeServer(
args.host,
- db_name=db_name
+ 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/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 aa3bf273f7..9cff444779 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/room.py b/synapse/handlers/room.py
index 6229ee9bfa..899b653fb7 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,9 +297,16 @@ 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")
+ 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..d1f99460c1 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,13 @@ 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.hostname, 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/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/schema/im.sql b/synapse/storage/schema/im.sql
index ea04261ff0..39a1ed703e 100644
--- a/synapse/storage/schema/im.sql
+++ b/synapse/storage/schema/im.sql
@@ -22,6 +22,7 @@ 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)
);
diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py
index e994017bf2..87ae961ccd 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,11 +250,10 @@ 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 <= ? "
|
