added the ability to wait forever by passing <0 to recv. explicitly setting sender...
[OpenSRF.git] / src / python / osrf / ses.py
1 # -----------------------------------------------------------------------
2 # Copyright (C) 2007  Georgia Public Library Service
3 # Bill Erickson <billserickson@gmail.com>
4
5 # This program is free software; you can redistribute it and/or
6 # modify it under the terms of the GNU General Public License
7 # as published by the Free Software Foundation; either version 2
8 # of the License, or (at your option) any later version.
9
10 # This program is distributed in the hope that it will be useful,
11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13 # GNU General Public License for more details.
14 # -----------------------------------------------------------------------
15
16 import osrf.json
17 import osrf.conf
18 import osrf.log
19 import osrf.net
20 import osrf.net_obj
21 from osrf.const import OSRF_APP_SESSION_CONNECTED, \
22     OSRF_APP_SESSION_CONNECTING, OSRF_APP_SESSION_DISCONNECTED, \
23     OSRF_MESSAGE_TYPE_CONNECT, OSRF_MESSAGE_TYPE_DISCONNECT, \
24     OSRF_MESSAGE_TYPE_REQUEST
25 import osrf.ex
26 import random, os, time, threading
27
28
29 # -----------------------------------------------------------------------
30 # Go ahead and register the common network objects
31 # -----------------------------------------------------------------------
32 osrf.net_obj.register_hint('osrfMessage', ['threadTrace', 'locale', 'type', 'payload'], 'hash')
33 osrf.net_obj.register_hint('osrfMethod', ['method', 'params'], 'hash')
34 osrf.net_obj.register_hint('osrfResult', ['status', 'statusCode', 'content'], 'hash')
35 osrf.net_obj.register_hint('osrfConnectStatus', ['status', 'statusCode'], 'hash')
36 osrf.net_obj.register_hint('osrfMethodException', ['status', 'statusCode'], 'hash')
37
38
39 class Session(object):
40     """Abstract session superclass."""
41
42     ''' Global cache of in-service sessions '''
43     session_cache = {}
44
45     def __init__(self):
46         # by default, we're connected to no one
47         self.state = OSRF_APP_SESSION_DISCONNECTED
48         self.remote_id = None
49         self.locale = None
50
51     def find_session(threadTrace):
52         return Session.session_cache.get(threadTrace)
53     find_session = staticmethod(find_session)
54
55     def wait(self, timeout=120):
56         """Wait up to <timeout> seconds for data to arrive on the network"""
57         osrf.log.log_internal("Session.wait(%d)" % timeout)
58         handle = osrf.net.get_network_handle()
59         handle.recv(timeout)
60
61     def send(self, omessage):
62         """Sends an OpenSRF message"""
63         net_msg = osrf.net.NetworkMessage(
64             recipient      = self.remote_id,
65             body    = osrf.json.to_json([omessage]),
66             thread = self.thread,
67             locale = self.locale,
68         )
69
70         handle = osrf.net.get_network_handle()
71         handle.send(net_msg)
72
73     def cleanup(self):
74         """Removes the session from the global session cache."""
75         del Session.session_cache[self.thread]
76
77 class ClientSession(Session):
78     """Client session object.  Use this to make server requests."""
79
80     def __init__(self, service, locale='en_US'):
81         
82         # call superclass constructor
83         Session.__init__(self)
84
85         # the remote service we want to make requests of
86         self.service = service
87
88         # the locale we want requests to be returned in
89         self.locale = locale
90
91         # find the remote service handle <router>@<domain>/<service>
92         domain = osrf.conf.get('domains.domain', 0)
93         router = osrf.conf.get('router_name')
94         self.remote_id = "%s@%s/%s" % (router, domain, service)
95         self.orig_remote_id = self.remote_id
96
97         # generate a random message thread
98         self.thread = "%s%s%s%s" % (os.getpid(), 
99             str(random.randint(100,100000)), str(time.time()),threading.currentThread().getName().lower())
100
101         # how many requests this session has taken part in
102         self.next_id = 0 
103
104         # cache of request objects 
105         self.requests = {}
106
107         # cache this session in the global session cache
108         Session.session_cache[self.thread] = self
109
110     def reset_request_timeout(self, rid):
111         req = self.find_request(rid)
112         if req:
113             req.reset_timeout = True
114             
115
116     def request2(self, method, arr):
117         """Creates a new request and sends the request to the server using a python array as the params."""
118         return self.__request(method, arr)
119
120     def request(self, method, *args):
121         """Creates a new request and sends the request to the server using a variable argument list as params"""
122         arr = list(args)
123         return self.__request(method, arr)
124
125     def __request(self, method, arr):
126         """Builds the request object and sends it."""
127         if self.state != OSRF_APP_SESSION_CONNECTED:
128             self.reset_remote_id()
129
130         osrf.log.log_debug("Sending request %s -> %s " % (self.service, method))
131         req = Request(self, self.next_id, method, arr, self.locale)
132         self.requests[str(self.next_id)] = req
133         self.next_id += 1
134         req.send()
135         return req
136
137
138     def connect(self, timeout=10):
139         """Connects to a remote service"""
140
141         if self.state == OSRF_APP_SESSION_CONNECTED:
142             return True
143         self.state == OSRF_APP_SESSION_CONNECTING
144
145         # construct and send a CONNECT message
146         self.send(
147             osrf.net_obj.NetworkObject.osrfMessage( 
148                 {   'threadTrace' : 0,
149                     'type' : OSRF_MESSAGE_TYPE_CONNECT
150                 } 
151             )
152         )
153
154         while timeout >= 0 and not self.state == OSRF_APP_SESSION_CONNECTED:
155             start = time.time()
156             self.wait(timeout)
157             timeout -= time.time() - start
158         
159         if self.state != OSRF_APP_SESSION_CONNECTED:
160             raise osrf.ex.OSRFServiceException("Unable to connect to " + self.service)
161         
162         return True
163
164     def disconnect(self):
165         """Disconnects from a remote service"""
166
167         if self.state == OSRF_APP_SESSION_DISCONNECTED:
168             return True
169
170         self.send(
171             osrf.net_obj.NetworkObject.osrfMessage( 
172                 {   'threadTrace' : 0,
173                     'type' : OSRF_MESSAGE_TYPE_DISCONNECT
174                 } 
175             )
176         )
177
178         self.state = OSRF_APP_SESSION_DISCONNECTED
179
180     
181     def set_remote_id(self, remoteid):
182         self.remote_id = remoteid
183         osrf.log.log_internal("Setting request remote ID to %s" % self.remote_id)
184
185     def reset_remote_id(self):
186         """Recovers the original remote id"""
187         self.remote_id = self.orig_remote_id
188         osrf.log.log_internal("Resetting remote ID to %s" % self.remote_id)
189
190     def push_response_queue(self, message):
191         """Pushes the message payload onto the response queue 
192             for the request associated with the message's ID."""
193         osrf.log.log_debug("pushing %s" % message.payload())
194         try:
195             self.find_request(message.threadTrace()).push_response(message.payload())
196         except Exception, e: 
197             osrf.log.log_warn("pushing respond to non-existent request %s : %s" % (message.threadTrace(), e))
198
199     def find_request(self, rid):
200         """Returns the original request matching this message's threadTrace."""
201         try:
202             return self.requests[str(rid)]
203         except KeyError:
204             osrf.log.log_debug('find_request(): non-existent request %s' % str(rid))
205             return None
206
207     @staticmethod
208     def atomic_request(service, method, *args):
209         ses = ClientSession(service)
210         req = ses.request2(method, list(args))
211         resp = req.recv()
212         data = resp.content()
213         req.cleanup()
214         ses.cleanup()
215         return data
216
217
218
219
220 class Request(object):
221     """Represents a single OpenSRF request.
222         A request is made and any resulting respones are 
223         collected for the client."""
224
225     def __init__(self, session, rid, method=None, params=[], locale='en-US'):
226
227         self.session = session # my session handle
228         self.rid     = rid # my unique request ID
229         self.method = method # method name
230         self.params = params # my method params
231         self.queue  = [] # response queue
232         self.reset_timeout = False # resets the recv timeout?
233         self.complete = False # has the server told us this request is done?
234         self.send_time = 0 # local time the request was put on the wire
235         self.complete_time =  0 # time the server told us the request was completed
236         self.first_response_time = 0 # time it took for our first reponse to be received
237         self.locale = locale
238
239     def send(self):
240         """Sends a request message"""
241
242         # construct the method object message with params and method name
243         method = osrf.net_obj.NetworkObject.osrfMethod( {
244             'method' : self.method,
245             'params' : self.params
246         } )
247
248         # construct the osrf message with our method message embedded
249         message = osrf.net_obj.NetworkObject.osrfMessage( {
250             'threadTrace' : self.rid,
251             'type' : OSRF_MESSAGE_TYPE_REQUEST,
252             'payload' : method,
253             'locale' : self.locale
254         } )
255
256         self.send_time = time.time()
257         self.session.send(message)
258
259     def recv(self, timeout=120):
260         """ Waits up to <timeout> seconds for a response to this request.
261         
262             If a message is received in time, the response message is returned.
263             Returns None otherwise."""
264
265         self.session.wait(0)
266
267         orig_timeout = timeout
268         while not self.complete and (timeout >= 0 or orig_timeout < 0) and len(self.queue) == 0:
269             s = time.time()
270             self.session.wait(timeout)
271             if orig_timeout >= 0:
272                 timeout -= time.time() - s
273             if self.reset_timeout:
274                 self.reset_timeout = False
275                 timeout = orig_timeout
276
277         now = time.time()
278
279         # -----------------------------------------------------------------
280         # log some statistics 
281         if len(self.queue) > 0:
282             if not self.first_response_time:
283                 self.first_response_time = now
284                 osrf.log.log_debug("time elapsed before first response: %f" \
285                     % (self.first_response_time - self.send_time))
286
287         if self.complete:
288             if not self.complete_time:
289                 self.complete_time = now
290                 osrf.log.log_debug("time elapsed before complete: %f" \
291                     % (self.complete_time - self.send_time))
292         # -----------------------------------------------------------------
293
294
295         if len(self.queue) > 0:
296             # we have a reponse, return it
297             return self.queue.pop(0)
298
299         return None
300
301     def push_response(self, content):
302         """Pushes a method response onto this requests response queue."""
303         self.queue.append(content)
304
305     def cleanup(self):
306         """Cleans up request data from the cache. 
307
308             Do this when you are done with a request to prevent "leaked" cache memory."""
309         del self.session.requests[str(self.rid)]
310
311     def set_complete(self):
312         """Sets me as complete.  This means the server has sent a 'request complete' message"""
313         self.complete = True
314
315
316 class ServerSession(Session):
317     """Implements a server-side session"""
318     pass
319
320
321
322
323
324 class MultiSession(object):
325     ''' Manages multiple requests.  With the current implementation, a 1 second 
326         lag time before the first response is practically guaranteed.  Use 
327         only for long running requests.
328
329         Another approach would be a threaded version, but that would require
330         build-up and breakdown of thread-specific xmpp connections somewhere.
331         conection pooling? 
332     '''
333     class Container(object):
334         def __init__(self, req):
335             self.req = req
336             self.id = None
337
338     def __init__(self):
339         self.complete = False
340         self.reqs = []
341
342     def request(self, service, method, *args):
343         ses = ClientSession(service)
344         cont = MultiSession.Container(ses.request(method, *args))
345         cont.id = len(self.reqs)
346         self.reqs.append(cont)
347
348     def recv(self, timeout=10):
349         ''' Returns a tuple of req_id, response '''
350         duration = 0
351         block_time = 1
352         while True:
353             for i in range(0, len(self.reqs)):
354                 cont = self.reqs[i]
355                 req = cont.req
356
357                 res = req.recv(0)
358                 if i == 0 and not res:
359                     res = req.recv(block_time)
360
361                 if res: break
362
363             if res: break
364
365             duration += block_time
366             if duration >= timeout:
367                 return None
368
369         if req.complete:
370             self.reqs.pop(self.reqs.index(cont))
371
372         if len(self.reqs) == 0:
373             self.complete = True
374
375         return cont.id, res.content()
376