Merge pull request #189 from nats-io/reconnect-with-lbs

Fix issue when reconnecting against LB

Author: wallyqs

nats/aio/client.py

@@ -28,7 +28,7 @@
 from nats.aio.nuid import NUID
 from nats.protocol.parser import *
 
-__version__ = '0.11.2'
+__version__ = '0.11.4'
 __lang__ = 'python3'
 _logger = logging.getLogger(__name__)
 PROTOCOL = 1
@@ -1353,7 +1353,7 @@ async def _attempt_reconnect(self):
                 self._err = e
                 await self.close()
                 break
-            except (OSError, NatsError, ErrTimeout) as e:
+            except (OSError, NatsError, asyncio.TimeoutError) as e:
                 self._err = e
                 await self._error_cb(e)
                 self._status = Client.RECONNECTING

tests/client_test.py

@@ -1507,7 +1507,6 @@ async def reconnected_cb():
             reconnected.set_result(True)
 
         async def err_cb(e):
-            print(e)
             nonlocal errors
             errors.append(e)
 
@@ -1583,7 +1582,7 @@ async def bad_server(reader, writer):
             await asyncio.sleep(0.2, loop=self.loop)
             writer.close()
 
-        await asyncio.start_server(
+        sv = await asyncio.start_server(
             bad_server, '127.0.0.1', 4555, loop=self.loop
         )
 
@@ -1605,6 +1604,53 @@ async def error_cb(e):
             await nc.connect(**options)
         self.assertEqual(1, len(errors))
         self.assertEqual(errors[0], nc.last_error)
+        sv.close()
+
+    @async_test
+    async def test_empty_response_from_server_after_reconnect(self):
+        async def bad_server(reader, writer):
+            writer.write("INFO {}\r\nPONG\r\n".encode())
+            await asyncio.sleep(0.2, loop=self.loop)
+            writer.close()
+
+        async def bad_server2(reader, writer):
+            writer.write(b'')
+
+        sv = await asyncio.start_server(
+            bad_server, '127.0.0.1', 4556, loop=self.loop
+        )
+        sv2 = await asyncio.start_server(
+            bad_server2, '127.0.0.1', 4557, loop=self.loop
+        )
+
+        future = asyncio.Future()
+
+        async def error_cb(e):
+            if type(e) is asyncio.TimeoutError:
+                future.set_result(True)
+
+        nc = NATS()
+        options = {
+            'servers': ["nats://127.0.0.1:4556", "nats://127.0.0.1:4557"],
+            'error_cb': error_cb,
+            'io_loop': self.loop,
+            'allow_reconnect': True,
+            'reconnect_time_wait': 0.1,
+            'dont_randomize': True,
+            'connect_timeout': 0.5
+        }
+
+        await nc.connect(**options)
+        sv.close()
+        await sv.wait_closed()
+        await asyncio.sleep(1, loop=self.loop)
+        sv2.close()
+        await sv2.wait_closed()
+
+        # Wait for the timeout error from the EOF to the second server.
+        await asyncio.wait_for(future, 1)
+
+        await nc.close()
 
     @async_test
     async def test_malformed_info_response_from_server(self):