Merge pull request #462 from bobh66/reuse_proxy_instance

Set stopped flag to allow PollingQueueConsumer object reuse

CHANGES

@@ -5,6 +5,15 @@ Here you can see the full list of changes between nameko versions. Versions
 are in form of *headline.major.minor* numbers. Backwards-compatible changes
 increment the minor version number only.
 
+Version TBD
+-----------
+
+Release: Pending
+
+* Set stopped flag in register_provider() to allow PollingQueueConsumer
+  object reuse [fixes #462]
+
+
 Version 2.6.0
 -------------
 

CONTRIBUTORS.txt

@@ -26,3 +26,4 @@ Alexandre Héaumé (@aheaume)
 Jesse Pollak (@jessepollak)
 Miklos Vincze (@vmikki)
 Yihang Yang (@mikegreen7892003)
+Bob Haddleton (@bobh66)

nameko/standalone/rpc.py

@@ -79,7 +79,7 @@ class PollingQueueConsumer(object):
     consumer = None
 
     def __init__(self, timeout=None):
-        self.stopped = False
+        self.stopped = True
         self.timeout = timeout
         self.replies = {}
 
@@ -119,6 +119,7 @@ def register_provider(self, provider):
 
         self.queue = provider.queue
         self._setup_consumer()
+        self.stopped = False
 
     def unregister_provider(self, provider):
         self.connection.close()

test/standalone/test_rpc_proxy.py

@@ -376,6 +376,18 @@ def test_cluster_proxy(container_factory, rabbit_manager, rabbit_config):
         assert proxy.foobar.spam(ham=1) == 1
 
 
+def test_cluster_proxy_reuse(container_factory, rabbit_manager, rabbit_config):
+    container = container_factory(FooService, rabbit_config)
+    container.start()
+
+    cluster_proxy = ClusterRpcProxy(rabbit_config)
+    with cluster_proxy as proxy:
+        assert proxy.foobar.spam(ham=1) == 1
+
+    with cluster_proxy as second_proxy:
+        assert second_proxy.foobar.spam(ham=1) == 1
+
+
 def test_cluster_proxy_dict_access(
     container_factory, rabbit_manager, rabbit_config
 ):