/usr/lib/python2.7/dist-packages/geventwebsocket/protocols/wamp.py is in python-gevent-websocket 0.9.3-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 | import inspect
import random
import string
import types
try:
import ujson as json
except ImportError:
try:
import simplejson as json
except ImportError:
import json
from ..exceptions import WebSocketError
from .base import BaseProtocol
def export_rpc(arg=None):
if isinstance(arg, types.FunctionType):
arg._rpc = arg.__name__
return arg
def serialize(data):
return json.dumps(data)
class Prefixes(object):
def __init__(self):
self.prefixes = {}
def add(self, prefix, uri):
self.prefixes[prefix] = uri
def resolve(self, curie_or_uri):
if "http://" in curie_or_uri:
return curie_or_uri
elif ':' in curie_or_uri:
prefix, proc = curie_or_uri.split(':', 1)
return self.prefixes[prefix] + proc
else:
raise Exception(curie_or_uri)
class RemoteProcedures(object):
def __init__(self):
self.calls = {}
def register_procedure(self, uri, proc):
self.calls[uri] = proc
def register_object(self, uri, obj):
for k in inspect.getmembers(obj, inspect.ismethod):
if '_rpc' in k[1].__dict__:
proc_uri = uri + k[1]._rpc
self.calls[proc_uri] = (obj, k[1])
def call(self, uri, args):
if uri in self.calls:
proc = self.calls[uri]
# Do the correct call whether it's a function or instance method.
if isinstance(proc, tuple):
if proc[1].__self__ is None:
# Create instance of object and call method
return proc[1](proc[0](), *args)
else:
# Call bound method on instance
return proc[1](*args)
else:
return self.calls[uri](*args)
else:
raise Exception("no such uri '{}'".format(uri))
class Channels(object):
def __init__(self):
self.channels = {}
def create(self, uri, prefix_matching=False):
if uri not in self.channels:
self.channels[uri] = []
# TODO: implement prefix matching
def subscribe(self, uri, client):
if uri in self.channels:
self.channels[uri].append(client)
def unsubscribe(self, uri, client):
if uri not in self.channels:
return
client_index = self.channels[uri].index(client)
self.channels[uri].pop(client_index)
if len(self.channels[uri]) == 0:
del self.channels[uri]
def publish(self, uri, event, exclude=None, eligible=None):
if uri not in self.channels:
return
# TODO: exclude & eligible
msg = [WampProtocol.MSG_EVENT, uri, event]
for client in self.channels[uri]:
try:
client.ws.send(serialize(msg))
except WebSocketError:
# Seems someone didn't unsubscribe before disconnecting
self.channels[uri].remove(client)
class WampProtocol(BaseProtocol):
MSG_WELCOME = 0
MSG_PREFIX = 1
MSG_CALL = 2
MSG_CALL_RESULT = 3
MSG_CALL_ERROR = 4
MSG_SUBSCRIBE = 5
MSG_UNSUBSCRIBE = 6
MSG_PUBLISH = 7
MSG_EVENT = 8
PROTOCOL_NAME = "wamp"
def __init__(self, *args, **kwargs):
self.procedures = RemoteProcedures()
self.prefixes = Prefixes()
self.session_id = ''.join(
[random.choice(string.digits + string.letters)
for i in xrange(16)])
super(WampProtocol, self).__init__(*args, **kwargs)
def register_procedure(self, *args, **kwargs):
self.procedures.register_procedure(*args, **kwargs)
def register_object(self, *args, **kwargs):
self.procedures.register_object(*args, **kwargs)
def register_pubsub(self, *args, **kwargs):
if not hasattr(self.server, 'channels'):
self.server.channels = Channels()
self.server.channels.create(*args, **kwargs)
def do_handshake(self):
from geventwebsocket import get_version
welcome = [
self.MSG_WELCOME,
self.session_id,
1,
'gevent-websocket/' + get_version()
]
self.app.ws.send(serialize(welcome))
def rpc_call(self, data):
call_id, curie_or_uri = data[1:3]
args = data[3:]
if not isinstance(call_id, (str, unicode)):
raise Exception()
if not isinstance(curie_or_uri, (str, unicode)):
raise Exception()
uri = self.prefixes.resolve(curie_or_uri)
try:
result = self.procedures.call(uri, args)
result_msg = [self.MSG_CALL_RESULT, call_id, result]
except Exception, e:
result_msg = [self.MSG_CALL_ERROR,
call_id, 'http://TODO#generic',
str(type(e)), str(e)]
self.app.on_message(serialize(result_msg))
def pubsub_action(self, data):
action = data[0]
curie_or_uri = data[1]
if not isinstance(action, int):
raise Exception()
if not isinstance(curie_or_uri, (str, unicode)):
raise Exception()
uri = self.prefixes.resolve(curie_or_uri)
if action == self.MSG_SUBSCRIBE and len(data) == 2:
self.server.channels.subscribe(data[1], self.handler.active_client)
elif action == self.MSG_UNSUBSCRIBE and len(data) == 2:
self.server.channels.unsubscribe(
data[1], self.handler.active_client)
elif action == self.MSG_PUBLISH and len(data) >= 3:
payload = data[2] if len(data) >= 3 else None
exclude = data[3] if len(data) >= 4 else None
eligible = data[4] if len(data) >= 5 else None
self.server.channels.publish(uri, payload, exclude, eligible)
def on_open(self):
self.app.on_open()
self.do_handshake()
def on_message(self, message):
data = json.loads(message)
if not isinstance(data, list):
raise Exception('incoming data is no list')
if data[0] == self.MSG_PREFIX and len(data) == 3:
prefix, uri = data[1:3]
self.prefixes.add(prefix, uri)
elif data[0] == self.MSG_CALL and len(data) >= 3:
return self.rpc_call(data)
elif data[0] in (self.MSG_SUBSCRIBE, self.MSG_UNSUBSCRIBE,
self.MSG_PUBLISH):
return self.pubsub_action(data)
else:
raise Exception("Unknown call")
|