Matrix-Mirrors/anonymousland-synapse
1
0
Fork 0
mirror of https://git.anonymousland.org/anonymousland/synapse.git synced 2025-09-20 00:44:34 -04:00
Code Issues Projects Releases Packages Wiki Activity

Merge branch 'develop' of github.com:matrix-org/synapse into client_server_url_rename

Browse source
This commit is contained in:
Kegan Dougal 2014-08-26 09:26:33 +01:00
commit 47c3a089c5
55 changed files with 2633 additions and 606 deletions
Download patch file Download diff file Expand all files Collapse all files

2
synapse/__init__.py
View file

@ -15,3 +15,5 @@
""" This is a reference implementation of a synapse home server.
"""
__version__ = "0.0.1"

1
synapse/api/events/__init__.py
View file

@ -51,6 +51,7 @@ class SynapseEvent(JsonEncodedObject):
"depth",
"destinations",
"origin",
"outlier",
]
required_keys = [

7
synapse/api/events/factory.py
View file

@ -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:

16
synapse/app/homeserver.py
View file

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

4
synapse/federation/replication.py
View file

@ -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))

1
synapse/handlers/_base.py
View file

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

29
synapse/handlers/federation.py
View file

@ -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)

8
synapse/handlers/presence.py
View file

@ -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")

33
synapse/handlers/room.py
View file

@ -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)

28
synapse/http/server.py
View file

@ -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:

4
synapse/rest/register.py
View file

@ -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,

2
synapse/server.py
View file

@ -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)

7
synapse/storage/__init__.py
View file

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

5
synapse/storage/_base.py
View file

@ -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"]

2
synapse/storage/roommember.py
View file

@ -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)

24
synapse/storage/schema/im.sql
View file

@ -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
);

10
synapse/storage/stream.py
View file

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