Update error handling of read loop

nats/io/client.py

@@ -832,9 +832,15 @@ def _read_loop(self, data=''):
     to the parsing callback for processing.
     """
     while True:
-      if not self.io.closed():
+      if not self.is_connected or self.is_connecting or self.io.closed():
+        break
+
+      try:
         yield self.io.read_bytes(DEFAULT_READ_CHUNK_SIZE, streaming_callback=self._ps.parse, partial=True)
-      else:
+      except tornado.iostream.StreamClosedError as e:
+        self._err = e
+        if self._error_cb is not None and not self.is_reconnecting:
+          self._error_cb(e)
         break
 
   @tornado.gen.coroutine

tests/client_test.py

@@ -1014,8 +1014,10 @@ def publisher(self):
           yield c.nc.subscribe("foo",  "", log.persist)
           self.io_loop.spawn_callback(c.publisher)
 
+          yield tornado.gen.sleep(0.001)
           orig_gnatsd = self.server_pool.pop(0)
           orig_gnatsd.finish()
+          yield tornado.gen.sleep(0.001)
 
           try:
                a = nc._current_server
@@ -1297,7 +1299,7 @@ def reconnected_cb(self):
 
           self.assertTrue(c.disconnected_cb_called)
           self.assertFalse(c.close_cb_called)
-          self.assertFalse(c.error_cb_called)
+          self.assertTrue(c.error_cb_called)
           self.assertTrue(c.reconnected_cb_called)
 
           for i in range(0, 5):
@@ -1309,7 +1311,7 @@ def reconnected_cb(self):
           yield c.nc.close()
           self.assertTrue(c.disconnected_cb_called)
           self.assertTrue(c.close_cb_called)
-          self.assertFalse(c.error_cb_called)
+          self.assertTrue(c.error_cb_called)
           self.assertTrue(c.reconnected_cb_called)
 
 class ClientTLSCertsTest(tornado.testing.AsyncTestCase):