"""
This type stub file was generated by pyright.
"""

import asyncio
import concurrent.futures
import multiprocessing
import multiprocessing.managers
import queue
import threading
import temporalio.activity
import temporalio.bridge.worker
import temporalio.common
import temporalio.converter
from abc import ABC, abstractmethod
from contextlib import contextmanager
from dataclasses import dataclass
from typing import Any, Callable, Iterator, Optional, Sequence, Tuple, Type
from ._interceptor import ActivityInboundInterceptor, ActivityOutboundInterceptor, ExecuteActivityInput, Interceptor

"""
This type stub file was generated by pyright.
"""
logger = ...
class _ActivityWorker:
    def __init__(self, *, bridge_worker: Callable[[], temporalio.bridge.worker.Worker], task_queue: str, activities: Sequence[Callable], activity_executor: Optional[concurrent.futures.Executor], shared_state_manager: Optional[SharedStateManager], data_converter: temporalio.converter.DataConverter, interceptors: Sequence[Interceptor], metric_meter: temporalio.common.MetricMeter) -> None:
        ...
    
    async def run(self) -> None:
        ...
    
    def notify_shutdown(self) -> None:
        ...
    
    async def drain_poll_queue(self) -> None:
        ...
    
    async def wait_all_completed(self) -> None:
        ...
    


@dataclass
class _RunningActivity:
    pending_heartbeats: asyncio.Queue[Sequence[Any]]
    info: Optional[temporalio.activity.Info] = ...
    task: Optional[asyncio.Task] = ...
    cancelled_event: Optional[temporalio.activity._CompositeEvent] = ...
    last_heartbeat_task: Optional[asyncio.Task] = ...
    cancel_thread_raiser: Optional[_ThreadExceptionRaiser] = ...
    sync: bool = ...
    done: bool = ...
    cancelled_by_request: bool = ...
    cancelled_due_to_heartbeat_error: Optional[Exception] = ...
    def cancel(self, *, cancelled_by_request: bool = ..., cancelled_due_to_heartbeat_error: Optional[Exception] = ...) -> None:
        ...
    


class _ThreadExceptionRaiser:
    def __init__(self) -> None:
        ...
    
    def set_thread_id(self, thread_id: int) -> None:
        ...
    
    def raise_in_thread(self, exc_type: Type[Exception]) -> None:
        ...
    
    @contextmanager
    def shielded(self) -> Iterator[None]:
        ...
    


class _ActivityInboundImpl(ActivityInboundInterceptor):
    def __init__(self, worker: _ActivityWorker, running_activity: _RunningActivity) -> None:
        ...
    
    def init(self, outbound: ActivityOutboundInterceptor) -> None:
        ...
    
    async def execute_activity(self, input: ExecuteActivityInput) -> Any:
        ...
    


class _ActivityOutboundImpl(ActivityOutboundInterceptor):
    def __init__(self, worker: _ActivityWorker, info: temporalio.activity.Info) -> None:
        ...
    
    def info(self) -> temporalio.activity.Info:
        ...
    
    def heartbeat(self, *details: Any) -> None:
        ...
    


class SharedStateManager(ABC):
    """Base class for a shared state manager providing cross-process-safe
    primitives for use by activity executors.

    Cross-worker use of the shared state manager is encouraged.
    :py:meth:`create_from_multiprocessing` provides the commonly used
    implementation.
    """
    @staticmethod
    def create_from_multiprocessing(mgr: multiprocessing.managers.SyncManager, queue_poller_executor: Optional[concurrent.futures.Executor] = ...) -> SharedStateManager:
        """Create a shared state manager from a multiprocessing manager.

        Args:
            mgr: Sync manager to create primitives from. This is usually
                :py:func:`multiprocessing.Manager`.
            queue_poller_executor: The executor used when running the
                synchronous heartbeat queue poller. This should be a
                :py:class:`concurrent.futures.ThreadPoolExecutor`. If unset, a
                thread pool executor is created with max-workers of 1.

        Returns:
            The shared state manager.
        """
        ...
    
    @abstractmethod
    def new_event(self) -> threading.Event:
        """Create a threading.Event that can be used across processes."""
        ...
    
    @abstractmethod
    async def register_heartbeater(self, task_token: bytes, heartbeat: Callable[..., None]) -> SharedHeartbeatSender:
        """Register a heartbeat function.

        Args:
            task_token: Unique task token for the heartbeater.
            heartbeat: Function that should be called when the resulting sender
                is sent a heartbeat.

        Returns:
            A sender that can be pickled for use in another process.
        """
        ...
    
    @abstractmethod
    async def unregister_heartbeater(self, task_token: bytes) -> None:
        """Unregisters a previously registered heartbeater for the task
        token. This should also flush any pending heartbeats.
        """
        ...
    


class SharedHeartbeatSender(ABC):
    """Base class for a heartbeat sender that is picklable for use in another
    process.
    """
    @abstractmethod
    def send_heartbeat(self, task_token: bytes, *details: Any) -> None:
        """Send a heartbeat for the given task token and details."""
        ...
    


_multiprocess_heartbeat_complete = ...
class _MultiprocessingSharedStateManager(SharedStateManager):
    def __init__(self, mgr: multiprocessing.managers.SyncManager, queue_poller_executor: concurrent.futures.Executor) -> None:
        ...
    
    def new_event(self) -> threading.Event:
        ...
    
    async def register_heartbeater(self, task_token: bytes, heartbeat: Callable[..., None]) -> SharedHeartbeatSender:
        ...
    
    async def unregister_heartbeater(self, task_token: bytes) -> None:
        ...
    


class _MultiprocessingSharedHeartbeatSender(SharedHeartbeatSender):
    def __init__(self, heartbeat_queue: queue.Queue[Tuple[bytes, Sequence[Any]]]) -> None:
        ...
    
    def send_heartbeat(self, task_token: bytes, *details: Any) -> None:
        ...
    


