[py-svn] r39982 - py/trunk/py/execnet

[hpk at codespeak.net]

Author: hpk
Date: Tue Mar  6 13:51:18 2007
New Revision: 39982

Modified:
   py/trunk/py/execnet/channel.py
   py/trunk/py/execnet/gateway.py
Log:
introduce gateway._send and have all places route
their sending of Messages (or None's) through that method. 


Modified: py/trunk/py/execnet/channel.py
==============================================================================
--- py/trunk/py/execnet/channel.py	(original)
+++ py/trunk/py/execnet/channel.py	Tue Mar  6 13:51:18 2007
@@ -85,7 +85,7 @@
                 Msg = Message.CHANNEL_LAST_MESSAGE
             else:
                 Msg = Message.CHANNEL_CLOSE
-            self.gateway._outgoing.put(Msg(self.id))
+            self.gateway._send(Msg(self.id))
 
     def _getremoteerror(self):
         try:
@@ -117,7 +117,7 @@
             # state transition "opened/sendonly" --> "closed"
             # threads warning: the channel might be closed under our feet,
             # but it's never damaging to send too many CHANNEL_CLOSE messages
-            put = self.gateway._outgoing.put
+            put = self.gateway._send 
             if error is not None:
                 put(Message.CHANNEL_CLOSE_ERROR(self.id, str(error)))
             else:
@@ -157,7 +157,7 @@
             data = Message.CHANNEL_NEW(self.id, item.id)
         else:
             data = Message.CHANNEL_DATA(self.id, item)
-        self.gateway._outgoing.put(data)
+        self.gateway._send(data)
 
     def receive(self):
         """receives an item that was sent from the other side,

Modified: py/trunk/py/execnet/gateway.py
==============================================================================
--- py/trunk/py/execnet/gateway.py	(original)
+++ py/trunk/py/execnet/gateway.py	Tue Mar  6 13:51:18 2007
@@ -111,10 +111,13 @@
                     self._traceex(exc_info())
                     break 
         finally:
-            self._outgoing.put(None)
+            self._send(None)
             self._channelfactory._finished_receiving()
             self._trace('leaving %r' % threading.currentThread())
 
+    def _send(self, msg):
+        self._outgoing.put(msg)
+
     def _thread_sender(self):
         """ thread to send Messages over the wire. """
         try:
@@ -219,8 +222,8 @@
         channel = self.newchannel() 
         outid = self._newredirectchannelid(stdout) 
         errid = self._newredirectchannelid(stderr) 
-        self._outgoing.put(Message.CHANNEL_OPEN(channel.id, 
-                               (source, outid, errid)))
+        self._send(Message.CHANNEL_OPEN(
+                    channel.id, (source, outid, errid)))
         return channel 
 
     def _remote_redirect(self, stdout=None, stderr=None): 
@@ -260,7 +263,7 @@
         except KeyError:
             pass
         else:
-            self._outgoing.put(None)
+            self._send(None)
 
     def join(self, joinexec=True):
         """ Wait for all IO (and by default all execution activity)