def register(self, fileobj, events, data=None):
        """
        Register a file object or a :class:`~asynctest.FileMock`.

        If a real selector object has been supplied to the
        :class:`~asynctest.TestSelector` object and ``fileobj`` is not
        a :class:`~asynctest.FileMock` or a :class:`~asynctest.FileDescriptor`
        returned by :meth:`FileMock.fileno()`, the object will be registered to
        the real selector.

        See :meth:`selectors.BaseSelector.register`.
        """
        if isfilemock(fileobj) or self._selector is None:
            key = super().register(fileobj, events, data)
        else:
            key = self._selector.register(fileobj, events, data)

        return key 

def process_decision_events(self, decision_events: DecisionEvents):
        self.decision_context.set_replaying(decision_events.replay)
        self.decision_context.set_replay_current_time_milliseconds(decision_events.replay_current_time_milliseconds)

        self.handle_decision_task_started(decision_events)
        for event in decision_events.markers:
            if not event.marker_recorded_event_attributes.marker_name == LOCAL_ACTIVITY_MARKER_NAME:
                self.process_event(event);
        for event in decision_events.events:
            self.process_event(event)
        if self.completed:
            return
        self.unblock_all()
        self.event_loop.run_event_loop_once()
        if decision_events.replay:
            self.notify_decision_sent()
        for event in decision_events.decision_events:
            self.process_event(event) 

def _patch_handle():
    """Patch Handle to allow recursive calls."""

    def update_from_context(ctx):
        """Copy context ctx to currently active context."""
        for var in ctx:
            var.set(ctx[var])

    def run(self):
        """
        Run the callback in a sub-context, then copy any sub-context vars
        over to the Handle's context.
        """
        try:
            ctx = self._context.copy()
            ctx.run(self._callback, *self._args)
            if ctx:
                self._context.run(update_from_context, ctx)
        except Exception as exc:
            cb = format_helpers._format_callback_source(
                self._callback, self._args)
            msg = 'Exception in callback {}'.format(cb)
            context = {
                'message': msg,
                'exception': exc,
                'handle': self,
            }
            if self._source_traceback:
                context['source_traceback'] = self._source_traceback
            self._loop.call_exception_handler(context)
        self = None

    if sys.version_info >= (3, 7, 0):
        from asyncio import format_helpers
        events.Handle._run = run 

def _get_event_loop():
            current_loop = _get_running_loop()
            if current_loop is not None:
                return current_loop
            return asyncio.events.get_event_loop_policy().get_event_loop() 

def set_read_ready(fileobj, loop):
    """
    Schedule callbacks registered on ``loop`` as if the selector notified that
    data is ready to be read on ``fileobj``.

    :param fileobj: file object or :class:`~asynctest.FileMock` on which the
                    event is mocked.

    :param loop: :class:`asyncio.SelectorEventLoop` watching for events on
                 ``fileobj``.

    ::

        mock = asynctest.SocketMock()
        mock.recv.return_value = b"Data"

        def read_ready(sock):
            print("received:", sock.recv(1024))

        loop.add_reader(mock, read_ready, mock)

        set_read_ready(mock, loop)

        loop.run_forever() # prints received: b"Data"

    .. versionadded:: 0.4
    """
    # since the selector would notify of events at the beginning of the next
    # iteration, we let this iteration finish before actually scheduling the
    # reader (hence the call_soon)
    loop.call_soon_threadsafe(_set_event_ready, fileobj, loop, selectors.EVENT_READ) 

def set_write_ready(fileobj, loop):
    """
    Schedule callbacks registered on ``loop`` as if the selector notified that
    data can be written to ``fileobj``.

    :param fileobj: file object or  :class:`~asynctest.FileMock` on which th
        event is mocked.
    :param loop: :class:`asyncio.SelectorEventLoop` watching for events on
        ``fileobj``.

    .. versionadded:: 0.4
    """
    loop.call_soon_threadsafe(_set_event_ready, fileobj, loop, selectors.EVENT_WRITE) 

def modify(self, fileobj, events, data=None):
        """
        Shortcut when calling :meth:`TestSelector.unregister` then
        :meth:`TestSelector.register` to update the registration of a an object
        to the selector.

        See :meth:`selectors.BaseSelector.modify`.
        """
        if isfilemock(fileobj) or self._selector is None:
            key = super().modify(fileobj, events, data)
        else:
            key = self._selector.modify(fileobj, events, data)

        return key 

def _format_callback(handle):
        return asyncio.events._format_callback(handle._callback, handle._args,
                                               None) 

def _format_event(event):
    callbacks = []

    if event.events & selectors.EVENT_READ:
        callbacks.append("add_reader({}, {})".format(
            event.fileobj, _format_callback(event.data[0])))

    if event.events & selectors.EVENT_WRITE:
        callbacks.append("add_writer({}, {})".format(
            event.fileobj, _format_callback(event.data[1])))

    return callbacks 

def fail_on_active_selector_callbacks(case):
    ignored_events = case._active_selector_callbacks
    active_events = get_registered_events(case.loop._selector)

    output = ["some events watched during the tests were not removed:"]
    for c in map(_format_event, active_events - ignored_events):
        output.extend(c)

    if len(output) > 1:
        case.fail("\n - ".join(output)) 

def _get_event_loop():
                current_loop = _get_running_loop()
                if current_loop is not None:
                    return current_loop
                return asyncio.events.get_event_loop_policy().get_event_loop() 

def __init__(self, events: List[HistoryEvent]):
        self.events = peekable(events) 

def has_next(self) -> bool:
        try:
            self.events.peek()
            return True
        except StopIteration:
            return False 

def next(self) -> Optional[DecisionEvents]:
        events = self.events
        if not self.has_next():
            return None
        decision_events: List[HistoryEvent] = []
        new_events: List[HistoryEvent] = []
        replay = True
        next_decision_event_id = -1
        # noinspection PyUnusedLocal
        event: HistoryEvent
        for event in events:
            event_type = event.event_type
            if event_type == EventType.DecisionTaskStarted or not self.has_next():
                replay_current_time_milliseconds = nano_to_milli(event.timestamp)
                if not self.has_next():
                    replay = False
                    next_decision_event_id = event.event_id + 2
                    break
                peeked: HistoryEvent = events.peek()
                peeked_type = peeked.event_type
                if peeked_type == EventType.DecisionTaskTimedOut or peeked_type == EventType.DecisionTaskFailed:
                    continue
                elif peeked_type == EventType.DecisionTaskCompleted:
                    next(events)
                    next_decision_event_id = peeked.event_id + 1
                    break
                else:
                    raise Exception(
                        "Unexpected event after DecisionTaskStarted: {}".format(peeked))
            new_events.append(event)
        while self.has_next():
            if not is_decision_event(events.peek()):
                break
            decision_events.append(next(events))
        result = DecisionEvents(new_events, decision_events, replay,
                                replay_current_time_milliseconds, next_decision_event_id)
        logger.debug("HistoryHelper next=%s", result)
        return result 

def decide(self, events: List[HistoryEvent]):
        helper = HistoryHelper(events)
        while helper.has_next():
            decision_events = helper.next()
            self.process_decision_events(decision_events)
        return self.get_decisions() 

def process_task(self, decision_task: PollForDecisionTaskResponse) -> List[Decision]:
        execution_id = str(decision_task.workflow_execution)
        decider = ReplayDecider(execution_id, decision_task.workflow_type, self.worker)
        decisions: List[Decision] = decider.decide(decision_task.history.events)
        decider.destroy()
        return decisions 

def process_query(self, decision_task: PollForDecisionTaskResponse) -> bytes:
        execution_id = str(decision_task.workflow_execution)
        decider = ReplayDecider(execution_id, decision_task.workflow_type, self.worker)
        decider.decide(decision_task.history.events)
        try:
            result = decider.query(decision_task, decision_task.query)
            return json.dumps(result)
        finally:
            decider.destroy() 

def _patch_loop(loop):
    """Patch loop to make it reentrent."""

    def run_until_complete(self, future):
        self._check_closed()
        events._set_running_loop(self)
        f = asyncio.ensure_future(future, loop=self)
        if f is not future:
            f._log_destroy_pending = False
        while not f.done():
            self._run_once()
            if self._stopping:
                break
        if not f.done():
            raise RuntimeError('Event loop stopped before Future completed.')
        return f.result()

    def _run_once(self):
        """
        Simplified re-implementation of asyncio's _run_once that
        runs handles as they become ready.
        """
        now = self.time()
        ready = self._ready
        scheduled = self._scheduled
        while scheduled and scheduled[0]._cancelled:
            heappop(scheduled)

        timeout = 0 if ready or self._stopping \
            else min(max(0, scheduled[0]._when - now), 10) if scheduled \
            else None
        event_list = self._selector.select(timeout)
        self._process_events(event_list)

        while scheduled and scheduled[0]._when < now + self._clock_resolution:
            handle = heappop(scheduled)
            ready.append(handle)

        while ready:
            handle = ready.popleft()
            if not handle._cancelled:
                handle._run()
        handle = None

    def _check_running(self):
        """Do not throw exception if loop is already running."""
        pass

    cls = loop.__class__
    cls._run_once_orig = cls._run_once
    cls._run_once = _run_once
    cls._run_until_complete_orig = cls.run_until_complete
    cls.run_until_complete = run_until_complete
    cls._check_running = _check_running
    cls._check_runnung = _check_running  # typo in Python 3.7 source
    cls._nest_patched = True