# Module `infinibatch.closablequeue`

Expand source code

    from collections import deque
    from threading import Condition, Lock, Thread

    class ClosedException(Exception):
        pass

    class ClosableQueue:
        """
        A thread-safe queue that can be closed

        As long as the the queue is not closed, it behaves just like a thread-safe queue with a capacity limit:
            - put blocks until the item can be added
            - get blocks until there is an item to be returned

        Once the queue is closed, no more items can be added but existing items can be removed:
            - put always raises a ClosedException
            - get returns an item if the queue is not empty and otherwise raises a ClosedException
        """
        def __init__(self, maxsize: int=1000):
            self._maxsize = maxsize
            self._queue = deque()
            self._mutex = Lock()
            self._not_empty = Condition(self._mutex)
            self._not_full = Condition(self._mutex)
            self._closed = False

        def put(self, item):
            with self._not_full:
                if self._closed:
                    raise ClosedException('This queue has been closed, no more items can be added.')
                while len(self._queue) >= self._maxsize:
                    self._not_full.wait()
                    if self._closed:
                        raise ClosedException('This queue has been closed, no more items can be added.')
                self._queue.append(item)
                self._not_empty.notify()

        def get(self):
            with self._not_empty:
                if self._closed and len(self._queue) == 0:
                    raise ClosedException('This queue has been closed and is empty, no more items can be retrieved.')
                while len(self._queue) == 0:
                    self._not_empty.wait()
                    if self._closed and len(self._queue) == 0:
                        raise ClosedException('This queue has been closed and is empty, no more items can be retrieved.')
                item = self._queue.popleft()
                self._not_full.notify()
            return item

        def close(self):
            with self._mutex:
                self._closed = True
                self._not_empty.notify_all()
                self._not_full.notify_all()

## Classes

`class ClosedException (...)`

  Common base class for all non-exit exceptions.

  Expand source code

      class ClosedException(Exception):
          pass

  ### Ancestors

  - builtins.Exception
  - builtins.BaseException
`class ClosableQueue (maxsize: int = 1000)`

  A thread-safe queue that can be closed

  As long as the the queue is not closed, it behaves just like a thread-safe queue with a capacity limit: - put blocks until the item can be added - get blocks until there is an item to be returned

  Once the queue is closed, no more items can be added but existing items can be removed: - put always raises a ClosedException - get returns an item if the queue is not empty and otherwise raises a ClosedException

  Expand source code

      class ClosableQueue:
          """
          A thread-safe queue that can be closed

          As long as the the queue is not closed, it behaves just like a thread-safe queue with a capacity limit:
              - put blocks until the item can be added
              - get blocks until there is an item to be returned

          Once the queue is closed, no more items can be added but existing items can be removed:
              - put always raises a ClosedException
              - get returns an item if the queue is not empty and otherwise raises a ClosedException
          """
          def __init__(self, maxsize: int=1000):
              self._maxsize = maxsize
              self._queue = deque()
              self._mutex = Lock()
              self._not_empty = Condition(self._mutex)
              self._not_full = Condition(self._mutex)
              self._closed = False

          def put(self, item):
              with self._not_full:
                  if self._closed:
                      raise ClosedException('This queue has been closed, no more items can be added.')
                  while len(self._queue) >= self._maxsize:
                      self._not_full.wait()
                      if self._closed:
                          raise ClosedException('This queue has been closed, no more items can be added.')
                  self._queue.append(item)
                  self._not_empty.notify()

          def get(self):
              with self._not_empty:
                  if self._closed and len(self._queue) == 0:
                      raise ClosedException('This queue has been closed and is empty, no more items can be retrieved.')
                  while len(self._queue) == 0:
                      self._not_empty.wait()
                      if self._closed and len(self._queue) == 0:
                          raise ClosedException('This queue has been closed and is empty, no more items can be retrieved.')
                  item = self._queue.popleft()
                  self._not_full.notify()
              return item

          def close(self):
              with self._mutex:
                  self._closed = True
                  self._not_empty.notify_all()
                  self._not_full.notify_all()

  ### Methods

  `def put(self, item)`
    Expand source code

        def put(self, item):
            with self._not_full:
                if self._closed:
                    raise ClosedException('This queue has been closed, no more items can be added.')
                while len(self._queue) >= self._maxsize:
                    self._not_full.wait()
                    if self._closed:
                        raise ClosedException('This queue has been closed, no more items can be added.')
                self._queue.append(item)
                self._not_empty.notify()

  `def get(self)`
    Expand source code

        def get(self):
            with self._not_empty:
                if self._closed and len(self._queue) == 0:
                    raise ClosedException('This queue has been closed and is empty, no more items can be retrieved.')
                while len(self._queue) == 0:
                    self._not_empty.wait()
                    if self._closed and len(self._queue) == 0:
                        raise ClosedException('This queue has been closed and is empty, no more items can be retrieved.')
                item = self._queue.popleft()
                self._not_full.notify()
            return item

  `def close(self)`
    Expand source code

        def close(self):
            with self._mutex:
                self._closed = True
                self._not_empty.notify_all()
                self._not_full.notify_all()