Cls issue 70

nsq/async.py

@@ -188,7 +188,7 @@ def _start_read(self):
     def _socket_close(self):
         self.state = 'DISCONNECTED'
         self.trigger('close', conn=self)
-
+        
     def close(self):
         self.stream.close()
 
@@ -276,6 +276,16 @@ def send_rdy(self, value):
         self.rdy = value
         return True
 
+    def send_cls(self):
+        try:
+            self.send(nsq.cls())
+        except Exception, e:
+            self.close()
+            self.trigger('error', conn=self,
+                         error=nsq.SendError('failed to send CLS'))
+            return False
+        return True
+
     def _on_connect(self, **kwargs):
         identify_data = {
             'short_id': self.short_hostname,

nsq/client.py

@@ -78,3 +78,17 @@ def heartbeat(self, conn):
         :param conn: the :class:`nsq.AsyncConn` over which the heartbeat was received
         """
         pass
+
+    def close(self):
+        """
+        Closes all connections stops all periodic callbacks
+        """
+
+        for conn in self.conns.values():
+            try:
+                conn.send_cls()
+            except Exception, e:
+                logger.warning('[%s:%s] error sending CLS, closing',
+                               conn.id, self.name)
+            conn.close()
+

nsq/nsq.py

@@ -102,6 +102,8 @@ def touch(id):
 def nop():
     return _command('NOP', None)
 
+def cls():
+    return _command('CLS', None)
 
 def pub(topic, data):
     return _command('PUB', data, topic)

nsq/reader.py

@@ -240,11 +240,11 @@ def close(self):
         """
         Closes all connections stops all periodic callbacks
         """
-        for conn in self.conns.values():
-            conn.close()
-
         self.redist_periodic.stop()
         self.query_periodic.stop()
+
+        super(Reader, self).close()
+
 
     def set_message_handler(self, message_handler):
         """