PubSub

__init__(atomic, strategy) -> None

def __init__(self, atomic: Atomic[A], strategy: Strategy) -> None:
    self._hub = Hub(atomic, strategy)

make(atomic, strategy) -> Effect[PubSub[T]]

@staticmethod
def make[T](atomic: Callable[[], Atomic[T]], strategy: Strategy) -> Effect[PubSub[T]]:
    """A hub over the store ``atomic()`` builds, handling a full store by
    ``strategy`` (Effect's ``make``, with the built-in strategies)."""
    return _Sync(lambda: PubSub(atomic(), strategy))

bounded(capacity, *, replay=None) -> Effect[PubSub[A]]

@classmethod
def bounded(cls, capacity: int | None, *, replay: int | None = None) -> Effect[PubSub[A]]:
    """Publishers wait while ``capacity`` messages are unread. ``replay``
    messages are kept for new subscribers. A capacity below 1 is a
    defect; ``None`` is unbounded."""
    return _Sync(lambda: cls(Atomic(capacity, replay), "suspend"))

dropping(capacity, *, replay=None) -> Effect[PubSub[A]]

@classmethod
def dropping(cls, capacity: int | None, *, replay: int | None = None) -> Effect[PubSub[A]]:
    """A publish to a full hub drops the message and returns ``False``."""
    return _Sync(lambda: cls(Atomic(capacity, replay), "dropping"))

sliding(capacity, *, replay=None) -> Effect[PubSub[A]]

@classmethod
def sliding(cls, capacity: int | None, *, replay: int | None = None) -> Effect[PubSub[A]]:
    """A publish to a full hub drops the oldest unread message."""
    return _Sync(lambda: cls(Atomic(capacity, replay), "sliding"))

unbounded(*, replay=None) -> Effect[PubSub[A]]

@classmethod
def unbounded(cls, *, replay: int | None = None) -> Effect[PubSub[A]]:
    return _Sync(lambda: cls(Atomic(None, replay), "dropping"))

publish(value) -> Effect[bool]

def publish(self, value: A) -> Effect[bool]:
    """Publish to every current subscriber. ``False`` if the hub ended or
    shut down, or a dropping hub is full; a bounded hub waits for room."""

    def register(resume: Callable[[Effect[bool]], None]) -> Effect[None] | None:
        hub = self._hub
        if self.unsafe_publish(value):
            resume(_Succeed(True))
        elif not hub.accepts() or hub.strategy == "dropping":
            resume(_Succeed(False))
        elif hub.strategy == "sliding":
            resume(_Succeed(self._slide_in([value])))
        else:
            return self._park([value], resume)
        return None

    return _Callback(register)

unsafe_publish(value) -> bool

def unsafe_publish(self, value: A) -> bool:
    """Publish without waiting, sliding or dropping: ``False`` if the
    store is full. Call on the scheduler thread of the hub's fibers."""
    hub = self._hub
    if not hub.accepts() or not hub.atomic.publish(value):
        return False
    hub.deliver()
    return True

publish_all(items) -> Effect[bool]

def publish_all(self, items: Iterable[A]) -> Effect[bool]:
    """Publish ``items`` in order; ``True`` if every one was published.
    A bounded hub publishes what fits and waits to publish the rest."""

    def register(resume: Callable[[Effect[bool]], None]) -> Effect[None] | None:
        hub = self._hub
        if not hub.accepts():
            resume(_Succeed(False))
            return None
        surplus = hub.atomic.publish_all(items)
        hub.deliver()
        if not surplus:
            resume(_Succeed(True))
        elif hub.strategy == "dropping":
            resume(_Succeed(False))
        elif hub.strategy == "sliding":
            resume(_Succeed(self._slide_in(surplus)))
        else:
            return self._park(surplus, resume)
        return None

    return _Callback(register)

end(value) -> Effect[bool]

def end(self, value: A) -> Effect[bool]:
    """Publish nothing more. Subscribers take what they have not read,
    then ``value`` from every take; waiting publishes return ``False``.
    ``False`` if the hub already ended or shut down."""
    return _Sync(lambda: self.unsafe_end(value))

unsafe_end(value) -> bool

def unsafe_end(self, value: A) -> bool:
    hub = self._hub
    if not hub.accepts():
        return False
    hub.ended = Some(value)
    hub.reject_publishers()
    hub.wake_takers()
    return True

subscribe() -> Effect[Subscription[A], Never, Scope]

def subscribe(self) -> Effect[Subscription[A], Never, Scope]:
    """Subscribe until the current scope closes (or the hub shuts down).
    The subscription reads the replayed messages, then everything
    published from now on."""
    hub = self._hub

    def body():
        local = yield from Scope()
        scope = yield from hub.scope.fork()
        reader = hub.subscribe()
        yield from scope.add_finalizer(
            lambda _: _WithFiber(lambda fiber: _Sync(lambda: hub.unsubscribe(reader, fiber.id)))
        )
        yield from local.add_finalizer(scope.close)
        return Subscription[A](hub, reader)

    return _Gen(body).uninterruptible()

shutdown() -> Effect[None]

def shutdown(self) -> Effect[None]:
    """Unsubscribe everyone, interrupting their takes. Later publishes
    return ``False``. Waiting publishes complete: with nobody subscribed
    the store empties and admits them (as in Effect, whose strategy
    shutdown then finds no publisher left to interrupt)."""

    def body(fiber: Fiber[Any, Any]) -> Effect[None]:
        hub = self._hub
        if hub.is_shutdown:
            return _Succeed(None)
        hub.is_shutdown = True
        hub.shutdown_waiters.notify_all()
        return hub.scope.close(Failure(interrupt_cause(fiber.id)))

    return _WithFiber(body).uninterruptible()

is_shutdown() -> Effect[bool]

def is_shutdown(self) -> Effect[bool]:
    return _Sync(self.unsafe_is_shutdown)

unsafe_is_shutdown() -> bool

def unsafe_is_shutdown(self) -> bool:
    return self._hub.is_shutdown

await_shutdown() -> Effect[None]

def await_shutdown(self) -> Effect[None]:
    """Wait until the hub is shut down."""
    return _Suspend(lambda: self._hub.shutdown_waiters.wait(self.unsafe_is_shutdown))

size() -> Effect[int]

def size(self) -> Effect[int]:
    """How many messages are stored unread (0 once shut down)."""
    return _Sync(self.unsafe_size)

unsafe_size() -> int

def unsafe_size(self) -> int:
    return 0 if self._hub.is_shutdown else self._hub.atomic.size()

is_full() -> Effect[bool]

def is_full(self) -> Effect[bool]:
    return _Sync(lambda: self.unsafe_size() == self.capacity)

is_empty() -> Effect[bool]

def is_empty(self) -> Effect[bool]:
    return _Sync(lambda: self.unsafe_size() == 0)