The queue Module: Part 2

#!/usr/bin/env python3
# mypy: standard
"""
./demo_14_queue_producer_consumer.py

Producer–consumer demonstration using ``queue.Queue`` with a fixed
message count. ``maxsize`` limits the number of queued items *in flight*
(backpressure), but does not cap total production — that's controlled by ``count``.

What this shows
---------------
- Deterministic production of exactly ``count`` items.
- Backpressure via ``Queue(maxsize=...)``.
- Safe draining with ``get(timeout=...)`` to avoid ``empty()`` races.
- Non-deterministic interleaving of start/end logs due to scheduling.

Notes
-----
This example module serves educational purposes solely.

See Also
--------
demo_13_production_consumer_pipeline.py
    Manual two-lock pipeline implementing a single-slot baton.
"""

import builtins
import concurrent.futures
import queue
import random
import threading
import time
from typing import Callable, List, Optional

# Force unbuffered console output by enabling ``flush=True``.
# By default, ``print()`` buffers output, which may delay display.
# This global alias ensures immediate flushing across the script.
print: Callable[..., None] = lambda *args, **kwargs: builtins.print(
    *args, **{**kwargs, "flush": True}
)


class Pipeline(queue.Queue):
    """
    Queue-backed single-stage pipeline with logging helpers.

    Parameters
    ----------
    maxsize : int, optional
        Maximum number of items allowed in the queue at once. Acts as
        backpressure; it does **not** cap total production.
    """

    def __init__(self, maxsize: int = 20) -> None:
        super().__init__(maxsize=maxsize)
        self.produced: List[int] = []
        self.consumed: List[int] = []

    def set_message(self, message: int) -> None:
        """
        Enqueue a message; blocks if the queue is full.

        Parameters
        ----------
        message : int
            The value to enqueue.
        """
        print(f"Producing message: {message}")
        self.produced.append(message)
        self.put(message)  # blocks when qsize == maxsize

    def get_message(self, timeout: Optional[float] = None) -> int:
        """
        Dequeue a message; blocks until an item is available.

        Parameters
        ----------
        timeout : float, optional
            Seconds to wait before raising ``queue.Empty``. ``None``
            waits indefinitely.

        Returns
        -------
        int
            The dequeued value.

        Raises
        ------
        queue.Empty
            If the queue remains empty until the timeout elapses.
        """
        msg: int = self.get(timeout=timeout)
        print(f"Consuming message: {msg}")
        self.consumed.append(msg)
        return msg


def producer(pipeline: Pipeline, done: threading.Event, count: int) -> None:
    """
    Produce exactly ``count`` random integers, then signal completion.

    Parameters
    ----------
    pipeline : Pipeline
        Target queue-backed pipeline.
    done : threading.Event
        Event set by the producer when all items are enqueued.
    count : int
        Total number of data items to produce.
    """
    for _ in range(count):
        pipeline.set_message(random.randint(1, 100))
    done.set()  # signal consumer no more items will be produced


def consumer(pipeline: Pipeline, done: threading.Event) -> None:
    """
    Drain the queue until the producer is done *and* the queue is empty.

    Uses ``get(timeout=...)`` to avoid a race where the queue becomes
    empty after checking ``empty()`` but before ``get()``.

    Parameters
    ----------
    pipeline : Pipeline
        Source pipeline to consume from.
    done : threading.Event
        Event indicating the producer has finished enqueuing.
    """
    while True:
        try:
            pipeline.get_message(timeout=0.1)

            # Simulate variable-processing latency.
            time.sleep(random.random() * 0.2)
        except queue.Empty:
            if done.is_set():
                break  # producer finished and nothing left to consume


def main() -> None:
    """
    Main entry point.

    Creates a pipeline with backpressure and runs a fixed-count producer
    alongside a draining consumer. Prints counts and items at the end.
    """
    count: int = 10
    pipeline: Pipeline = Pipeline(maxsize=20)
    done: threading.Event = threading.Event()

    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
        executor.submit(producer, pipeline, done, count)
        executor.submit(consumer, pipeline, done)

    print(f"producer count: {len(pipeline.produced)}")
    print(f"producer items: {pipeline.produced}")
    print(f"consumer count: {len(pipeline.consumed)}")
    print(f"consumer items: {pipeline.consumed}")


if __name__ == "__main__":
    main()