manage subscriptions in network.py

This commit is contained in:
ThomasV 2013-10-02 13:00:02 +02:00
parent a22ae33a81
commit 40e393187a
3 changed files with 36 additions and 9 deletions

View File

@ -482,6 +482,7 @@ class Interface(threading.Thread):
self.subscriptions[callback].append(message) self.subscriptions[callback].append(message)
if not self.is_connected: if not self.is_connected:
print_error("interface: trying to send while not connected")
return return
if self.protocol in 'st': if self.protocol in 'st':

View File

@ -61,6 +61,29 @@ class Network(threading.Thread):
os.mkdir(dir_path) os.mkdir(dir_path)
# default subscriptions
self.subscriptions = {}
self.subscriptions[self.on_banner] = [('server.banner',[])]
self.subscriptions[self.on_peers] = [('server.peers.subscribe',[])]
def send_subscriptions(self):
for cb, sub in self.subscriptions.items():
self.interface.send(sub, cb)
def subscribe(self, messages, callback):
with self.lock:
if self.subscriptions.get(callback) is None:
self.subscriptions[callback] = []
for message in messages:
if message not in self.subscriptions[callback]:
self.subscriptions[callback].append(message)
if self.interface and self.interface.is_connected:
self.interface.send( messages, callback )
def register_callback(self, event, callback): def register_callback(self, event, callback):
with self.lock: with self.lock:
if not self.callbacks.get(event): if not self.callbacks.get(event):
@ -138,17 +161,21 @@ class Network(threading.Thread):
return return
# stop the interface in order to terminate subscriptions # stop the interface in order to terminate subscriptions
subscriptions = self.interface.subscriptions
self.interface.stop() self.interface.stop()
# notify gui # notify gui
self.trigger_callback('disconnecting') self.trigger_callback('disconnecting')
# start interface # start interface
self.default_server = server self.default_server = server
self.start_interface(server)
self.interface = self.interfaces[server] if server in self.interfaces.keys():
# send subscriptions self.interface = self.interfaces[server]
for cb, sub in subscriptions.items(): self.send_subscriptions()
self.interface.send(sub, cb) else:
self.start_interface(server)
self.interface = self.interfaces[server]
def run(self): def run(self):
@ -163,8 +190,7 @@ class Network(threading.Thread):
if i.is_connected: if i.is_connected:
i.send([ ('blockchain.headers.subscribe',[])], self.on_header) i.send([ ('blockchain.headers.subscribe',[])], self.on_header)
if i == self.interface: if i == self.interface:
i.send([('server.banner',[])], self.on_banner) self.send_subscriptions()
i.send([('server.peers.subscribe',[])], self.on_peers)
self.trigger_callback('connected') self.trigger_callback('connected')
else: else:
self.disconnected_servers.append(i.server) self.disconnected_servers.append(i.server)

View File

@ -1492,7 +1492,7 @@ class WalletSynchronizer(threading.Thread):
messages = [] messages = []
for addr in addresses: for addr in addresses:
messages.append(('blockchain.address.subscribe', [addr])) messages.append(('blockchain.address.subscribe', [addr]))
self.network.interface.send( messages, lambda i,r: self.queue.put(r)) self.network.subscribe( messages, lambda i,r: self.queue.put(r))
def run(self): def run(self):