1
0

test_messaging.py 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394
  1. # -*- coding: utf-8 -*-
  2. # Copyright 2014-2016 OpenMarket Ltd
  3. #
  4. # Licensed under the Apache License, Version 2.0 (the "License");
  5. # you may not use this file except in compliance with the License.
  6. # You may obtain a copy of the License at
  7. #
  8. # http://www.apache.org/licenses/LICENSE-2.0
  9. #
  10. # Unless required by applicable law or agreed to in writing, software
  11. # distributed under the License is distributed on an "AS IS" BASIS,
  12. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13. # See the License for the specific language governing permissions and
  14. # limitations under the License.
  15. """ This is an example of using the server to server implementation to do a
  16. basic chat style thing. It accepts commands from stdin and outputs to stdout.
  17. It assumes that ucids are of the form <user>@<domain>, and uses <domain> as
  18. the address of the remote home server to hit.
  19. Usage:
  20. python test_messaging.py <port>
  21. Currently assumes the local address is localhost:<port>
  22. """
  23. from synapse.federation import (
  24. ReplicationHandler
  25. )
  26. from synapse.federation.units import Pdu
  27. from synapse.util import origin_from_ucid
  28. from synapse.app.homeserver import SynapseHomeServer
  29. #from synapse.util.logutils import log_function
  30. from twisted.internet import reactor, defer
  31. from twisted.python import log
  32. import argparse
  33. import json
  34. import logging
  35. import os
  36. import re
  37. import cursesio
  38. import curses.wrapper
  39. logger = logging.getLogger("example")
  40. def excpetion_errback(failure):
  41. logging.exception(failure)
  42. class InputOutput(object):
  43. """ This is responsible for basic I/O so that a user can interact with
  44. the example app.
  45. """
  46. def __init__(self, screen, user):
  47. self.screen = screen
  48. self.user = user
  49. def set_home_server(self, server):
  50. self.server = server
  51. def on_line(self, line):
  52. """ This is where we process commands.
  53. """
  54. try:
  55. m = re.match("^join (\S+)$", line)
  56. if m:
  57. # The `sender` wants to join a room.
  58. room_name, = m.groups()
  59. self.print_line("%s joining %s" % (self.user, room_name))
  60. self.server.join_room(room_name, self.user, self.user)
  61. #self.print_line("OK.")
  62. return
  63. m = re.match("^invite (\S+) (\S+)$", line)
  64. if m:
  65. # `sender` wants to invite someone to a room
  66. room_name, invitee = m.groups()
  67. self.print_line("%s invited to %s" % (invitee, room_name))
  68. self.server.invite_to_room(room_name, self.user, invitee)
  69. #self.print_line("OK.")
  70. return
  71. m = re.match("^send (\S+) (.*)$", line)
  72. if m:
  73. # `sender` wants to message a room
  74. room_name, body = m.groups()
  75. self.print_line("%s send to %s" % (self.user, room_name))
  76. self.server.send_message(room_name, self.user, body)
  77. #self.print_line("OK.")
  78. return
  79. m = re.match("^backfill (\S+)$", line)
  80. if m:
  81. # we want to backfill a room
  82. room_name, = m.groups()
  83. self.print_line("backfill %s" % room_name)
  84. self.server.backfill(room_name)
  85. return
  86. self.print_line("Unrecognized command")
  87. except Exception as e:
  88. logger.exception(e)
  89. def print_line(self, text):
  90. self.screen.print_line(text)
  91. def print_log(self, text):
  92. self.screen.print_log(text)
  93. class IOLoggerHandler(logging.Handler):
  94. def __init__(self, io):
  95. logging.Handler.__init__(self)
  96. self.io = io
  97. def emit(self, record):
  98. if record.levelno < logging.WARN:
  99. return
  100. msg = self.format(record)
  101. self.io.print_log(msg)
  102. class Room(object):
  103. """ Used to store (in memory) the current membership state of a room, and
  104. which home servers we should send PDUs associated with the room to.
  105. """
  106. def __init__(self, room_name):
  107. self.room_name = room_name
  108. self.invited = set()
  109. self.participants = set()
  110. self.servers = set()
  111. self.oldest_server = None
  112. self.have_got_metadata = False
  113. def add_participant(self, participant):
  114. """ Someone has joined the room
  115. """
  116. self.participants.add(participant)
  117. self.invited.discard(participant)
  118. server = origin_from_ucid(participant)
  119. self.servers.add(server)
  120. if not self.oldest_server:
  121. self.oldest_server = server
  122. def add_invited(self, invitee):
  123. """ Someone has been invited to the room
  124. """
  125. self.invited.add(invitee)
  126. self.servers.add(origin_from_ucid(invitee))
  127. class HomeServer(ReplicationHandler):
  128. """ A very basic home server implentation that allows people to join a
  129. room and then invite other people.
  130. """
  131. def __init__(self, server_name, replication_layer, output):
  132. self.server_name = server_name
  133. self.replication_layer = replication_layer
  134. self.replication_layer.set_handler(self)
  135. self.joined_rooms = {}
  136. self.output = output
  137. def on_receive_pdu(self, pdu):
  138. """ We just received a PDU
  139. """
  140. pdu_type = pdu.pdu_type
  141. if pdu_type == "sy.room.message":
  142. self._on_message(pdu)
  143. elif pdu_type == "sy.room.member" and "membership" in pdu.content:
  144. if pdu.content["membership"] == "join":
  145. self._on_join(pdu.context, pdu.state_key)
  146. elif pdu.content["membership"] == "invite":
  147. self._on_invite(pdu.origin, pdu.context, pdu.state_key)
  148. else:
  149. self.output.print_line("#%s (unrec) %s = %s" %
  150. (pdu.context, pdu.pdu_type, json.dumps(pdu.content))
  151. )
  152. #def on_state_change(self, pdu):
  153. ##self.output.print_line("#%s (state) %s *** %s" %
  154. ##(pdu.context, pdu.state_key, pdu.pdu_type)
  155. ##)
  156. #if "joinee" in pdu.content:
  157. #self._on_join(pdu.context, pdu.content["joinee"])
  158. #elif "invitee" in pdu.content:
  159. #self._on_invite(pdu.origin, pdu.context, pdu.content["invitee"])
  160. def _on_message(self, pdu):
  161. """ We received a message
  162. """
  163. self.output.print_line("#%s %s %s" %
  164. (pdu.context, pdu.content["sender"], pdu.content["body"])
  165. )
  166. def _on_join(self, context, joinee):
  167. """ Someone has joined a room, either a remote user or a local user
  168. """
  169. room = self._get_or_create_room(context)
  170. room.add_participant(joinee)
  171. self.output.print_line("#%s %s %s" %
  172. (context, joinee, "*** JOINED")
  173. )
  174. def _on_invite(self, origin, context, invitee):
  175. """ Someone has been invited
  176. """
  177. room = self._get_or_create_room(context)
  178. room.add_invited(invitee)
  179. self.output.print_line("#%s %s %s" %
  180. (context, invitee, "*** INVITED")
  181. )
  182. if not room.have_got_metadata and origin is not self.server_name:
  183. logger.debug("Get room state")
  184. self.replication_layer.get_state_for_context(origin, context)
  185. room.have_got_metadata = True
  186. @defer.inlineCallbacks
  187. def send_message(self, room_name, sender, body):
  188. """ Send a message to a room!
  189. """
  190. destinations = yield self.get_servers_for_context(room_name)
  191. try:
  192. yield self.replication_layer.send_pdu(
  193. Pdu.create_new(
  194. context=room_name,
  195. pdu_type="sy.room.message",
  196. content={"sender": sender, "body": body},
  197. origin=self.server_name,
  198. destinations=destinations,
  199. )
  200. )
  201. except Exception as e:
  202. logger.exception(e)
  203. @defer.inlineCallbacks
  204. def join_room(self, room_name, sender, joinee):
  205. """ Join a room!
  206. """
  207. self._on_join(room_name, joinee)
  208. destinations = yield self.get_servers_for_context(room_name)
  209. try:
  210. pdu = Pdu.create_new(
  211. context=room_name,
  212. pdu_type="sy.room.member",
  213. is_state=True,
  214. state_key=joinee,
  215. content={"membership": "join"},
  216. origin=self.server_name,
  217. destinations=destinations,
  218. )
  219. yield self.replication_layer.send_pdu(pdu)
  220. except Exception as e:
  221. logger.exception(e)
  222. @defer.inlineCallbacks
  223. def invite_to_room(self, room_name, sender, invitee):
  224. """ Invite someone to a room!
  225. """
  226. self._on_invite(self.server_name, room_name, invitee)
  227. destinations = yield self.get_servers_for_context(room_name)
  228. try:
  229. yield self.replication_layer.send_pdu(
  230. Pdu.create_new(
  231. context=room_name,
  232. is_state=True,
  233. pdu_type="sy.room.member",
  234. state_key=invitee,
  235. content={"membership": "invite"},
  236. origin=self.server_name,
  237. destinations=destinations,
  238. )
  239. )
  240. except Exception as e:
  241. logger.exception(e)
  242. def backfill(self, room_name, limit=5):
  243. room = self.joined_rooms.get(room_name)
  244. if not room:
  245. return
  246. dest = room.oldest_server
  247. return self.replication_layer.backfill(dest, room_name, limit)
  248. def _get_room_remote_servers(self, room_name):
  249. return [i for i in self.joined_rooms.setdefault(room_name,).servers]
  250. def _get_or_create_room(self, room_name):
  251. return self.joined_rooms.setdefault(room_name, Room(room_name))
  252. def get_servers_for_context(self, context):
  253. return defer.succeed(
  254. self.joined_rooms.setdefault(context, Room(context)).servers
  255. )
  256. def main(stdscr):
  257. parser = argparse.ArgumentParser()
  258. parser.add_argument('user', type=str)
  259. parser.add_argument('-v', '--verbose', action='count')
  260. args = parser.parse_args()
  261. user = args.user
  262. server_name = origin_from_ucid(user)
  263. ## Set up logging ##
  264. root_logger = logging.getLogger()
  265. formatter = logging.Formatter('%(asctime)s - %(name)s - %(lineno)d - '
  266. '%(levelname)s - %(message)s')
  267. if not os.path.exists("logs"):
  268. os.makedirs("logs")
  269. fh = logging.FileHandler("logs/%s" % user)
  270. fh.setFormatter(formatter)
  271. root_logger.addHandler(fh)
  272. root_logger.setLevel(logging.DEBUG)
  273. # Hack: The only way to get it to stop logging to sys.stderr :(
  274. log.theLogPublisher.observers = []
  275. observer = log.PythonLoggingObserver()
  276. observer.start()
  277. ## Set up synapse server
  278. curses_stdio = cursesio.CursesStdIO(stdscr)
  279. input_output = InputOutput(curses_stdio, user)
  280. curses_stdio.set_callback(input_output)
  281. app_hs = SynapseHomeServer(server_name, db_name="dbs/%s" % user)
  282. replication = app_hs.get_replication_layer()
  283. hs = HomeServer(server_name, replication, curses_stdio)
  284. input_output.set_home_server(hs)
  285. ## Add input_output logger
  286. io_logger = IOLoggerHandler(input_output)
  287. io_logger.setFormatter(formatter)
  288. root_logger.addHandler(io_logger)
  289. ## Start! ##
  290. try:
  291. port = int(server_name.split(":")[1])
  292. except:
  293. port = 12345
  294. app_hs.get_http_server().start_listening(port)
  295. reactor.addReader(curses_stdio)
  296. reactor.run()
  297. if __name__ == "__main__":
  298. curses.wrapper(main)