/
server.py
238 lines (174 loc) · 6.36 KB
/
server.py
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
229
230
231
232
233
234
235
236
237
238
"""
orbital server
~~~~~~~~~~~~~~
The server consists of two subprocesses, and several greenlets.
The first process runs the actual web server (which includes a websocket
server). It does this via two greenlets, one which handles the actual web
requests, and the other which handles publishing messages to the websocket
client.
The second process runs the zeromq subscriber, which listens for messages
and republishes them to all websocket server clients.
:copyright: (c) 2012 DISQUS.
:license: Apache License 2.0, see LICENSE for more details.
"""
import json
import gevent
import mimetypes
import os.path
import traceback
import uuid
from gevent import pywsgi, monkey
from gevent_zeromq import zmq
from geventwebsocket.handler import WebSocketHandler
from urlparse import urlparse
try:
from raven.base import Client
raven = Client(os.environ['SENTRY_DSN'])
except:
raven = None
def log_exception(self, e):
if raven:
raven.captureException()
traceback.print_exc()
monkey.patch_all()
ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), 'site'))
class WebsocketPublisher(object):
"""
Represents the websocket publisher, which receives messages
from the feeder, and sends them to the websocket client.
It's important that this is in a separate greenlet from the receive
call as we'll block otherwise, and either not be able to send messages
or not being able to receive future commands.
"""
def __init__(self, subscriber, ws, params=None):
self.subscriber = subscriber
self.ws = ws
self.params = params
def run(self):
ws = self.ws
subscriber = self.subscriber
while True:
params = self.params
message = subscriber.recv()
if params is None:
continue
try:
data = json.loads(message)['post']
except KeyError, e:
log_exception(e)
print 'Invalid data', message
continue
if params.get('domain'):
if not data.get('link'):
continue
domain = urlparse(data['link']).hostname.split('.')
if not any(domain[-len(d):] == d for d in params['domain']):
continue
if params.get('query') and not any(t in data['title'].lower() for t in params['query']):
continue
ws.send(message)
def run_publisher():
"""
Pulls data from postevent and publishes to local subscribers
"""
print "Publisher running, waiting for data.."
context = zmq.Context()
publisher = context.socket(zmq.PUB)
publisher.bind('tcp://127.0.0.1:5555')
server = context.socket(zmq.PULL)
server.bind('tcp://127.0.0.1:5556')
while True:
gevent.sleep(0)
try:
message = server.recv()
publisher.send(message)
except Exception, e:
log_exception(e)
raise
publisher.close()
def run_websockets():
"""
Runs the websocket server (which also handles normal HTTP connections for serving
the static assets).
"""
clients = set()
def handle_ws(ws, environ):
context = zmq.Context()
subscriber = context.socket(zmq.SUB)
subscriber.connect("tcp://127.0.0.1:5555")
subscriber.setsockopt(zmq.SUBSCRIBE, "")
client_id = uuid.uuid4().hex
clients.add(client_id)
publisher = WebsocketPublisher(subscriber, ws)
publisher_thread = gevent.spawn(publisher.run)
print "[%s] Client connected %r (%s clients total)" % (client_id, environ['REMOTE_ADDR'], len(clients), )
try:
while True:
gevent.sleep(0)
message = ws.receive()
if not message:
return
message_bits = message.split(' ', 1)
cmd = message_bits[0]
if cmd == 'OK':
continue
if cmd == 'SUB':
if len(message_bits) == 2:
args = message_bits[-1]
else:
args = ''
args = args.strip()
if args == '':
args = '*'
if args == '*':
params = {}
else:
params = dict(a.split('=') for a in args.split('&'))
for key, value in params.iteritems():
params[key] = filter(bool, map(lambda x: x.strip().lower(), value.split(' OR ')))
if key == 'domain':
params[key] = [d.split('.') for d in params[key]]
publisher.params = params
print "[%s] Subscription established %s" % (client_id, params)
finally:
publisher_thread.kill()
subscriber.close()
clients.remove(client_id)
print "[%s] Client disconnected" % client_id
def handle(environ, start_response):
path_info = environ['PATH_INFO']
if path_info.endswith('/'):
path_info += 'index.html'
path = os.path.abspath(os.path.join(ROOT, path_info[1:]))
if not path.startswith(ROOT) or not os.path.exists(path):
start_response("404 Not Found", [])
return []
content_type = mimetypes.guess_type(path)[0]
start_response("200 OK", [('Content-Type', content_type)])
return [open(path, 'r').read()]
def app(environ, start_response):
try:
if environ['PATH_INFO'] == '/' and 'wsgi.websocket' in environ:
return handle_ws(environ['wsgi.websocket'], environ)
return handle(environ, start_response)
except Exception, e:
log_exception(e)
raise
context = zmq.Context()
server = context.socket(zmq.PULL)
server.connect('tcp://127.0.0.1:5556')
host = '0.0.0.0'
port = 7000
print "Listening for connections on %s:%s" % (host, port)
server = pywsgi.WSGIServer((host, port), app,
handler_class=WebSocketHandler)
server.serve_forever()
def main():
procs = [
gevent.spawn(run_publisher),
gevent.spawn(run_websockets),
]
for p in procs:
p.join()
if __name__ == "__main__":
main()