mir/pyrogram
2
0
Fork 0
mirror of https://github.com/pyrogram/pyrogram synced 2025-08-28 21:07:59 +00:00
Code Issues Releases Wiki Activity

Join threads before closing the connection

Browse Source
This commit is contained in:
Dan 2018-04-13 19:09:00 +02:00
parent e3c70c84ba
commit 17259d1b7b
2 changed files with 39 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
pyrogram/client/client.py
View File

@ -191,7 +191,9 @@ class Client:
self.is_idle = None
self.updates_queue = Queue()
self.updates_workers_list = []
self.download_queue = Queue()
self.download_workers_list = []
self.dispatcher = Dispatcher(self, workers)
self.update_handler = None
@ -301,10 +303,24 @@ class Client:
self.send(functions.updates.GetState())
for i in range(self.UPDATES_WORKERS):
Thread(target=self.updates_worker, name="UpdatesWorker#{}".format(i + 1)).start()
self.updates_workers_list.append(
Thread(
target=self.updates_worker,
name="UpdatesWorker#{}".format(i + 1)
)
)
self.updates_workers_list[-1].start()
for i in range(self.DOWNLOAD_WORKERS):
Thread(target=self.download_worker, name="DownloadWorker#{}".format(i + 1)).start()
self.download_workers_list.append(
Thread(
target=self.download_worker,
name="DownloadWorker#{}".format(i + 1)
)
)
self.download_workers_list[-1].start()
self.dispatcher.start()
@ -318,17 +334,23 @@ class Client:
if not self.is_started:
raise ConnectionError("Client is already stopped")
self.is_started = False
self.session.stop()
for _ in range(self.UPDATES_WORKERS):
self.updates_queue.put(None)
for i in self.updates_workers_list:
i.join()
for _ in range(self.DOWNLOAD_WORKERS):
self.download_queue.put(None)
for i in self.download_workers_list:
i.join()
self.dispatcher.stop()
self.is_started = False
self.session.stop()
Syncer.remove(self)
def authorize_bot(self):

16
pyrogram/client/dispatcher/dispatcher.py
View File

@ -46,20 +46,28 @@ class Dispatcher:
def __init__(self, client, workers):
self.client = client
self.workers = workers
self.workers_list = []
self.updates = Queue()
self.groups = OrderedDict()
def start(self):
for i in range(self.workers):
Thread(
target=self.update_worker,
name="UpdateWorker#{}".format(i + 1)
).start()
self.workers_list.append(
Thread(
target=self.update_worker,
name="UpdateWorker#{}".format(i + 1)
)
)
self.workers_list[-1].start()
def stop(self):
for _ in range(self.workers):
self.updates.put(None)
for i in self.workers_list:
i.join()
def add_handler(self, handler, group: int):
if group not in self.groups:
self.groups[group] = []