Source Code for Module SpecClient.SpecConnectionsManager

  1  #$Id: SpecConnectionsManager.py,v 1.13 2005/09/27 13:54:58 guijarro Exp $ 
  2  """Module for managing connections to Spec 
  3   
  4  The SpecConnectionsManager module provides facilities to get 
  5  a connection to Spec. It can run a thread for 'asynchronous' 
  6  polling of socket events. It prevents from having more than 
  7  one connection to the same Spec server at the same time, and 
  8  automatically reconnects lost connections. 
  9   
 10   
 11  Classes : 
 12    _ThreadedSpecConnectionsManager 
 13    _SpecConnectionsManager 
 14  """ 
 15   
 16  __author__ = 'Matias Guijarro' 
 17  __version__ = '1.1' 
 18   
 19  import atexit 
 20  import threading 
 21  import time 
 22  import weakref 
 23  import asyncore 
 24  import sys 
 25  import gc 
 26   
 27  import SpecConnection 
 28  import SpecEventsDispatcher 
 29   
 30   
 31  _SpecConnectionsManagerInstance = None #keep a reference to the Singleton instance 
 32   
 33   
 34 -def SpecConnectionsManager(pollingThread = True, also_dispatch_events=False): 
 35      """Return the Singleton Spec connections manager instance""" 
 36      global _SpecConnectionsManagerInstance 
 37   
 38      if _SpecConnectionsManagerInstance is None: 
 39          if pollingThread: 
 40              _SpecConnectionsManagerInstance = _ThreadedSpecConnectionsManager(also_dispatch_events) 
 41   
 42              def _endSpecConnectionsManager(): 
 43                  global _SpecConnectionsManagerInstance 
 44   
 45                  if _SpecConnectionsManagerInstance is not None: 
 46                      _SpecConnectionsManagerInstance.stop() 
 47                      _SpecConnectionsManagerInstance = None 
 48   
 49              # register _endSpecConnectionsManager() to be called on Python interpreter exit 
 50              atexit.register(_endSpecConnectionsManager) 
 51          else: 
 52              _SpecConnectionsManagerInstance = _SpecConnectionsManager() 
 53   
 54      return _SpecConnectionsManagerInstance 
 55   
 56   
 57 -class _ThreadedSpecConnectionsManager(threading.Thread): 
 58      """Class for managing connections to Spec 
 59   
 60      Polling of asynchronous socket events is delegated to a separate thread 
 61   
 62      Warning: should never be instanciated directly ; use the module level SpecConnectionsManager() 
 63      function instead. 
 64      """ 
 65 -    def __init__(self, dispatch_events): 
 66          """Constructor""" 
 67          threading.Thread.__init__(self) 
 68   
 69          self.lock = threading.Lock() 
 70          self.connections = {} 
 71          self.connectionDispatchers = {} 
 72          self.stopEvent = threading.Event() 
 73          self.__started = False 
 74          self.doEventsDispatching = dispatch_events 
 75          self.setDaemon(True) 
 76   
 77   
 78 -    def run(self): 
 79          """Override Thread.run() ; define behaviour for the connections manager thread 
 80   
 81          For each SpecConnection object in the connections dictionary, try to make 
 82          a connection. If the connection is already established, nothing occurs. 
 83          Poll the asyncore dispatchers for processing socket events. 
 84          """ 
 85          self.__started = True 
 86   
 87          while not self.stopEvent.isSet(): 
 88              self.lock.acquire() 
 89              try: 
 90                  connection_dispatcher_keys = self.connectionDispatchers.keys() 
 91                  for k in connection_dispatcher_keys: 
 92                    connection = self.connectionDispatchers.get(k) 
 93                    if connection is not None: 
 94                      connection.makeConnection() 
 95   
 96                  if self.stopEvent.isSet(): 
 97                      break 
 98              finally: 
 99                  self.lock.release() 
100   
101              if len(self.connectionDispatchers) > 0: 
102                  asyncore.loop(0.01, False, None, 1) 
103                  if self.doEventsDispatching: 
104                    SpecEventsDispatcher.dispatch() 
105              else: 
106                  time.sleep(0.01) 
107   
108          asyncore.loop(0.01, False, None, 1) 
109   
110   
111 -    def stop(self): 
112          """Stop the connections manager thread and dereferences all connections""" 
113          self.stopEvent.set() 
114   
115          self.join() 
116   
117          self.__started = False 
118          self.connections = {} 
119   
120   
121 -    def getConnection(self, specVersion): 
122          """Return a SpecConnection object 
123   
124          Arguments: 
125          specVersion -- a string in the 'host:port' form 
126          """ 
127          gc.collect() 
128   
129          try: 
130              con = self.connections[specVersion]() 
131          except KeyError: 
132              con = SpecConnection.SpecConnection(specVersion) 
133   
134              def removeConnection(ref, connectionName = specVersion): 
135                  self.closeConnection(connectionName) 
136   
137              self.connections[specVersion] = weakref.ref(con, removeConnection) 
138   
139              self.lock.acquire() 
140              try: 
141                  self.connectionDispatchers[specVersion] = con.dispatcher 
142              finally: 
143                  self.lock.release() 
144   
145          if not self.__started: 
146              self.start() 
147              self.__started = True 
148   
149          return con 
150   
151   
152 -    def closeConnection(self, specVersion): 
153          self.lock.acquire() 
154          try: 
155              self.connectionDispatchers[specVersion].handle_close() 
156   
157              del self.connectionDispatchers[specVersion] 
158              del self.connections[specVersion] 
159          finally: 
160              self.lock.release() 
161   
162   
163 -    def closeConnections(self): 
164          for connectionName in self.connectionDispatchers.keys(): 
165              self.closeConnection(connectionName) 
166   
167   
168 -class _SpecConnectionsManager: 
169      """Class for managing connections to Spec 
170   
171      The poll() method should be called inside a GUI loop during idle time. 
172      Unlike the threaded class, the poll method will also dispatch SpecClient events 
173   
174      Warning: should never be instanciated directly ; use the module level SpecConnectionsManager() 
175      function instead. 
176      """ 
177 -    def __init__(self): 
178          """Constructor""" 
179          self.connections = {} 
180          self.connectionDispatchers = {} 
181   
182   
183 -    def poll(self, timeout=0.01): 
184          """Poll the asynchronous socket connections and dispatch incomming events""" 
185          connection_dispatcher_keys = self.connectionDispatchers.keys() 
186          for k in connection_dispatcher_keys: 
187            connection = self.connectionDispatchers.get(k) 
188            if connection is not None: 
189              connection.makeConnection() 
190   
191          asyncore.loop(timeout, False, None, 1) 
192   
193          SpecEventsDispatcher.dispatch() 
194   
195   
196 -    def stop(self): 
197          """Stop the connections manager thread and dereferences all connections""" 
198          self.connections = {} 
199   
200   
201 -    def getConnection(self, specVersion): 
202          """Return a SpecConnection object 
203   
204          Arguments: 
205          specVersion -- a string in the 'host:port' form 
206          """ 
207          con = self.connections.get(specVersion) 
208          if con is None or con() is None: 
209              con = SpecConnection.SpecConnection(specVersion) 
210   
211              def removeConnection(ref, connectionName = specVersion): 
212                  self.closeConnection(connectionName) 
213   
214              self.connections[specVersion] = weakref.ref(con, removeConnection) 
215              self.connectionDispatchers[specVersion] = con.dispatcher 
216          else: 
217              con = con() 
218   
219          return con 
220       
221   
222 -    def closeConnection(self, specVersion): 
223          try: 
224              self.connectionDispatchers[specVersion].handle_close() 
225   
226              del self.connectionDispatchers[specVersion] 
227              del self.connections[specVersion] 
228          except: 
229              pass 
230   
231   
232 -    def closeConnections(self): 
233          for connectionName in self.connectionDispatchers.keys(): 
234              self.closeConnection(connectionName) 
235