svn commit: r824082 - /qpid/trunk/qpid/python/qpid/driver.py

[rh...@apache.org]

Author: rhs
Date: Sun Oct 11 14:36:14 2009
New Revision: 824082

URL: http://svn.apache.org/viewvc?rev=824082&view=rev
Log:
catch socket errors on write as well as read

Modified:
    qpid/trunk/qpid/python/qpid/driver.py

Modified: qpid/trunk/qpid/python/qpid/driver.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/python/qpid/driver.py?rev=824082&r1=824081&r2=824082&view=diff
==============================================================================
--- qpid/trunk/qpid/python/qpid/driver.py (original)
+++ qpid/trunk/qpid/python/qpid/driver.py Sun Oct 11 14:36:14 2009
@@ -190,10 +190,10 @@
         log.debug("READ: %r", data)
       else:
         log.debug("ABORTED: %s", self._socket.getpeername())
-        error = ("connection aborted",)
+        error = "connection aborted"
         recoverable = True
     except socket.error, e:
-      error = (e,)
+      error = e
       recoverable = True
 
     if not error:
@@ -214,20 +214,13 @@
           log.debug("RCVD: %r", op)
           op.dispatch(self)
       except VersionError, e:
-        error = (e,)
+        error = e
       except:
         msg = compat.format_exc()
-        error = (msg,)
+        error = msg
 
     if error:
-      self._socket.close()
-      self.reset()
-      if recoverable and self.connection.reconnect:
-        self._timeout = time.time() + 3
-        log.warn("recoverable error: %s" % error)
-        log.warn("sleeping 3 seconds")
-      else:
-        self.connection.error = error
+      self._error(error, recoverable)
     else:
       self.dispatch()
 
@@ -241,9 +234,13 @@
 
   @synchronized
   def writeable(self):
-    n = self._socket.send(self._buf)
-    log.debug("SENT: %r", self._buf[:n])
-    self._buf = self._buf[n:]
+    try:
+      n = self._socket.send(self._buf)
+      log.debug("SENT: %r", self._buf[:n])
+      self._buf = self._buf[n:]
+    except socket.error, e:
+      self._error(e, True)
+      self.connection._waiter.notifyAll()
 
   @synchronized
   def timeout(self):
@@ -251,6 +248,17 @@
     self.dispatch()
     self.connection._waiter.notifyAll()
 
+  def _error(self, err, recoverable):
+    if self._socket is not None:
+      self._socket.close()
+    self.reset()
+    if recoverable and self.connection.reconnect:
+      self._timeout = time.time() + 3
+      log.warn("recoverable error: %s" % err)
+      log.warn("sleeping 3 seconds")
+    else:
+      self.connection.error = (err,)
+
   def write_op(self, op):
     log.debug("SENT: %r", op)
     self._op_enc.write(op)
@@ -372,10 +380,7 @@
       self._timeout = None
     except socket.error, e:
       if self.connection.reconnect:
-        self.reset()
-        self._timeout = time.time() + 3
-        log.warn("recoverable error: %s", e)
-        log.warn("sleeping 3 seconds")
+        self._error(e, True)
         return
       else:
         raise e



---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project:      http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org