Merge pull request #1219 from kyuupichan/network_queues

Have queues passed to the network constructor.

lib/daemon.py

@@ -117,9 +117,10 @@ class NetworkServer(util.DaemonThread):
         util.DaemonThread.__init__(self)
         self.debug = False
         self.config = config
-        self.network = Network(config)
         # network sends responses on that queue
         self.network_queue = Queue.Queue()
+        self.requests_queue = Queue.Queue()
+        self.network = Network(self.requests_queue, self.network_queue, config)
 
         self.running = False
         self.lock = threading.RLock()
@@ -150,11 +151,11 @@ class NetworkServer(util.DaemonThread):
 
         if self.debug:
             print_error("-->", request)
-        self.network.requests_queue.put(request)
+        self.requests_queue.put(request)
 
 
     def run(self):
-        self.network.start(self.network_queue)
+        self.network.start()
         while self.is_running():
             try:
                 response = self.network_queue.get(timeout=0.1)

lib/network.py

@@ -123,7 +123,7 @@ def serialize_server(host, port, protocol):
 
 class Network(util.DaemonThread):
 
-    def __init__(self, config=None):
+    def __init__(self, requests_queue, response_queue, config=None):
         if config is None:
             config = {}  # Do not use mutables as default values!
         util.DaemonThread.__init__(self)
@@ -132,6 +132,8 @@ class Network(util.DaemonThread):
         self.blockchain = Blockchain(self.config, self)
         self.interfaces = {}
         self.queue = Queue.Queue()
+        self.requests_queue = requests_queue
+        self.response_queue = response_queue
         # Server for addresses and transactions
         self.default_server = self.config.get('server')
         # Sanitize default server
@@ -168,7 +170,6 @@ class Network(util.DaemonThread):
         self.unanswered_requests = {}
 
         self.connection_status = 'connecting'
-        self.requests_queue = Queue.Queue()
         self.set_proxy(deserialize_proxy(self.config.get('proxy')))
         # retry times
         self.server_retry_time = time.time()
@@ -297,9 +298,8 @@ class Network(util.DaemonThread):
         for i in range(self.num_server):
             self.start_random_interface()
 
-    def start(self, response_queue):
+    def start(self):
         self.running = True
-        self.response_queue = response_queue
         self.start_interfaces()
         self.blockchain.start()
         util.DaemonThread.start(self)

lib/network_proxy.py

@@ -54,9 +54,9 @@ class NetworkProxy(util.DaemonThread):
             self.pipe = util.SocketPipe(socket)
             self.network = None
         else:
-            self.network = Network(config)
-            self.pipe = util.QueuePipe(send_queue=self.network.requests_queue)
-            self.network.start(self.pipe.get_queue)
+            self.pipe = util.QueuePipe()
+            self.network = Network(self.pipe.send_queue, self.pipe.get_queue, config)
+            self.network.start()
             for key in ['status','banner','updated','servers','interfaces']:
                 value = self.network.get_status_value(key)
                 self.pipe.get_queue.put({'method':'network.status', 'params':[key, value]})