Do never parse any packet from aborted connection

neo/lib/connection.py

@@ -22,7 +22,7 @@ from .connector import ConnectorException, ConnectorDelayedConnection
 from .locking import RLock
 from .protocol import uuid_str, Errors, \
         PacketMalformedError, Packets, ParserState
-from .util import ReadBuffer
+from .util import dummy_read_buffer, ReadBuffer
 
 CRITICAL_TIMEOUT = 30
 
@@ -445,6 +445,7 @@ class Connection(BaseConnection):
             return
         logging.debug('aborting a connector for %r', self)
         self.aborted = True
+        self.read_buf = dummy_read_buffer
         if self._on_close is not None:
             self._on_close()
             self._on_close = None
@@ -482,8 +483,6 @@ class Connection(BaseConnection):
         except PacketMalformedError, e:
             logging.error('malformed packet from %r: %s', self, e)
             self._closure()
-        if self.aborted:
-            self.em.removeReader(self)
         return not not self._queue
 
     def hasPendingMessages(self):

neo/lib/util.py

@@ -214,6 +214,8 @@ class ReadBuffer(object):
         self.size = 0
         self.content.clear()
 
+dummy_read_buffer = ReadBuffer()
+dummy_read_buffer.append = lambda _: None
 
 class cached_property(object):
     """

neo/tests/threaded/test.py

@@ -1326,6 +1326,36 @@ class Test(NEOThreadedTest):
             for s in cluster.storage_list:
                 self.assertEqual(s.dm.getLastIDs()[0], truncate_tid)
 
+    def testConnectionAbort(self):
+        with self.getLoopbackConnection() as client:
+            poll = client.em.poll
+            while client.connecting:
+                poll(1)
+            server, = (c for c in client.em.connection_dict.itervalues()
+                         if c.isServer())
+            client.send(Packets.NotifyReady())
+            def writable(orig):
+                p.revert()
+                r = orig()
+                client.send(Packets.Ping())
+                return r
+            def process(orig):
+                self.assertFalse(server.aborted)
+                r = orig()
+                self.assertTrue(server.aborted)
+                server.em.removeWriter(server)
+                return r
+            with Patch(client, writable=writable) as p, \
+                 Patch(server, process=process):
+                poll(0)
+                poll(0)
+                server.em.addWriter(server)
+                self.assertIsNot(server.connector, None)
+                poll(0)
+                self.assertIs(server.connector, None)
+                poll(0)
+                self.assertIs(client.connector, None)
+
     def testConnectionTimeout(self):
         with self.getLoopbackConnection() as conn:
             conn.KEEP_ALIVE