# Module `src.communicatorutils.uploadblocks.mixmate.pool`

Onionr - Private P2P Communication.

Upload pool

Expand source code

    """Onionr - Private P2P Communication.

    Upload pool
    """
    from typing import List

    import onionrutils
    import onionrtypes
    from onionrcrypto import cryptoutils
    """
        This program is free software: you can redistribute it and/or modify
        it under the terms of the GNU General Public License as published by
        the Free Software Foundation, either version 3 of the License, or
        (at your option) any later version.

        This program is distributed in the hope that it will be useful,
        but WITHOUT ANY WARRANTY; without even the implied warranty of
        MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
        GNU General Public License for more details.

        You should have received a copy of the GNU General Public License
        along with this program.  If not, see <https://www.gnu.org/licenses/>.
    """

    class PoolFullException(Exception):
        """For when the UploadPool is full.

        Raise when a new hash is attempted to be added
        """

    class PoolNotReady(Exception):
        """Raise when UploadPool pool access is attempted without it being full."""

    class AlreadyInPool(Exception):
        """Raise when a hash already in pool is attempted to be added again."""

    class UploadPool:
        """Upload pool for mixing blocks together and delaying uploads."""

        def __init__(self, pool_size: int):
            """Create a new pool with a specified max size.

            Uses private var and getter to avoid direct adding
            """
            self._pool: List[onionrtypes.BlockHash] = []
            self._pool_size = pool_size
            self.birthday = onionrutils.epoch.get_epoch()

        def add_to_pool(self, item: List[onionrtypes.BlockHash]):
            """Add a new hash to the pool. Raise PoolFullException if full."""
            if len(self._pool) >= self._pool_size:
                raise PoolFullException
            if not onionrutils.stringvalidators.validate_hash(item):
                raise ValueError
            self._pool.append(item)

        def get_pool(self) -> List[onionrtypes.BlockHash]:
            """Get the hash pool in secure random order."""
            if len(self._pool) != self._pool_size:
                raise PoolNotReady
            final_pool: List[onionrtypes.BlockHash] = cryptoutils.random_shuffle(
                list(self._pool))

            self._pool.clear()
            self.birthday = onionrutils.epoch.get_epoch()
            return final_pool

## Classes

`class AlreadyInPool (...)`

  Raise when a hash already in pool is attempted to be added again.

  Expand source code

      class AlreadyInPool(Exception):
          """Raise when a hash already in pool is attempted to be added again."""

  ### Ancestors

  - builtins.Exception
  - builtins.BaseException
`class PoolFullException (...)`

  For when the UploadPool is full.

  Raise when a new hash is attempted to be added

  Expand source code

      class PoolFullException(Exception):
          """For when the UploadPool is full.

          Raise when a new hash is attempted to be added
          """

  ### Ancestors

  - builtins.Exception
  - builtins.BaseException
`class PoolNotReady (...)`

  Raise when UploadPool pool access is attempted without it being full.

  Expand source code

      class PoolNotReady(Exception):
          """Raise when UploadPool pool access is attempted without it being full."""

  ### Ancestors

  - builtins.Exception
  - builtins.BaseException
`class UploadPool (pool_size)`

  Upload pool for mixing blocks together and delaying uploads.

  Create a new pool with a specified max size.

  Uses private var and getter to avoid direct adding

  Expand source code

      class UploadPool:
          """Upload pool for mixing blocks together and delaying uploads."""

          def __init__(self, pool_size: int):
              """Create a new pool with a specified max size.

              Uses private var and getter to avoid direct adding
              """
              self._pool: List[onionrtypes.BlockHash] = []
              self._pool_size = pool_size
              self.birthday = onionrutils.epoch.get_epoch()

          def add_to_pool(self, item: List[onionrtypes.BlockHash]):
              """Add a new hash to the pool. Raise PoolFullException if full."""
              if len(self._pool) >= self._pool_size:
                  raise PoolFullException
              if not onionrutils.stringvalidators.validate_hash(item):
                  raise ValueError
              self._pool.append(item)

          def get_pool(self) -> List[onionrtypes.BlockHash]:
              """Get the hash pool in secure random order."""
              if len(self._pool) != self._pool_size:
                  raise PoolNotReady
              final_pool: List[onionrtypes.BlockHash] = cryptoutils.random_shuffle(
                  list(self._pool))

              self._pool.clear()
              self.birthday = onionrutils.epoch.get_epoch()
              return final_pool

  ### Methods

  `def add_to_pool(self, item)`

    Add a new hash to the pool. Raise PoolFullException if full.

    Expand source code

        def add_to_pool(self, item: List[onionrtypes.BlockHash]):
            """Add a new hash to the pool. Raise PoolFullException if full."""
            if len(self._pool) >= self._pool_size:
                raise PoolFullException
            if not onionrutils.stringvalidators.validate_hash(item):
                raise ValueError
            self._pool.append(item)

  `def get_pool(self)`

    Get the hash pool in secure random order.

    Expand source code

        def get_pool(self) -> List[onionrtypes.BlockHash]:
            """Get the hash pool in secure random order."""
            if len(self._pool) != self._pool_size:
                raise PoolNotReady
            final_pool: List[onionrtypes.BlockHash] = cryptoutils.random_shuffle(
                list(self._pool))

            self._pool.clear()
            self.birthday = onionrutils.epoch.get_epoch()
            return final_pool