# Path: src/apscheduler/abc.py
# class Subscription(metaclass=ABCMeta):
#     """
#     Represents a subscription with an event source.
# 
#     If used as a context manager, unsubscribes on exit.
#     """
# 
#     def __enter__(self) -> Subscription:
#         return self
# 
#     def __exit__(self, exc_type, exc_val, exc_tb) -> None:
#         self.unsubscribe()
# 
#     @abstractmethod
#     def unsubscribe(self) -> None:
#         """
#         Cancel this subscription.
# 
#         Does nothing if the subscription has already been cancelled.
#         """
# 
# Path: src/apscheduler/events.py
# class Event:
#     timestamp: datetime = attrs.field(factory=partial(datetime.now, timezone.utc),
#                                       converter=as_aware_datetime)
# 
# Path: src/apscheduler/util.py
# def reentrant(cls: type[T]) -> type[T]:
#     """
#     Modifies a class so that its ``__enter__`` / ``__exit__`` (or ``__aenter__`` / ``__aexit__``)
#     methods track the number of times it has been entered and exited and only actually invoke
#     the ``__enter__()`` method on the first entry and ``__exit__()`` on the last exit.
# 
#     """
# 
#     def __enter__(self: T) -> T:
#         loans[self] += 1
#         if loans[self] == 1:
#             previous_enter(self)
# 
#         return self
# 
#     def __exit__(self, exc_type, exc_val, exc_tb) -> None:
#         assert loans[self]
#         loans[self] -= 1
#         if loans[self] == 0:
#             return previous_exit(self, exc_type, exc_val, exc_tb)
# 
#     async def __aenter__(self: T) -> T:
#         loans[self] += 1
#         if loans[self] == 1:
#             await previous_aenter(self)
# 
#         return self
# 
#     async def __aexit__(self, exc_type, exc_val, exc_tb):
#         assert loans[self]
#         loans[self] -= 1
#         if loans[self] == 0:
#             del loans[self]
#             return await previous_aexit(self, exc_type, exc_val, exc_tb)
# 
#     loans: dict[T, int] = defaultdict(lambda: 0)
#     previous_enter: Callable = getattr(cls, '__enter__', None)
#     previous_exit: Callable = getattr(cls, '__exit__', None)
#     previous_aenter: Callable = getattr(cls, '__aenter__', None)
#     previous_aexit: Callable = getattr(cls, '__aexit__', None)
#     if previous_enter and previous_exit:
#         cls.__enter__ = __enter__
#         cls.__exit__ = __exit__
#     elif previous_aenter and previous_aexit:
#         cls.__aenter__ = __aenter__
#         cls.__aexit__ = __aexit__
# 
#     return cls
# 
# Path: src/apscheduler/eventbrokers/base.py
# class BaseEventBroker(EventBroker):
#     _logger: Logger = attrs.field(init=False)
#     _subscriptions: dict[object, LocalSubscription] = attrs.field(init=False, factory=dict)
# 
#     def __attrs_post_init__(self) -> None:
#         self._logger = getLogger(self.__class__.__module__)
# 
#     def subscribe(self, callback: Callable[[Event], Any],
#                   event_types: Iterable[type[Event]] | None = None, *,
#                   one_shot: bool = False) -> Subscription:
#         types = set(event_types) if event_types else None
#         token = object()
#         subscription = LocalSubscription(callback, types, one_shot, token, self)
#         self._subscriptions[token] = subscription
#         return subscription
# 
#     def unsubscribe(self, token: object) -> None:
#         self._subscriptions.pop(token, None)


from asyncio import iscoroutinefunction
from concurrent.futures import ThreadPoolExecutor
from contextlib import ExitStack
from threading import Lock
from typing import Any, Callable, Iterable
from ..abc import Subscription
from ..events import Event
from ..util import reentrant
from .base import BaseEventBroker
import attrs
from __future__ import annotations





@reentrant
@attrs.define(eq=False)
class LocalEventBroker(BaseEventBroker):
    _executor: ThreadPoolExecutor = attrs.field(init=False)
    _exit_stack: ExitStack = attrs.field(init=False)
    _subscriptions_lock: Lock = attrs.field(init=False, factory=Lock)

    def __enter__(self):
        self._exit_stack = ExitStack()
        self._executor = self._exit_stack.enter_context(ThreadPoolExecutor(1))
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self._exit_stack.__exit__(exc_type, exc_val, exc_tb)
        del self._executor


# Path: src/apscheduler/abc.py
# class Subscription(metaclass=ABCMeta):
#     """
#     Represents a subscription with an event source.
# 
#     If used as a context manager, unsubscribes on exit.
#     """
# 
#     def __enter__(self) -> Subscription:
#         return self
# 
#     def __exit__(self, exc_type, exc_val, exc_tb) -> None:
#         self.unsubscribe()
# 
#     @abstractmethod
#     def unsubscribe(self) -> None:
#         """
#         Cancel this subscription.
# 
#         Does nothing if the subscription has already been cancelled.
#         """
# 
# Path: src/apscheduler/events.py
# class Event:
#     timestamp: datetime = attrs.field(factory=partial(datetime.now, timezone.utc),
#                                       converter=as_aware_datetime)
# 
# Path: src/apscheduler/util.py
# def reentrant(cls: type[T]) -> type[T]:
#     """
#     Modifies a class so that its ``__enter__`` / ``__exit__`` (or ``__aenter__`` / ``__aexit__``)
#     methods track the number of times it has been entered and exited and only actually invoke
#     the ``__enter__()`` method on the first entry and ``__exit__()`` on the last exit.
# 
#     """
# 
#     def __enter__(self: T) -> T:
#         loans[self] += 1
#         if loans[self] == 1:
#             previous_enter(self)
# 
#         return self
# 
#     def __exit__(self, exc_type, exc_val, exc_tb) -> None:
#         assert loans[self]
#         loans[self] -= 1
#         if loans[self] == 0:
#             return previous_exit(self, exc_type, exc_val, exc_tb)
# 
#     async def __aenter__(self: T) -> T:
#         loans[self] += 1
#         if loans[self] == 1:
#             await previous_aenter(self)
# 
#         return self
# 
#     async def __aexit__(self, exc_type, exc_val, exc_tb):
#         assert loans[self]
#         loans[self] -= 1
#         if loans[self] == 0:
#             del loans[self]
#             return await previous_aexit(self, exc_type, exc_val, exc_tb)
# 
#     loans: dict[T, int] = defaultdict(lambda: 0)
#     previous_enter: Callable = getattr(cls, '__enter__', None)
#     previous_exit: Callable = getattr(cls, '__exit__', None)
#     previous_aenter: Callable = getattr(cls, '__aenter__', None)
#     previous_aexit: Callable = getattr(cls, '__aexit__', None)
#     if previous_enter and previous_exit:
#         cls.__enter__ = __enter__
#         cls.__exit__ = __exit__
#     elif previous_aenter and previous_aexit:
#         cls.__aenter__ = __aenter__
#         cls.__aexit__ = __aexit__
# 
#     return cls
# 
# Path: src/apscheduler/eventbrokers/base.py
# class BaseEventBroker(EventBroker):
#     _logger: Logger = attrs.field(init=False)
#     _subscriptions: dict[object, LocalSubscription] = attrs.field(init=False, factory=dict)
# 
#     def __attrs_post_init__(self) -> None:
#         self._logger = getLogger(self.__class__.__module__)
# 
#     def subscribe(self, callback: Callable[[Event], Any],
#                   event_types: Iterable[type[Event]] | None = None, *,
#                   one_shot: bool = False) -> Subscription:
#         types = set(event_types) if event_types else None
#         token = object()
#         subscription = LocalSubscription(callback, types, one_shot, token, self)
#         self._subscriptions[token] = subscription
#         return subscription
# 
#     def unsubscribe(self, token: object) -> None:
#         self._subscriptions.pop(token, None)


# Path: src/apscheduler/eventbrokers/local.py
from asyncio import iscoroutinefunction
from concurrent.futures import ThreadPoolExecutor
from contextlib import ExitStack
from threading import Lock
from typing import Any, Callable, Iterable
from ..abc import Subscription
from ..events import Event
from ..util import reentrant
from .base import BaseEventBroker
import attrs

from __future__ import annotations





@reentrant
@attrs.define(eq=False)
class LocalEventBroker(BaseEventBroker):
    _executor: ThreadPoolExecutor = attrs.field(init=False)
    _exit_stack: ExitStack = attrs.field(init=False)
    _subscriptions_lock: Lock = attrs.field(init=False, factory=Lock)

    def __enter__(self):
        self._exit_stack = ExitStack()
        self._executor = self._exit_stack.enter_context(ThreadPoolExecutor(1))
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self._exit_stack.__exit__(exc_type, exc_val, exc_tb)
        del self._executor


    def subscribe(self, callback: Callable[[Event], Any],