Исходный код EJIO.utils.struct.queue

## /utils/struct/queue.py

from threading import Lock

__all__ = [
    'Queue',
]

from typing import Any, Generator


class Node[T]:
    """
    Узел связного списка для хранения значения и указателя на следующий элемент.
    """
    __slots__ = ('value', 'next')

    def __init__(self, value: T, next: 'Node[T] | None' = None) -> None:
        """
        Создает узел связного списка.

        :param value: Значение записываемое в узел.
        :param next: Ссылка на следующий узел.
        """
        self.value: T = value
        self.next: 'Node[T] | None' = next




class Queue[T]:
    """
    Потокобезопасная очередь на односвязном списке.
    Класс поддерживает основные операции push, pop, peek, size, empty.
    """
    __slots__ = ('_head', '_tail', '_size', '_lock')

    def __init__(self) -> None:
        """Создает объект очереди."""
        self._head: Node[T] | None = None
        self._tail: Node[T] | None = None
        self._lock: Lock = Lock()
        self._size: int = 0



    def push(self, item: T) -> None:
        """
        Добавляет элемент в конец очереди.

        :param item: Объект для добавления.
        """
        new_node = Node(item)
        with self._lock:
            if self._tail is None:
                self._head = new_node
                self._tail = new_node
            else:
                self._tail.next = new_node
                self._tail = new_node
            self._size += 1




    def pop(self) -> T:
        """
        Удаляет и возвращает первый элемент очереди.

        :return: Объект из начала очереди.
        :raises IndexError: Если очередь пуста.
        """
        with self._lock:
            if self._head is None:
                raise IndexError("Pop from empty queue")
            node = self._head
            self._head = node.next

            if self._head is None:
                self._tail = None

            self._size -= 1
            return node.value




    def peek(self) -> T:
        """
        Возвращает первый элемент очереди без удаления.

        :return: Объект из начала очереди.
        :raises IndexError: Если очередь пуста.
        """
        with self._lock:
            if self._head is None:
                raise IndexError("Peek from empty queue")
            return self._head.value


    @property
    def empty(self) -> bool:
        """
        Проверяет пуста ли очередь.

        :return: True если очередь пуста, иначе False.
        """
        with self._lock:
            return self._head is None

    @property
    def size(self) -> int:
        """
        Возвращает текущий размер очереди.

        :return: Размер очереди.
        """
        with self._lock:
            return self._size

    def __iter__(self) -> Generator[T, Any, None]:
        """
        Возвращает итератор, который извлекает элементы из очереди
        до тех пор, пока она не опустеет.
        Потокобезопасно на уровне отдельных операций.

        :yields: Элементы очереди в порядке с начала до конца.
        """
        while True:
            try:
                yield self.pop()
            except IndexError:
                return

    def __str__(self) -> str:
        """
        Возвращает строковое представление всей очереди в виде списка элементов
        от начала к концу.

        :return: Представление очереди в виде строки.
        """
        with self._lock:
            elements = []
            current = self._head
            while current:
                elements.append(repr(current.value))
                current = current.next
            return f"[{', '.join(elements)}]"

    def __repr__(self) -> str:
        """
        Возвращает строковое представление всей очереди в виде списка элементов
        от начала (head) к концу (tail).

        :return: Представление очереди в виде строки.
        """
        return f"Queue({str(self)})"

    def __del__(self) -> None:
        """Очищает память при удалении объекта, чтобы избежать рекурсии."""
        with self._lock:
            current = self._head
            while current:
                next_node = current.next
                current.next = None
                current = next_node
            self._head = None
            self._tail = None
            self._size = 0