|
| 1 | +import json |
| 2 | +import uuid |
| 3 | + |
| 4 | +from twisted.internet import threads |
| 5 | +from twisted.internet.protocol import Protocol |
| 6 | + |
| 7 | +from hendrix.facilities.resources import NamedResource |
| 8 | +from .messaging import hxdispatcher |
| 9 | + |
| 10 | + |
| 11 | +def send_django_signal(transport, data): |
| 12 | + from .signals import message_signal |
| 13 | + message_signal.send(None, dispatcher=transport, data=data) |
| 14 | + |
| 15 | + |
| 16 | +class MessageHandlerProtocol(Protocol): |
| 17 | + """ |
| 18 | + A basic protocol for socket messaging |
| 19 | + using a hendrix messaging dispatcher to handle |
| 20 | + addressing messages to active sockets from |
| 21 | + different contexts |
| 22 | + """ |
| 23 | + dispatcher = hxdispatcher |
| 24 | + guid = None |
| 25 | + |
| 26 | + def dataReceived(self, data): |
| 27 | + |
| 28 | + """ |
| 29 | + Takes "data" which we assume is json encoded |
| 30 | + If data has a subject_id attribute, we pass that to the dispatcher |
| 31 | + as the subject_id so it will get carried through into any |
| 32 | + return communications and be identifiable to the client |
| 33 | +
|
| 34 | + falls back to just passing the message along... |
| 35 | +
|
| 36 | + """ |
| 37 | + try: |
| 38 | + address = self.guid |
| 39 | + data = json.loads(data) |
| 40 | + threads.deferToThread(send_signal, self.dispatcher, data) |
| 41 | + |
| 42 | + if 'hx_subscribe' in data: |
| 43 | + return self.dispatcher.subscribe(self.transport, data) |
| 44 | + |
| 45 | + if 'address' in data: |
| 46 | + address = data['address'] |
| 47 | + else: |
| 48 | + address = self.guid |
| 49 | + |
| 50 | + self.dispatcher.send(address, data) |
| 51 | + |
| 52 | + except Exception as e: |
| 53 | + raise |
| 54 | + self.dispatcher.send( |
| 55 | + self.guid, |
| 56 | + {'message': data, 'error': str(e)} |
| 57 | + ) |
| 58 | + |
| 59 | + def connectionMade(self): |
| 60 | + """ |
| 61 | + establish the address of this new connection and add it to the list of |
| 62 | + sockets managed by the dispatcher |
| 63 | +
|
| 64 | + reply to the transport with a "setup_connection" notice |
| 65 | + containing the recipient's address for use by the client as a return |
| 66 | + address for future communications |
| 67 | + """ |
| 68 | + self.transport.uid = str(uuid.uuid1()) |
| 69 | + |
| 70 | + self.guid = self.dispatcher.add(self.transport) |
| 71 | + self.dispatcher.send(self.guid, {'setup_connection': self.guid}) |
| 72 | + |
| 73 | + def connectionLost(self, something): |
| 74 | + "clean up the no longer useful socket in the dispatcher" |
| 75 | + self.dispatcher.remove(self.transport) |
| 76 | + |
| 77 | + |
| 78 | +MessageResource = NamedResource('messages') |
0 commit comments