diff --git a/experiments/test_messaging.py b/experiments/test_messaging.py
new file mode 100644
index 0000000000..9a45425173
--- /dev/null
+++ b/experiments/test_messaging.py
@@ -0,0 +1,380 @@
+# -*- coding: utf-8 -*-
+
+""" This is an example of using the server to server implementation to do a
+basic chat style thing. It accepts commands from stdin and outputs to stdout.
+
+It assumes that ucids are of the form <user>@<domain>, and uses <domain> as
+the address of the remote home server to hit.
+
+Usage:
+ python test_messaging.py <port>
+
+Currently assumes the local address is localhost:<port>
+
+"""
+
+
+from synapse.federation import (
+ ReplicationHandler
+)
+
+from synapse.federation.units import Pdu
+
+from synapse.util import origin_from_ucid
+
+from synapse.app.homeserver import SynapseHomeServer
+
+#from synapse.util.logutils import log_function
+
+from twisted.internet import reactor, defer
+from twisted.python import log
+
+import argparse
+import json
+import logging
+import os
+import re
+
+import cursesio
+import curses.wrapper
+
+
+logger = logging.getLogger("example")
+
+
+def excpetion_errback(failure):
+ logging.exception(failure)
+
+
+class InputOutput(object):
+ """ This is responsible for basic I/O so that a user can interact with
+ the example app.
+ """
+
+ def __init__(self, screen, user):
+ self.screen = screen
+ self.user = user
+
+ def set_home_server(self, server):
+ self.server = server
+
+ def on_line(self, line):
+ """ This is where we process commands.
+ """
+
+ try:
+ m = re.match("^join (\S+)$", line)
+ if m:
+ # The `sender` wants to join a room.
+ room_name, = m.groups()
+ self.print_line("%s joining %s" % (self.user, room_name))
+ self.server.join_room(room_name, self.user, self.user)
+ #self.print_line("OK.")
+ return
+
+ m = re.match("^invite (\S+) (\S+)$", line)
+ if m:
+ # `sender` wants to invite someone to a room
+ room_name, invitee = m.groups()
+ self.print_line("%s invited to %s" % (invitee, room_name))
+ self.server.invite_to_room(room_name, self.user, invitee)
+ #self.print_line("OK.")
+ return
+
+ m = re.match("^send (\S+) (.*)$", line)
+ if m:
+ # `sender` wants to message a room
+ room_name, body = m.groups()
+ self.print_line("%s send to %s" % (self.user, room_name))
+ self.server.send_message(room_name, self.user, body)
+ #self.print_line("OK.")
+ return
+
+ m = re.match("^paginate (\S+)$", line)
+ if m:
+ # we want to paginate a room
+ room_name, = m.groups()
+ self.print_line("paginate %s" % room_name)
+ self.server.paginate(room_name)
+ return
+
+ self.print_line("Unrecognized command")
+
+ except Exception as e:
+ logger.exception(e)
+
+ def print_line(self, text):
+ self.screen.print_line(text)
+
+ def print_log(self, text):
+ self.screen.print_log(text)
+
+
+class IOLoggerHandler(logging.Handler):
+
+ def __init__(self, io):
+ logging.Handler.__init__(self)
+ self.io = io
+
+ def emit(self, record):
+ if record.levelno < logging.WARN:
+ return
+
+ msg = self.format(record)
+ self.io.print_log(msg)
+
+
+class Room(object):
+ """ Used to store (in memory) the current membership state of a room, and
+ which home servers we should send PDUs associated with the room to.
+ """
+ def __init__(self, room_name):
+ self.room_name = room_name
+ self.invited = set()
+ self.participants = set()
+ self.servers = set()
+
+ self.oldest_server = None
+
+ self.have_got_metadata = False
+
+ def add_participant(self, participant):
+ """ Someone has joined the room
+ """
+ self.participants.add(participant)
+ self.invited.discard(participant)
+
+ server = origin_from_ucid(participant)
+ self.servers.add(server)
+
+ if not self.oldest_server:
+ self.oldest_server = server
+
+ def add_invited(self, invitee):
+ """ Someone has been invited to the room
+ """
+ self.invited.add(invitee)
+ self.servers.add(origin_from_ucid(invitee))
+
+
+class HomeServer(ReplicationHandler):
+ """ A very basic home server implentation that allows people to join a
+ room and then invite other people.
+ """
+ def __init__(self, server_name, replication_layer, output):
+ self.server_name = server_name
+ self.replication_layer = replication_layer
+ self.replication_layer.set_handler(self)
+
+ self.joined_rooms = {}
+
+ self.output = output
+
+ def on_receive_pdu(self, pdu):
+ """ We just received a PDU
+ """
+ pdu_type = pdu.pdu_type
+
+ if pdu_type == "sy.room.message":
+ self._on_message(pdu)
+ elif pdu_type == "sy.room.member" and "membership" in pdu.content:
+ if pdu.content["membership"] == "join":
+ self._on_join(pdu.context, pdu.state_key)
+ elif pdu.content["membership"] == "invite":
+ self._on_invite(pdu.origin, pdu.context, pdu.state_key)
+ else:
+ self.output.print_line("#%s (unrec) %s = %s" %
+ (pdu.context, pdu.pdu_type, json.dumps(pdu.content))
+ )
+
+ #def on_state_change(self, pdu):
+ ##self.output.print_line("#%s (state) %s *** %s" %
+ ##(pdu.context, pdu.state_key, pdu.pdu_type)
+ ##)
+
+ #if "joinee" in pdu.content:
+ #self._on_join(pdu.context, pdu.content["joinee"])
+ #elif "invitee" in pdu.content:
+ #self._on_invite(pdu.origin, pdu.context, pdu.content["invitee"])
+
+ def _on_message(self, pdu):
+ """ We received a message
+ """
+ self.output.print_line("#%s %s %s" %
+ (pdu.context, pdu.content["sender"], pdu.content["body"])
+ )
+
+ def _on_join(self, context, joinee):
+ """ Someone has joined a room, either a remote user or a local user
+ """
+ room = self._get_or_create_room(context)
+ room.add_participant(joinee)
+
+ self.output.print_line("#%s %s %s" %
+ (context, joinee, "*** JOINED")
+ )
+
+ def _on_invite(self, origin, context, invitee):
+ """ Someone has been invited
+ """
+ room = self._get_or_create_room(context)
+ room.add_invited(invitee)
+
+ self.output.print_line("#%s %s %s" %
+ (context, invitee, "*** INVITED")
+ )
+
+ if not room.have_got_metadata and origin is not self.server_name:
+ logger.debug("Get room state")
+ self.replication_layer.get_state_for_context(origin, context)
+ room.have_got_metadata = True
+
+ @defer.inlineCallbacks
+ def send_message(self, room_name, sender, body):
+ """ Send a message to a room!
+ """
+ destinations = yield self.get_servers_for_context(room_name)
+
+ try:
+ yield self.replication_layer.send_pdu(
+ Pdu.create_new(
+ context=room_name,
+ pdu_type="sy.room.message",
+ content={"sender": sender, "body": body},
+ origin=self.server_name,
+ destinations=destinations,
+ )
+ )
+ except Exception as e:
+ logger.exception(e)
+
+ @defer.inlineCallbacks
+ def join_room(self, room_name, sender, joinee):
+ """ Join a room!
+ """
+ self._on_join(room_name, joinee)
+
+ destinations = yield self.get_servers_for_context(room_name)
+
+ try:
+ pdu = Pdu.create_new(
+ context=room_name,
+ pdu_type="sy.room.member",
+ is_state=True,
+ state_key=joinee,
+ content={"membership": "join"},
+ origin=self.server_name,
+ destinations=destinations,
+ )
+ yield self.replication_layer.send_pdu(pdu)
+ except Exception as e:
+ logger.exception(e)
+
+ @defer.inlineCallbacks
+ def invite_to_room(self, room_name, sender, invitee):
+ """ Invite someone to a room!
+ """
+ self._on_invite(self.server_name, room_name, invitee)
+
+ destinations = yield self.get_servers_for_context(room_name)
+
+ try:
+ yield self.replication_layer.send_pdu(
+ Pdu.create_new(
+ context=room_name,
+ is_state=True,
+ pdu_type="sy.room.member",
+ state_key=invitee,
+ content={"membership": "invite"},
+ origin=self.server_name,
+ destinations=destinations,
+ )
+ )
+ except Exception as e:
+ logger.exception(e)
+
+ def paginate(self, room_name, limit=5):
+ room = self.joined_rooms.get(room_name)
+
+ if not room:
+ return
+
+ dest = room.oldest_server
+
+ return self.replication_layer.paginate(dest, room_name, limit)
+
+ def _get_room_remote_servers(self, room_name):
+ return [i for i in self.joined_rooms.setdefault(room_name,).servers]
+
+ def _get_or_create_room(self, room_name):
+ return self.joined_rooms.setdefault(room_name, Room(room_name))
+
+ def get_servers_for_context(self, context):
+ return defer.succeed(
+ self.joined_rooms.setdefault(context, Room(context)).servers
+ )
+
+
+def main(stdscr):
+ parser = argparse.ArgumentParser()
+ parser.add_argument('user', type=str)
+ parser.add_argument('-v', '--verbose', action='count')
+ args = parser.parse_args()
+
+ user = args.user
+ server_name = origin_from_ucid(user)
+
+ ## Set up logging ##
+
+ root_logger = logging.getLogger()
+
+ formatter = logging.Formatter('%(asctime)s - %(name)s - %(lineno)d - '
+ '%(levelname)s - %(message)s')
+ if not os.path.exists("logs"):
+ os.makedirs("logs")
+ fh = logging.FileHandler("logs/%s" % user)
+ fh.setFormatter(formatter)
+
+ root_logger.addHandler(fh)
+ root_logger.setLevel(logging.DEBUG)
+
+ # Hack: The only way to get it to stop logging to sys.stderr :(
+ log.theLogPublisher.observers = []
+ observer = log.PythonLoggingObserver()
+ observer.start()
+
+ ## Set up synapse server
+
+ curses_stdio = cursesio.CursesStdIO(stdscr)
+ input_output = InputOutput(curses_stdio, user)
+
+ curses_stdio.set_callback(input_output)
+
+ app_hs = SynapseHomeServer(server_name, db_name="dbs/%s" % user)
+ replication = app_hs.get_replication_layer()
+
+ hs = HomeServer(server_name, replication, curses_stdio)
+
+ input_output.set_home_server(hs)
+
+ ## Add input_output logger
+ io_logger = IOLoggerHandler(input_output)
+ io_logger.setFormatter(formatter)
+ root_logger.addHandler(io_logger)
+
+ ## Start! ##
+
+ try:
+ port = int(server_name.split(":")[1])
+ except:
+ port = 12345
+
+ app_hs.get_http_server().start_listening(port)
+
+ reactor.addReader(curses_stdio)
+
+ reactor.run()
+
+
+if __name__ == "__main__":
+ curses.wrapper(main)
|
