:py:mod:`leads.comm.prototype`
==============================

.. py:module:: leads.comm.prototype

.. autodoc2-docstring:: leads.comm.prototype
   :allowtitles:

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

Classes
~~~~~~~

.. list-table::
   :class: autosummary longtable
   :align: left

   * - :py:obj:`Service <leads.comm.prototype.Service>`
     - .. autodoc2-docstring:: leads.comm.prototype.Service
          :summary:
   * - :py:obj:`ConnectionBase <leads.comm.prototype.ConnectionBase>`
     - .. autodoc2-docstring:: leads.comm.prototype.ConnectionBase
          :summary:
   * - :py:obj:`Connection <leads.comm.prototype.Connection>`
     - .. autodoc2-docstring:: leads.comm.prototype.Connection
          :summary:
   * - :py:obj:`Callback <leads.comm.prototype.Callback>`
     - .. autodoc2-docstring:: leads.comm.prototype.Callback
          :summary:
   * - :py:obj:`Entity <leads.comm.prototype.Entity>`
     - .. autodoc2-docstring:: leads.comm.prototype.Entity
          :summary:

Functions
~~~~~~~~~

.. list-table::
   :class: autosummary longtable
   :align: left

   * - :py:obj:`my_ip_addresses <leads.comm.prototype.my_ip_addresses>`
     - .. autodoc2-docstring:: leads.comm.prototype.my_ip_addresses
          :summary:

API
~~~

.. py:function:: my_ip_addresses() -> list[str]
   :canonical: leads.comm.prototype.my_ip_addresses

   .. autodoc2-docstring:: leads.comm.prototype.my_ip_addresses

.. py:class:: Service(port: int)
   :canonical: leads.comm.prototype.Service

   .. autodoc2-docstring:: leads.comm.prototype.Service

   .. rubric:: Initialization

   .. autodoc2-docstring:: leads.comm.prototype.Service.__init__

   .. py:method:: port() -> int
      :canonical: leads.comm.prototype.Service.port

      .. autodoc2-docstring:: leads.comm.prototype.Service.port

   .. py:method:: run(*args, **kwargs) -> None
      :canonical: leads.comm.prototype.Service.run
      :abstractmethod:

      .. autodoc2-docstring:: leads.comm.prototype.Service.run

   .. py:method:: _run(*args, **kwargs) -> None
      :canonical: leads.comm.prototype.Service._run

      .. autodoc2-docstring:: leads.comm.prototype.Service._run

   .. py:method:: _register_process(*args, **kwargs) -> None
      :canonical: leads.comm.prototype.Service._register_process

      .. autodoc2-docstring:: leads.comm.prototype.Service._register_process

   .. py:method:: _parallel_run(*args, **kwargs) -> None
      :canonical: leads.comm.prototype.Service._parallel_run

      .. autodoc2-docstring:: leads.comm.prototype.Service._parallel_run

   .. py:method:: start(parallel: bool = False, *args, **kwargs) -> typing.Self
      :canonical: leads.comm.prototype.Service.start

      .. autodoc2-docstring:: leads.comm.prototype.Service.start

   .. py:method:: close() -> None
      :canonical: leads.comm.prototype.Service.close
      :abstractmethod:

      .. autodoc2-docstring:: leads.comm.prototype.Service.close

.. py:class:: ConnectionBase(remainder: bytes, separator: bytes)
   :canonical: leads.comm.prototype.ConnectionBase

   .. autodoc2-docstring:: leads.comm.prototype.ConnectionBase

   .. rubric:: Initialization

   .. autodoc2-docstring:: leads.comm.prototype.ConnectionBase.__init__

   .. py:method:: drop_remainder() -> None
      :canonical: leads.comm.prototype.ConnectionBase.drop_remainder

      .. autodoc2-docstring:: leads.comm.prototype.ConnectionBase.drop_remainder

   .. py:method:: use_remainder() -> bytes
      :canonical: leads.comm.prototype.ConnectionBase.use_remainder

      .. autodoc2-docstring:: leads.comm.prototype.ConnectionBase.use_remainder

   .. py:method:: with_remainder(msg: bytes) -> bytes
      :canonical: leads.comm.prototype.ConnectionBase.with_remainder

      .. autodoc2-docstring:: leads.comm.prototype.ConnectionBase.with_remainder

   .. py:method:: closed() -> bool
      :canonical: leads.comm.prototype.ConnectionBase.closed
      :abstractmethod:

      .. autodoc2-docstring:: leads.comm.prototype.ConnectionBase.closed

   .. py:method:: receive() -> bytes | None
      :canonical: leads.comm.prototype.ConnectionBase.receive
      :abstractmethod:

      .. autodoc2-docstring:: leads.comm.prototype.ConnectionBase.receive

   .. py:method:: send(msg: bytes) -> None
      :canonical: leads.comm.prototype.ConnectionBase.send
      :abstractmethod:

      .. autodoc2-docstring:: leads.comm.prototype.ConnectionBase.send

   .. py:method:: disconnect() -> None
      :canonical: leads.comm.prototype.ConnectionBase.disconnect

      .. autodoc2-docstring:: leads.comm.prototype.ConnectionBase.disconnect

   .. py:method:: close() -> None
      :canonical: leads.comm.prototype.ConnectionBase.close
      :abstractmethod:

      .. autodoc2-docstring:: leads.comm.prototype.ConnectionBase.close

.. py:class:: Connection(socket: socket.socket, address: tuple[str, int], remainder: bytes = b'', separator: bytes = b';', on_close: typing.Callable[[typing.Self], None] = lambda _: None)
   :canonical: leads.comm.prototype.Connection

   Bases: :py:obj:`leads.comm.prototype.ConnectionBase`

   .. autodoc2-docstring:: leads.comm.prototype.Connection

   .. rubric:: Initialization

   .. autodoc2-docstring:: leads.comm.prototype.Connection.__init__

   .. py:method:: __str__() -> str
      :canonical: leads.comm.prototype.Connection.__str__

      .. autodoc2-docstring:: leads.comm.prototype.Connection.__str__

   .. py:method:: closed() -> bool
      :canonical: leads.comm.prototype.Connection.closed

      .. autodoc2-docstring:: leads.comm.prototype.Connection.closed

   .. py:method:: _require_open_socket(mandatory: bool = True) -> socket.socket
      :canonical: leads.comm.prototype.Connection._require_open_socket

      .. autodoc2-docstring:: leads.comm.prototype.Connection._require_open_socket

   .. py:method:: receive(chunk_size: int = 512) -> bytes | None
      :canonical: leads.comm.prototype.Connection.receive

      .. autodoc2-docstring:: leads.comm.prototype.Connection.receive

   .. py:method:: send(msg: bytes) -> None
      :canonical: leads.comm.prototype.Connection.send

      .. autodoc2-docstring:: leads.comm.prototype.Connection.send

   .. py:method:: close() -> None
      :canonical: leads.comm.prototype.Connection.close

      .. autodoc2-docstring:: leads.comm.prototype.Connection.close

.. py:class:: Callback(chain: typing.Self | None = None)
   :canonical: leads.comm.prototype.Callback

   Bases: :py:obj:`leads.callback.CallbackChain`

   .. autodoc2-docstring:: leads.comm.prototype.Callback

   .. rubric:: Initialization

   .. autodoc2-docstring:: leads.comm.prototype.Callback.__init__

   .. py:method:: on_initialize(service: leads.comm.prototype.Service) -> None
      :canonical: leads.comm.prototype.Callback.on_initialize

      .. autodoc2-docstring:: leads.comm.prototype.Callback.on_initialize

   .. py:method:: on_fail(service: leads.comm.prototype.Service, error: Exception) -> None
      :canonical: leads.comm.prototype.Callback.on_fail

      .. autodoc2-docstring:: leads.comm.prototype.Callback.on_fail

   .. py:method:: on_connect(service: leads.comm.prototype.Service, connection: leads.comm.prototype.ConnectionBase) -> None
      :canonical: leads.comm.prototype.Callback.on_connect

      .. autodoc2-docstring:: leads.comm.prototype.Callback.on_connect

   .. py:method:: on_receive(service: leads.comm.prototype.Service, msg: bytes) -> None
      :canonical: leads.comm.prototype.Callback.on_receive

      .. autodoc2-docstring:: leads.comm.prototype.Callback.on_receive

   .. py:method:: on_disconnect(service: leads.comm.prototype.Service, connection: leads.comm.prototype.ConnectionBase) -> None
      :canonical: leads.comm.prototype.Callback.on_disconnect

      .. autodoc2-docstring:: leads.comm.prototype.Callback.on_disconnect

.. py:class:: Entity(port: int, callback: leads.comm.prototype.Callback)
   :canonical: leads.comm.prototype.Entity

   Bases: :py:obj:`leads.comm.prototype.Service`

   .. autodoc2-docstring:: leads.comm.prototype.Entity

   .. rubric:: Initialization

   .. autodoc2-docstring:: leads.comm.prototype.Entity.__init__

   .. py:method:: set_callback(callback: leads.comm.prototype.Callback) -> None
      :canonical: leads.comm.prototype.Entity.set_callback

      .. autodoc2-docstring:: leads.comm.prototype.Entity.set_callback

   .. py:method:: _stage(connection: leads.comm.prototype.ConnectionBase) -> None
      :canonical: leads.comm.prototype.Entity._stage

      .. autodoc2-docstring:: leads.comm.prototype.Entity._stage

   .. py:method:: _run(*args, **kwargs) -> None
      :canonical: leads.comm.prototype.Entity._run

      .. autodoc2-docstring:: leads.comm.prototype.Entity._run