slipstream.utils
================

.. py:module:: slipstream.utils

.. autoapi-nested-parse::

   Slipstream utilities.



Attributes
----------

.. autoapisummary::

   slipstream.utils.T
   slipstream.utils.P
   slipstream.utils.AsyncCallable
   slipstream.utils.Pipe


Classes
-------

.. autoapisummary::

   slipstream.utils.Signal
   slipstream.utils.Singleton
   slipstream.utils.PubSub
   slipstream.utils.AsyncSynchronizedGenerator


Functions
---------

.. autoapisummary::

   slipstream.utils.awaitable
   slipstream.utils.iscoroutinecallable
   slipstream.utils.get_param_names


Module Contents
---------------

.. py:data:: T

.. py:data:: P

.. py:type:: AsyncCallable
   :canonical: Callable[P, T | Awaitable[T]]


.. py:type:: Pipe
   :canonical: Callable[[AsyncIterable[Any]], AsyncIterable[Any]]


.. py:class:: Signal

   Bases: :py:obj:`enum.Enum`


   Signals can be exchanged with streams.

   SENTINEL represents an absent yield value
   PAUSE    represents the signal to pause stream
   RESUME   represents the signal to resume stream


   .. py:attribute:: SENTINEL
      :value: 0



   .. py:attribute:: PAUSE
      :value: 1



   .. py:attribute:: RESUME
      :value: 2



   .. py:attribute:: STOP
      :value: 3



.. py:function:: awaitable(x: Any) -> Any
   :async:


   Convert into awaitable.


.. py:function:: iscoroutinecallable(o: Any) -> bool

   Check whether object is coroutine.


.. py:function:: get_param_names(o: Any) -> tuple[str, Ellipsis]

   Return function parameter names.


.. py:class:: Singleton

   Bases: :py:obj:`type`


   Maintain a single instance of a class.


.. py:class:: PubSub

   Singleton publish subscribe pattern class.


   .. py:method:: subscribe(topic: str, listener: AsyncCallable) -> None

      Subscribe callable to topic.



   .. py:method:: unsubscribe(topic: str, listener: AsyncCallable) -> None

      Unsubscribe callable from topic.



   .. py:method:: publish(topic: str, *args: Any, **kwargs: Any) -> None

      Publish message to subscribers of topic.



   .. py:method:: apublish(topic: str, *args: Any, **kwargs: Any) -> None
      :async:


      Publish message to subscribers of topic.



   .. py:method:: iter_topic(topic: str) -> collections.abc.AsyncIterator[Any]
      :async:


      Asynchronously iterate over messages published to a topic.



.. py:class:: AsyncSynchronizedGenerator(gen: collections.abc.AsyncIterable[Any])

   Async generator that synchronizes values across copies.


   .. py:property:: value
      :type: Any


      Get current value the generator is holding.



   .. py:method:: copy() -> _GeneratorCopy

      Create a synchronized copy of this generator.