*channels takes ownership on nested *channels

bombela · bombela · commit 72086db52f26 · 2012-06-05T22:42:06.000-07:00
diff --git a/tests/test_buffered_channel.py b/tests/test_buffered_channel.py
@@ -55,14 +55,10 @@ def test_close_server_bufchan():
     gevent.sleep(3)
     print 'CLOSE SERVER SOCKET!!!'
     server_bufchan.close()
-    server_hbchan.close()
-    server_channel.close()
     with assert_raises(zerorpc.LostRemote):
         client_bufchan.recv()
     print 'CLIENT LOST SERVER :)'
     client_bufchan.close()
-    client_hbchan.close()
-    client_channel.close()
     server.close()
     client.close()
 
@@ -91,14 +87,10 @@ def test_close_client_bufchan():
     gevent.sleep(3)
     print 'CLOSE CLIENT SOCKET!!!'
     client_bufchan.close()
-    server_hbchan.close()
-    client_channel.close()
     with assert_raises(zerorpc.LostRemote):
         server_bufchan.recv()
     print 'SERVER LOST CLIENT :)'
     server_bufchan.close()
-    client_hbchan.close()
-    server_channel.close()
     server.close()
     client.close()
 
@@ -125,14 +117,10 @@ def test_heartbeat_can_open_channel_server_close():
     gevent.sleep(3)
     print 'CLOSE SERVER SOCKET!!!'
     server_bufchan.close()
-    server_hbchan.close()
-    server_channel.close()
     with assert_raises(zerorpc.LostRemote):
         client_bufchan.recv()
     print 'CLIENT LOST SERVER :)'
     client_bufchan.close()
-    client_hbchan.close()
-    client_channel.close()
     server.close()
     client.close()
 
@@ -159,15 +147,11 @@ def test_heartbeat_can_open_channel_client_close():
     gevent.sleep(3)
     print 'CLOSE CLIENT SOCKET!!!'
     client_bufchan.close()
-    server_hbchan.close()
-    client_channel.close()
     client.close()
     with assert_raises(zerorpc.LostRemote):
         server_bufchan.recv()
     print 'SERVER LOST CLIENT :)'
     server_bufchan.close()
-    client_hbchan.close()
-    server_channel.close()
     server.close()
 
 
@@ -197,8 +181,6 @@ def client_do():
             assert event.name == 'OK'
             assert event.args == (x + x * x,)
         client_bufchan.close()
-        client_hbchan.close()
-        client_channel.close()
 
     client_task = gevent.spawn(client_do)
 
@@ -208,8 +190,6 @@ def server_do():
             assert event.name == 'add'
             server_bufchan.emit('OK', (sum(event.args),))
         server_bufchan.close()
-        server_hbchan.close()
-        server_channel.close()
 
     server_task = gevent.spawn(server_do)
 
@@ -243,8 +223,6 @@ def client_do():
         with assert_raises(zerorpc.LostRemote):
             event = client_bufchan.recv()
         client_bufchan.close()
-        client_hbchan.close()
-        client_channel.close()
 
     client_task = gevent.spawn(client_do)
 
@@ -258,8 +236,6 @@ def server_do():
             assert event.name == 'add'
             server_bufchan.emit('OK', (sum(event.args),))
         server_bufchan.close()
-        server_hbchan.close()
-        server_channel.close()
 
     server_task = gevent.spawn(server_do)
 
@@ -290,7 +266,6 @@ def client_do():
             assert event.name == 'OK'
             assert event.args == (x + x * x,)
         client_bufchan.close()
-        client_channel.close()
 
     client_task = gevent.spawn(client_do)
 
@@ -308,7 +283,6 @@ def server_do():
         with assert_raises(zerorpc.LostRemote):
             event = server_bufchan.recv()
         server_bufchan.close()
-        server_channel.close()
 
     server_task = gevent.spawn(server_do)
 
@@ -340,7 +314,6 @@ def client_do():
                 assert event.name == 'OK'
                 assert event.args == (x,)
         client_bufchan.close()
-        client_channel.close()
 
     client_task = gevent.spawn(client_do)
 
@@ -357,7 +330,6 @@ def server_do():
                 gevent.sleep(event.args[0])
                 server_bufchan.emit('OK', event.args)
         server_bufchan.close()
-        server_channel.close()
 
     server_task = gevent.spawn(server_do)
 
@@ -416,9 +388,7 @@ def server_do():
     server_task.get()
     client_task.get()
     client_bufchan.close()
-    client_hbchan.close()
     client.close()
     server_task.get()
     server_bufchan.close()
-    server_hbchan.close()
     server.close()
diff --git a/tests/test_channel.py b/tests/test_channel.py
@@ -114,4 +114,4 @@ def test_events_channel_both_side():
     server_events.close()
     server_channel.close()
     client_channel.close()
-    client_channel.close()
+    client_events.close()
diff --git a/tests/test_heartbeat.py b/tests/test_heartbeat.py
@@ -53,12 +53,10 @@ def test_close_server_hbchan():
     gevent.sleep(3)
     print 'CLOSE SERVER SOCKET!!!'
     server_hbchan.close()
-    server_channel.close()
     with assert_raises(zerorpc.LostRemote):
         client_hbchan.recv()
     print 'CLIENT LOST SERVER :)'
     client_hbchan.close()
-    client_channel.close()
     server.close()
     client.close()
 
@@ -85,12 +83,10 @@ def test_close_client_hbchan():
     gevent.sleep(3)
     print 'CLOSE CLIENT SOCKET!!!'
     client_hbchan.close()
-    client_channel.close()
     with assert_raises(zerorpc.LostRemote):
         server_hbchan.recv()
     print 'SERVER LOST CLIENT :)'
     server_hbchan.close()
-    server_channel.close()
     server.close()
     client.close()
 
@@ -115,12 +111,10 @@ def test_heartbeat_can_open_channel_server_close():
     gevent.sleep(3)
     print 'CLOSE SERVER SOCKET!!!'
     server_hbchan.close()
-    server_channel.close()
     with assert_raises(zerorpc.LostRemote):
         client_hbchan.recv()
     print 'CLIENT LOST SERVER :)'
     client_hbchan.close()
-    client_channel.close()
     server.close()
     client.close()
 
@@ -145,13 +139,11 @@ def test_heartbeat_can_open_channel_client_close():
     gevent.sleep(3)
     print 'CLOSE CLIENT SOCKET!!!'
     client_hbchan.close()
-    client_channel.close()
     client.close()
     with assert_raises(zerorpc.LostRemote):
         server_hbchan.recv()
     print 'SERVER LOST CLIENT :)'
     server_hbchan.close()
-    server_channel.close()
     server.close()
 
 
@@ -179,7 +171,6 @@ def client_do():
             assert event.name == 'OK'
             assert event.args == (x + x * x,)
         client_hbchan.close()
-        client_channel.close()
 
     client_task = gevent.spawn(client_do)
 
@@ -189,7 +180,6 @@ def server_do():
             assert event.name == 'add'
             server_hbchan.emit('OK', (sum(event.args),))
         server_hbchan.close()
-        server_channel.close()
 
     server_task = gevent.spawn(server_do)
 
@@ -222,7 +212,6 @@ def client_do():
         with assert_raises(zerorpc.LostRemote):
             event = client_hbchan.recv()
         client_hbchan.close()
-        client_channel.close()
 
     client_task = gevent.spawn(client_do)
 
@@ -235,7 +224,6 @@ def server_do():
             assert event.name == 'add'
             server_hbchan.emit('OK', (sum(event.args),))
         server_hbchan.close()
-        server_channel.close()
 
     server_task = gevent.spawn(server_do)
 
@@ -265,7 +253,6 @@ def client_do():
             assert event.name == 'OK'
             assert event.args == (x + x * x,)
         client_hbchan.close()
-        client_channel.close()
 
     client_task = gevent.spawn(client_do)
 
@@ -282,7 +269,6 @@ def server_do():
         with assert_raises(zerorpc.LostRemote):
             event = server_hbchan.recv()
         server_hbchan.close()
-        server_channel.close()
 
     server_task = gevent.spawn(server_do)
 
@@ -313,7 +299,6 @@ def client_do():
                 assert event.name == 'OK'
                 assert event.args == (x,)
         client_hbchan.close()
-        client_channel.close()
 
     client_task = gevent.spawn(client_do)
 
@@ -329,7 +314,6 @@ def server_do():
                 gevent.sleep(event.args[0])
                 server_hbchan.emit('OK', event.args)
         server_hbchan.close()
-        server_channel.close()
 
     server_task = gevent.spawn(server_do)
 
diff --git a/zerorpc/channel.py b/zerorpc/channel.py
@@ -181,8 +181,12 @@ def __del__(self):
         self.close()
 
     def close(self):
-        if self._recv_task:
+        if self._recv_task is not None:
             self._recv_task.kill()
+            self._recv_task = None
+        if self._channel is not None:
+            self._channel.close()
+            self._channel = None
 
     def _recver(self):
         while True:
diff --git a/zerorpc/core.py b/zerorpc/core.py
@@ -147,8 +147,6 @@ def _async_task(self, initial_event):
                     self._context.middleware_get_task_context())
         finally:
             bufchan.close()
-            bufchan.channel.close()
-            bufchan.channel.channel.close()
 
     def _acceptor(self):
         while True:
@@ -213,8 +211,6 @@ def _process_response(self, method, bufchan, timeout):
                     self._raise_remote_error)
         except:
             bufchan.close()
-            bufchan.channel.close()
-            bufchan.channel.channel.close()
             raise
 
     def __call__(self, method, *args, **kargs):
@@ -237,8 +233,6 @@ def __call__(self, method, *args, **kargs):
             return async_result
         except:
             bufchan.close()
-            bufchan.channel.close()
-            bufchan.channel.channel.close()
             raise
 
     def __getattr__(self, method):
diff --git a/zerorpc/heartbeat.py b/zerorpc/heartbeat.py
@@ -56,10 +56,15 @@ def __del__(self):
         self.close()
 
     def close(self):
-        if self._heartbeat_task:
+        if self._heartbeat_task is not None:
             self._heartbeat_task.kill()
-        if self._recv_task:
+            self._heartbeat_task = None
+        if self._recv_task is not None:
             self._recv_task.kill()
+            self._recv_task = None
+        if self._channel is not None:
+            self._channel.close()
+            self._channel = None
 
     def _heartbeat(self):
         while True:
diff --git a/zerorpc/patterns.py b/zerorpc/patterns.py
@@ -24,7 +24,7 @@
 
 
 
-class ReqRep():
+class ReqRep:
 
     def process_call(self, context, bufchan, event, functor):
         result = context.middleware_call_procedure(functor, *event.args)
@@ -39,12 +39,10 @@ def process_answer(self, context, bufchan, event, method,
         if event.name == 'ERR':
             raise_remote_error(event)
         bufchan.close()
-        bufchan.channel.close()
-        bufchan.channel.channel.close()
         return result
 
 
-class ReqStream():
+class ReqStream:
 
     def process_call(self, context, bufchan, event, functor):
         xheader = context.middleware_get_task_context()
@@ -65,7 +63,7 @@ def iterator(event):
             if event.name == 'ERR':
                 raise_remote_error(event)
             bufchan.close()
-            bufchan.channel.channel.close()
         return iterator(event)
 
+
 patterns_list = [ReqStream(), ReqRep()]
