2017-09-30 11:12:01 +03:00
|
|
|
import logging
|
2017-09-07 19:49:08 +03:00
|
|
|
from collections import deque
|
2017-09-07 21:17:40 +03:00
|
|
|
from datetime import datetime
|
2017-09-30 11:12:01 +03:00
|
|
|
from threading import RLock, Event, Thread
|
2017-09-07 21:17:40 +03:00
|
|
|
|
|
|
|
from .tl import types as tl
|
2017-09-07 19:49:08 +03:00
|
|
|
|
|
|
|
|
|
|
|
class UpdateState:
|
|
|
|
"""Used to hold the current state of processed updates.
|
2017-09-08 13:54:38 +03:00
|
|
|
To retrieve an update, .poll() should be called.
|
2017-09-07 19:49:08 +03:00
|
|
|
"""
|
2017-09-30 12:21:07 +03:00
|
|
|
WORKER_POLL_TIMEOUT = 5.0 # Avoid waiting forever on the workers
|
|
|
|
|
2017-09-30 12:17:31 +03:00
|
|
|
def __init__(self, workers=None):
|
|
|
|
"""
|
|
|
|
:param workers: This integer parameter has three possible cases:
|
|
|
|
workers is None: Updates will *not* be stored on self.
|
|
|
|
workers = 0: Another thread is responsible for calling self.poll()
|
|
|
|
workers > 0: 'workers' background threads will be spawned, any
|
|
|
|
any of them will invoke all the self.handlers.
|
|
|
|
"""
|
|
|
|
self._workers = workers
|
2017-09-30 11:12:01 +03:00
|
|
|
self._worker_threads = []
|
|
|
|
|
2017-09-07 19:58:54 +03:00
|
|
|
self.handlers = []
|
2017-09-07 21:17:40 +03:00
|
|
|
self._updates_lock = RLock()
|
2017-09-07 19:49:08 +03:00
|
|
|
self._updates_available = Event()
|
|
|
|
self._updates = deque()
|
|
|
|
|
2017-09-30 11:12:01 +03:00
|
|
|
self._logger = logging.getLogger(__name__)
|
|
|
|
|
2017-09-07 21:17:40 +03:00
|
|
|
# https://core.telegram.org/api/updates
|
|
|
|
self._state = tl.updates.State(0, 0, datetime.now(), 0, 0)
|
2017-09-30 11:12:01 +03:00
|
|
|
|
2017-09-08 13:54:38 +03:00
|
|
|
def can_poll(self):
|
|
|
|
"""Returns True if a call to .poll() won't lock"""
|
2017-09-07 19:49:08 +03:00
|
|
|
return self._updates_available.is_set()
|
|
|
|
|
2017-09-30 12:21:07 +03:00
|
|
|
def poll(self, timeout=None):
|
|
|
|
"""Polls an update or blocks until an update object is available.
|
|
|
|
If 'timeout is not None', it should be a floating point value,
|
|
|
|
and the method will 'return None' if waiting times out.
|
|
|
|
"""
|
|
|
|
if not self._updates_available.wait(timeout=timeout):
|
|
|
|
return
|
|
|
|
|
2017-09-07 19:49:08 +03:00
|
|
|
with self._updates_lock:
|
2017-09-30 11:59:33 +03:00
|
|
|
if not self._updates_available.is_set():
|
|
|
|
return
|
|
|
|
|
2017-09-07 19:49:08 +03:00
|
|
|
update = self._updates.popleft()
|
|
|
|
if not self._updates:
|
|
|
|
self._updates_available.clear()
|
|
|
|
|
2017-09-18 11:59:54 +03:00
|
|
|
if isinstance(update, Exception):
|
2017-09-30 19:39:31 +03:00
|
|
|
raise update # Some error was set through (surely StopIteration)
|
2017-09-18 11:59:54 +03:00
|
|
|
|
|
|
|
return update
|
2017-09-07 19:49:08 +03:00
|
|
|
|
2017-09-30 11:12:01 +03:00
|
|
|
def get_workers(self):
|
|
|
|
return self._workers
|
|
|
|
|
|
|
|
def set_workers(self, n):
|
2017-09-30 12:17:31 +03:00
|
|
|
"""Changes the number of workers running.
|
|
|
|
If 'n is None', clears all pending updates from memory.
|
|
|
|
"""
|
2017-10-01 20:56:24 +03:00
|
|
|
self.stop_workers()
|
2017-09-30 11:12:01 +03:00
|
|
|
self._workers = n
|
2017-09-30 12:17:31 +03:00
|
|
|
if n is None:
|
|
|
|
self._updates.clear()
|
|
|
|
else:
|
2017-10-01 20:56:24 +03:00
|
|
|
self.setup_workers()
|
2017-09-30 11:12:01 +03:00
|
|
|
|
|
|
|
workers = property(fget=get_workers, fset=set_workers)
|
|
|
|
|
2017-10-01 20:56:24 +03:00
|
|
|
def stop_workers(self):
|
2017-09-30 11:12:01 +03:00
|
|
|
"""Raises "StopIterationException" on the worker threads to stop them,
|
|
|
|
and also clears all of them off the list
|
|
|
|
"""
|
2017-09-30 19:44:37 +03:00
|
|
|
if self._workers:
|
|
|
|
with self._updates_lock:
|
|
|
|
# Insert at the beginning so the very next poll causes an error
|
|
|
|
# on all the worker threads
|
|
|
|
# TODO Should this reset the pts and such?
|
|
|
|
for _ in range(self._workers):
|
|
|
|
self._updates.appendleft(StopIteration())
|
|
|
|
self._updates_available.set()
|
2017-09-30 19:39:31 +03:00
|
|
|
|
2017-09-30 11:12:01 +03:00
|
|
|
for t in self._worker_threads:
|
|
|
|
t.join()
|
|
|
|
|
|
|
|
self._worker_threads.clear()
|
|
|
|
|
2017-10-01 20:56:24 +03:00
|
|
|
def setup_workers(self):
|
2017-09-30 12:17:31 +03:00
|
|
|
if self._worker_threads or not self._workers:
|
|
|
|
# There already are workers, or workers is None or 0. Do nothing.
|
2017-09-30 11:12:01 +03:00
|
|
|
return
|
|
|
|
|
|
|
|
for i in range(self._workers):
|
|
|
|
thread = Thread(
|
|
|
|
target=UpdateState._worker_loop,
|
|
|
|
name='UpdateWorker{}'.format(i),
|
|
|
|
daemon=True,
|
|
|
|
args=(self, i)
|
|
|
|
)
|
|
|
|
self._worker_threads.append(thread)
|
|
|
|
thread.start()
|
|
|
|
|
|
|
|
def _worker_loop(self, wid):
|
|
|
|
while True:
|
|
|
|
try:
|
2017-09-30 12:21:07 +03:00
|
|
|
update = self.poll(timeout=UpdateState.WORKER_POLL_TIMEOUT)
|
2017-09-30 11:12:01 +03:00
|
|
|
# TODO Maybe people can add different handlers per update type
|
2017-09-30 12:21:07 +03:00
|
|
|
if update:
|
|
|
|
for handler in self.handlers:
|
|
|
|
handler(update)
|
2017-09-30 11:12:01 +03:00
|
|
|
except StopIteration:
|
|
|
|
break
|
|
|
|
except Exception as e:
|
|
|
|
# We don't want to crash a worker thread due to any reason
|
|
|
|
self._logger.debug(
|
|
|
|
'[ERROR] Unhandled exception on worker {}'.format(wid), e
|
|
|
|
)
|
|
|
|
|
2017-09-07 19:49:08 +03:00
|
|
|
def process(self, update):
|
|
|
|
"""Processes an update object. This method is normally called by
|
|
|
|
the library itself.
|
|
|
|
"""
|
2017-09-30 12:17:31 +03:00
|
|
|
if self._workers is None:
|
|
|
|
return # No processing needs to be done if nobody's working
|
2017-09-07 21:17:40 +03:00
|
|
|
|
|
|
|
with self._updates_lock:
|
|
|
|
if isinstance(update, tl.updates.State):
|
|
|
|
self._state = update
|
2017-09-19 14:17:40 +03:00
|
|
|
return # Nothing else to be done
|
2017-09-07 19:58:54 +03:00
|
|
|
|
2017-09-19 14:17:40 +03:00
|
|
|
pts = getattr(update, 'pts', self._state.pts)
|
2017-09-23 00:12:36 +03:00
|
|
|
if hasattr(update, 'pts') and pts <= self._state.pts:
|
2017-09-19 14:17:40 +03:00
|
|
|
return # We already handled this update
|
|
|
|
|
|
|
|
self._state.pts = pts
|
2017-10-01 17:30:27 +03:00
|
|
|
|
|
|
|
if type(update).SUBCLASS_OF_ID == 0x8af52aac: # crc32(b'Updates')
|
|
|
|
# Expand "Updates" into "Update", and pass these to callbacks.
|
|
|
|
# Since .users and .chats have already been processed, we
|
|
|
|
# don't need to care about those either.
|
|
|
|
if isinstance(update, tl.UpdateShort):
|
|
|
|
self._updates.append(update.update)
|
|
|
|
self._updates_available.set()
|
|
|
|
|
|
|
|
elif isinstance(update, tl.Updates) or \
|
|
|
|
isinstance(update, tl.UpdatesCombined):
|
|
|
|
self._updates.extend(update.updates)
|
|
|
|
self._updates_available.set()
|
|
|
|
|
|
|
|
elif not isinstance(update, tl.UpdatesTooLong):
|
|
|
|
# TODO Handle "Updates too long"
|
|
|
|
self._updates.append(update)
|
|
|
|
self._updates_available.set()
|
|
|
|
|
|
|
|
elif type(update).SUBCLASS_OF_ID == 0x9f89304e: # crc32(b'Update')
|
|
|
|
self._updates.append(update)
|
|
|
|
self._updates_available.set()
|
|
|
|
else:
|
|
|
|
self._logger.debug('Ignoring "update" of type {}'.format(
|
|
|
|
type(update).__name__)
|
|
|
|
)
|