795d316220be48221c95ce0298df75c12f8fb6d3
[OpenSRF.git] / src / python / osrf / http_translator.py
1 import sys, os, time, md5, random
2 from mod_python import apache, util
3 import osrf.system, osrf.cache, osrf.json, osrf.conf, osrf.net, osrf.log
4 from osrf.const import OSRF_MESSAGE_TYPE_DISCONNECT, OSRF_MESSAGE_TYPE_CONNECT, \
5     OSRF_STATUS_CONTINUE, OSRF_STATUS_TIMEOUT, OSRF_MESSAGE_TYPE_STATUS
6
7
8 ''' 
9 Proof of concept OpenSRF-HTTP multipart streaming gateway.
10
11 Example Apache mod_python config:
12
13 <Location /osrf-http-translator>
14    SetHandler mod_python
15    PythonPath "['/path/to/osrf-python'] + sys.path"
16    PythonHandler osrf.http_translator
17    PythonOption OSRF_CONFIG /path/to/opensrf_core.xml
18    PythonOption OSRF_CONFIG_CONTEXT config.gateway
19    PythonOption OSRF_CACHE_SERVERS 127.0.0.1:11211
20    # testing only
21    PythonAutoReload On
22 </Location>
23 '''
24
25 OSRF_HTTP_HEADER_TO = 'X-OpenSRF-to'
26 OSRF_HTTP_HEADER_XID = 'X-OpenSRF-thread'
27 OSRF_HTTP_HEADER_FROM = 'X-OpenSRF-from'
28 OSRF_HTTP_HEADER_THREAD = 'X-OpenSRF-thread'
29 OSRF_HTTP_HEADER_TIMEOUT = 'X-OpenSRF-timeout'
30 OSRF_HTTP_HEADER_SERVICE = 'X-OpenSRF-service'
31 OSRF_HTTP_HEADER_MULTIPART = 'X-OpenSRF-multipart'
32 MULTIPART_CONTENT_TYPE = 'multipart/x-mixed-replace;boundary="%s"'
33 JSON_CONTENT_TYPE = 'text/plain'
34 CACHE_TIME = 300
35
36 ROUTER_NAME = None
37 OSRF_DOMAIN = None
38
39 # If DEBUG_WRITE = True, all data sent to the client is also written
40 # to stderr (apache error log)
41 DEBUG_WRITE = False
42
43 def _dbg(msg):
44     ''' testing only '''
45     sys.stderr.write("%s\n\n" % str(msg))
46     sys.stderr.flush()
47
48
49 INIT_COMPLETE = False
50 def child_init(req):
51     ''' At time of writing, mod_python doesn't support a child_init handler,
52         so this function is called once per process to initialize 
53         the opensrf connection '''
54
55     global INIT_COMPLETE, ROUTER_NAME, OSRF_DOMAIN
56     if INIT_COMPLETE: 
57         return
58
59     # Apache complains with: UnboundLocalError: local variable 'osrf' referenced before assignment
60     # if the following import line is removed, even though its also at the top of the file...
61     import osrf.system 
62
63     ops = req.get_options()
64     conf = ops['OSRF_CONFIG']
65     ctxt = ops.get('OSRF_CONFIG_CONTEXT') or 'opensrf'
66     osrf.system.System.net_connect(config_file=conf, config_context=ctxt)
67
68     ROUTER_NAME = osrf.conf.get('router_name')
69     OSRF_DOMAIN = osrf.conf.get('domain')
70     INIT_COMPLETE = True
71
72     servers = ops.get('OSRF_CACHE_SERVERS')
73     if servers:
74         servers = servers.split(',')
75     else:
76         # no cache servers configured, see if we can talk to the settings server
77         import osrf.set
78         servers = osrf.set.get('cache.global.servers.server')
79         if not isinstance(servers, list):
80             servers = [servers]
81
82     osrf.cache.CacheClient.connect(servers)
83
84
85 def handler(req):
86     ''' Create the translator and tell it to process the request. '''
87     child_init(req)
88     return HTTPTranslator(req).process()
89
90 class HTTPTranslator(object):
91     def __init__(self, apreq):
92
93         self.apreq = apreq
94         if apreq.header_only: 
95             return
96
97         try:
98             post = util.parse_qsl(apreq.read(int(apreq.headers_in['Content-length'])))
99             self.body = [d for d in post if d[0] == 'osrf-msg'][0][1]
100         except: 
101             self.body = None
102             return
103
104         self.messages = []
105         self.complete = False
106         self.handle = osrf.net.get_network_handle()
107         self.handle.set_receive_callback(None)
108
109         self.recipient = apreq.headers_in.get(OSRF_HTTP_HEADER_TO)
110         self.service = apreq.headers_in.get(OSRF_HTTP_HEADER_SERVICE)
111         self.thread = apreq.headers_in.get(OSRF_HTTP_HEADER_THREAD) or \
112             "%s%s" % (os.getpid(), time.time())
113         self.timeout = apreq.headers_in.get(OSRF_HTTP_HEADER_TIMEOUT) or 1200
114         self.multipart = str(
115             apreq.headers_in.get(OSRF_HTTP_HEADER_MULTIPART)).lower() == 'true'
116         self.connect_only = False
117         self.disconnect_only = False
118
119         # generate a random multipart delimiter
120         mpart = md5.new()
121         mpart.update("%f%d%d" % (time.time(), os.getpid(), \
122             random.randint(100, 10000000)))
123         self.delim = mpart.hexdigest()
124         self.remote_host = self.apreq.get_remote_host(apache.REMOTE_NOLOOKUP)
125         self.cache = osrf.cache.CacheClient()
126
127
128     def process(self):
129
130         if self.apreq.header_only: 
131             return apache.OK
132         if not self.body:
133             return apache.HTTP_BAD_REQUEST
134         if not self.set_to_addr():
135             return apache.HTTP_BAD_REQUEST
136         if not self.parse_request():
137             return apache.HTTP_BAD_REQUEST
138
139         while self.handle.recv(0):
140             pass # drop stale messages
141
142
143         net_msg = osrf.net.NetworkMessage(
144             recipient=self.recipient, thread=self.thread, body=self.body)
145         self.handle.send(net_msg)
146
147         if self.disconnect_only:
148             osrf.log.log_debug("exiting early on DISCONNECT")
149             return apache.OK
150
151         first_write = True
152         while not self.complete:
153
154             net_msg = self.handle.recv(self.timeout)
155             if not net_msg: 
156                 return apache.GATEWAY_TIME_OUT
157
158             if not self.check_status(net_msg):
159                 continue 
160
161             if first_write:
162                 self.init_headers(net_msg)
163                 first_write = False
164
165             if self.multipart:
166                 self.respond_chunk(net_msg)
167                 if self.connect_only:
168                     break
169             else:
170                 self.messages.append(net_msg.body)
171
172                 # condense the sets of arrays into a single array of messages
173                 if self.complete or self.connect_only:
174                     json = self.messages.pop(0)
175                     while len(self.messages) > 0:
176                         msg = self.messages.pop(0)
177                         json = "%s,%s" % (json[0:len(json)-1], msg[1:])
178                         
179                     self.write("%s" % json)
180
181
182         return apache.OK
183
184     def parse_request(self):
185         '''
186         If this is solely a DISCONNECT message, we set self.disconnect_only
187         to true
188         @return True if the body parses correctly, False otherwise
189         '''
190         osrf_msgs = osrf.json.to_object(self.body)
191         if not osrf_msgs:
192             return False
193         
194         if len(osrf_msgs) == 1:
195             if osrf_msgs[0].type() == OSRF_MESSAGE_TYPE_CONNECT:
196                 self.connect_only = True
197             elif osrf_msgs[0].type() == OSRF_MESSAGE_TYPE_DISCONNECT:
198                 self.disconnect_only = True
199
200         return True
201
202
203     def set_to_addr(self):
204         ''' Determines the TO address.  Returns false if 
205             the address is missing or ambiguous. 
206             Also returns false if an explicit TO is specified and the
207             thread/IP/TO combination is not found in the session cache
208             '''
209         if self.service:
210             if self.recipient:
211                 osrf.log.log_warn("specifying both SERVICE and TO is not allowed")
212                 return False
213             self.recipient = "%s@%s/%s" % \
214                 (ROUTER_NAME, OSRF_DOMAIN, self.service)
215             return True
216         else:
217             if self.recipient:
218                 # If the client specifies a specific TO address, verify it's
219                 # the same address that was cached with the previous request.  
220                 obj = self.cache.get(self.thread)
221                 if obj and obj['ip'] == self.remote_host and \
222                     obj['jid'] == self.recipient:
223                     return True
224         osrf.log.log_warn("client [%s] attempted to send directly "
225             "[%s] without a session" % (self.remote_host, self.recipient))
226         return False
227
228         
229     def init_headers(self, net_msg):
230         self.apreq.headers_out[OSRF_HTTP_HEADER_FROM] = net_msg.sender
231         self.apreq.headers_out[OSRF_HTTP_HEADER_THREAD] = self.thread
232         if self.multipart:
233             self.apreq.content_type = MULTIPART_CONTENT_TYPE % self.delim
234             self.write("--%s\n" % self.delim)
235         else:
236             self.apreq.content_type = JSON_CONTENT_TYPE
237         self.cache.put(self.thread, \
238             {'ip':self.remote_host, 'jid': net_msg.sender}, CACHE_TIME)
239
240         osrf.log.log_debug("caching session [%s] for host [%s] and server "
241             " drone [%s]" % (self.thread, self.remote_host, net_msg.sender))
242
243
244
245     def check_status(self, net_msg): 
246         ''' Checks the status of the server response. 
247             If we received a timeout message, we drop it.
248             if it's any other non-continue status, we mark this session as
249             complete and carry on.
250             @return False if there is no data to return to the caller 
251             (dropped message, eg. timeout), True otherwise '''
252
253         osrf_msgs = osrf.json.to_object(net_msg.body)
254         last_msg = osrf_msgs.pop()
255
256         if last_msg.type() == OSRF_MESSAGE_TYPE_STATUS:
257             code = int(last_msg.payload().statusCode())
258
259             if code == OSRF_STATUS_TIMEOUT:
260                 osrf.log.log_debug("removing cached session [%s] and "
261                     "dropping TIMEOUT message" % net_msg.thread)
262                 self.cache.delete(net_msg.thread)
263                 return False 
264
265             if code != OSRF_STATUS_CONTINUE:
266                 self.complete = True
267
268         return True
269
270
271     def respond_chunk(self, resp):
272         ''' Writes a single multipart-delimited chunk of data '''
273
274         self.write("Content-type: %s\n\n" % JSON_CONTENT_TYPE)
275         self.write("%s\n\n" % resp.body)
276         if self.complete:
277             self.write("--%s--\n" % self.delim)
278         else:
279             self.write("--%s\n" % self.delim)
280         self.apreq.flush()
281
282     def write(self, msg):
283         ''' Writes data to the client stream. '''
284
285         if DEBUG_WRITE:
286             sys.stderr.write(msg)
287             sys.stderr.flush()
288         self.apreq.write(msg)
289             
290