### SyncQueueProxy get() examples

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/sync-queue-proxy.md

Illustrates retrieving items from the synchronous queue with various blocking and timeout options. `get()` is used for blocking retrieval or when a timeout is required. Set `block=False` or provide a `timeout` to prevent indefinite waiting.

```python
item = sync_q.get()  # Block indefinitely if empty
item = sync_q.get(block=False)  # Raise immediately if empty
item = sync_q.get(timeout=5.0)  # Wait up to 5 seconds
```

--------------------------------

### Example Usage of Queue Shutdown

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/queue.md

Demonstrates how to initialize a Janus Queue and shut it down immediately.

```python
import janus

queue = janus.Queue()
# ... use the queue ...
queue.shutdown(immediate=True)
```

--------------------------------

### SyncQueueProxy get_nowait() example

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/sync-queue-proxy.md

Demonstrates retrieving an item from the queue without blocking. This is equivalent to `get(block=False)`. Use `get_nowait()` when you need to immediately check if an item is available and avoid blocking.

```python
item = sync_q.get_nowait()
```

--------------------------------

### Example Usage of Queue Close

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/queue.md

Demonstrates the direct call to close the queue immediately.

```python
queue.close()
```

--------------------------------

### Get Janus Library Version

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/configuration.md

This snippet shows how to import the janus library and print its current version. Ensure the janus library is installed.

```python
import janus

print(janus.__version__)  # "2.0.0"
```

--------------------------------

### Mixed Sync-Async LifoQueue Example

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/lifo-queue.md

Illustrates a producer-consumer pattern using LifoQueue, where a synchronous producer thread adds items and an asynchronous consumer coroutine retrieves them in LIFO order. The example utilizes `run_in_executor` to run the synchronous producer in a separate thread.

```python
import asyncio
import janus
import threading

def producer_thread(sync_q):
    """Producer thread that pushes items onto the stack"""
    for i in range(1, 6):
        sync_q.put(f"item_{i}")
    sync_q.join()

async def consumer_coro(async_q):
    """Consumer coroutine that pops items in LIFO order"""
    for _ in range(5):
        item = await async_q.get()
        print(f"Processing {item}")
        async_q.task_done()

async def main():
    queue = janus.LifoQueue()
    
    # Start producer thread
    loop = asyncio.get_running_loop()
    producer = loop.run_in_executor(None, producer_thread, queue.sync_q)
    
    # Run consumer
    await consumer_coro(queue.async_q)
    
    # Wait for producer to finish
    await producer
    
    # Clean up
    await queue.aclose()

asyncio.run(main())
```

--------------------------------

### Basic Janus Queue Operations (Sync and Async)

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/INDEX.md

Demonstrates how to put and get items from a Janus Queue using both its synchronous and asynchronous interfaces. Ensure the `janus` library is imported.

```python
import janus

queue = janus.Queue()

# Sync side
queue.sync_q.put(item)
item = queue.sync_q.get()

# Async side
await queue.async_q.put(item)
item = await queue.async_q.get()
```

--------------------------------

### Synchronous Task Tracking and Joining

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/README.md

Demonstrates how to track tasks in a synchronous queue using `put()`, `get()`, and `task_done()`. `sync_q.join()` blocks until all tasks are completed.

```python
# Put items
for i in range(10):
    sync_q.put(i)

# Consume items
for _ in range(10):
    item = sync_q.get()
    process(item) # Assuming process is defined elsewhere
    sync_q.task_done()

# Wait until all items processed
sync_q.join()  # Blocks until all task_done() calls match put() calls
```

--------------------------------

### Example Usage of Wait Closed

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/queue.md

Shows how to close the queue and then asynchronously wait for all operations to finish.

```python
queue.close()
await queue.wait_closed()
```

--------------------------------

### Basic LifoQueue Usage Example

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/lifo-queue.md

Demonstrates putting items into a LifoQueue using its synchronous interface and retrieving them in LIFO order using its asynchronous interface. Ensure to call task_done() for each retrieved item and close the queue when finished.

```python
import janus

async def main():
    queue = janus.LifoQueue()
    
    # Put items
    sync_q = queue.sync_q
    sync_q.put("first")
    sync_q.put("second")
    sync_q.put("third")
    
    # Get items in LIFO order
    async_q = queue.async_q
    item1 = await async_q.get()  # Returns "third" (most recent)
    item2 = await async_q.get()  # Returns "second"
    item3 = await async_q.get()  # Returns "first" (oldest)
    
    async_q.task_done()
    async_q.task_done()
    async_q.task_done()
    
    await queue.aclose()

import asyncio
asyncio.run(main())
```

--------------------------------

### LIFO Queue (Stack) Example

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/usage-patterns.md

Demonstrates using Janus LifoQueue for stack-like behavior, where items are pushed synchronously and popped asynchronously in LIFO order. Ensure the queue is closed after use.

```python
import asyncio
import janus

def sync_push_items(sync_q):
    """Push items onto stack"""
    items = ["item-1", "item-2", "item-3", "item-4", "item-5"]
    for item in items:
        sync_q.put(item)
        print(f"Pushed: {item}")
    sync_q.join()

async def async_pop_items(async_q):
    """Pop items from stack (LIFO)"""
    for _ in range(5):
        item = await async_q.get()
        print(f"Popped: {item}")  # Prints item-5, item-4, etc.
        async_q.task_done()

async def main():
    queue = janus.LifoQueue()
    loop = asyncio.get_running_loop()
    
    producer = loop.run_in_executor(None, sync_push_items, queue.sync_q)
    await async_pop_items(queue.async_q)
    await producer
    await queue.aclose()

asyncio.run(main())
```

--------------------------------

### SyncQueueProxy put() examples

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/sync-queue-proxy.md

Demonstrates how to put items into the synchronous queue with different blocking and timeout behaviors. Use `put()` for blocking operations or when a timeout is needed. Ensure `block=False` or a `timeout` is specified to avoid indefinite blocking.

```python
import janus

sync_q = janus.Queue().sync_q
sync_q.put(42)  # Block indefinitely if full
sync_q.put(42, block=False)  # Raise immediately if full
sync_q.put(42, timeout=2.0)  # Wait up to 2 seconds
```

--------------------------------

### Type Hinting with Janus Queues

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/public-api.md

Provides an example of using type hints with Janus queue classes. It shows how to specify item types for generic queues and access synchronous/asynchronous interfaces.

```python
from janus import Queue, SyncQueue, AsyncQueue
from typing import TypeVar

T = TypeVar('T')

def process_queue(q: Queue[int]) -> None:
    sync_q: SyncQueue[int] = q.sync_q
    # ...
```

--------------------------------

### SyncQueueProxy task_done() example

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/sync-queue-proxy.md

Illustrates the usage of `task_done()` to signal the completion of processing for an item retrieved with `get()`. This is crucial for tracking unfinished tasks and enabling the `join()` method to resume.

```python
item = sync_q.get()
try:
    process(item)
finally:
    sync_q.task_done()
```

--------------------------------

### Basic PriorityQueue Usage

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/priority-queue.md

Demonstrates putting and getting items with different priorities using the synchronous and asynchronous interfaces of a PriorityQueue. Ensure the queue is closed after use.

```python
import janus

async def main():
    queue = janus.PriorityQueue()
    
    # Put items with priorities
    sync_q = queue.sync_q
    sync_q.put((1, "high priority"))
    sync_q.put((3, "low priority"))
    sync_q.put((2, "medium priority"))
    
    # Get items in priority order
    async_q = queue.async_q
    item1 = await async_q.get()  # Returns (1, "high priority")
    item2 = await async_q.get()  # Returns (2, "medium priority")
    item3 = await async_q.get()  # Returns (3, "low priority")
    
    async_q.task_done()
    async_q.task_done()
    async_q.task_done()
    
    await queue.aclose()

import asyncio
asyncio.run(main())
```

--------------------------------

### SyncQueueProxy put_nowait() example

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/sync-queue-proxy.md

Shows how to add an item to the queue without blocking. This is equivalent to `put(item, block=False)`. Use `put_nowait()` when you need to immediately check if a slot is available and avoid blocking.

```python
sync_q.put_nowait(42)
```

--------------------------------

### Example Usage of Async Close

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/queue.md

Illustrates the combined asynchronous operation to close the queue and wait for completion.

```python
queue = janus.Queue()
# ... use the queue ...
await queue.aclose()
```

--------------------------------

### Asynchronous Task Tracking and Joining

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/README.md

Shows task tracking in an asynchronous queue using `await put()`, `await get()`, and `task_done()`. `await async_q.join()` waits for all tasks to be completed.

```python
# Put items
for i in range(10):
    await async_q.put(i)

# Consume items
for _ in range(10):
    item = await async_q.get()
    await process(item) # Assuming process is defined elsewhere
    async_q.task_done()

# Wait until all items processed
await async_q.join()
```

--------------------------------

### Get item from SyncQueue (blocking)

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/configuration.md

Use SyncQueue.get to retrieve an item. By default, it blocks indefinitely until an item is available. Can be configured to not block or to use a timeout.

```python
sync_q.get(block=True, timeout=None)
```

```python
item = sync_q.get()  # Block indefinitely
```

```python
item = sync_q.get(block=False)  # Raise immediately if empty
```

```python
item = sync_q.get(timeout=10.0)  # Block up to 10 seconds
```

--------------------------------

### Mixed Sync-Async PriorityQueue Example

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/priority-queue.md

Illustrates a producer-consumer pattern using a PriorityQueue where a synchronous producer thread adds items and an asynchronous consumer coroutine retrieves them in priority order. The producer thread uses sync_q.join() to signal completion, and the main coroutine awaits the producer's completion before closing the queue.

```python
import asyncio
import janus
import threading

def producer_thread(sync_q):
    """Producer thread that puts items with priorities"""
    items = [
        (3, "item3"),
        (1, "item1"),
        (2, "item2"),
    ]
    for item in items:
        sync_q.put(item)
    sync_q.join()

async def consumer_coro(async_q):
    """Consumer coroutine that gets items in priority order"""
    for _ in range(3):
        priority, item = await async_q.get()
        print(f"Processing {item} with priority {priority}")
        async_q.task_done()

async def main():
    queue = janus.PriorityQueue()
    
    # Start producer thread
    loop = asyncio.get_running_loop()
    producer = loop.run_in_executor(None, producer_thread, queue.sync_q)
    
    # Run consumer
    await consumer_coro(queue.async_q)
    
    # Wait for producer to finish
    await producer
    
    # Clean up
    await queue.aclose()

asyncio.run(main())
```

--------------------------------

### Sync Put to Async Get Data Flow

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/architecture.md

Illustrates the sequence of operations when an item is put from a synchronous context and retrieved from an asynchronous context. Shows lock acquisition, internal put logic, and notification mechanisms for both sync and async waiters.

```text
1. sync_q.put(item)
   ├─ Acquires _sync_mutex
   ├─ Calls _put_internal(item)
   │  └─ _unfinished_tasks += 1
   ├─ Calls _sync_not_empty.notify()
   │  └─ Wakes one waiting sync thread
   ├─ Checks if async waiters: if _async_not_empty_waiting > 0
   │  ├─ Calls _notify_async(_async_not_empty.notify)
   │  └─ Schedules async task to notify
   └─ Releases _sync_mutex

2. async_q.get() [waiting]
   ├─ Waits on _async_not_empty
   └─ Woken by notification task
   
3. await async_q.get() [resumes]
   ├─ Acquires _async_mutex
   ├─ Calls _get()
   ├─ Checks if sync waiters: if _sync_not_full_waiting > 0
   │  └─ Calls _sync_not_full.notify()
   │     └─ Wakes one waiting sync thread
   └─ Returns item
```

--------------------------------

### Priority Queue Usage with Janus

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/README.md

Demonstrates using Janus PriorityQueue where lower numerical values indicate higher priority. Items are retrieved in order of their priority. Note that this example uses the synchronous interface for putting items.

```python
import janus

queue = janus.PriorityQueue()
sync_q = queue.sync_q

# Higher priority = lower value
sync_q.put((1, "urgent"))
sync_q.put((3, "low"))
sync_q.put((2, "medium"))

# Retrieved in priority order
async_q = queue.async_q
urgent = await async_q.get()  # (1, "urgent")
medium = await async_q.get()  # (2, "medium")
low = await async_q.get()     # (3, "low")
```

--------------------------------

### Graceful and Immediate Janus Queue Shutdown

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/INDEX.md

Provides examples for shutting down a Janus Queue. `shutdown()` allows existing items to be processed, while `close()` discards them immediately. `wait_closed()` should be awaited after either.

```python
# Graceful (allow items to drain)
queue.shutdown()
await queue.wait_closed()

# Immediate (discard items)
queue.close()
await queue.wait_closed()

# Combined
await queue.aclose()
```

--------------------------------

### Async Put to Sync Get Data Flow

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/architecture.md

Illustrates the sequence of operations when an item is put from an asynchronous context and retrieved from a synchronous context. Highlights nested lock acquisition and notification order.

```text
1. await async_q.put(item)
   ├─ Acquires _async_not_full lock
   ├─ Acquires _sync_mutex (nested!)
   ├─ Calls _put_internal(item)
   ├─ Checks _sync_not_empty_waiting
   │  └─ Calls _sync_not_empty.notify()
   ├─ Checks _async_not_empty_waiting
   │  └─ Calls _async_not_empty.notify()
   └─ Releases locks

2. sync_q.get() [blocking]
   ├─ Waits on _sync_not_empty
   └─ Woken by put() notification
   
3. sync_q.get() [resumes]
   ├─ Acquires _sync_mutex
   ├─ Calls _get()
   ├─ Checks _async_not_full_waiting
   │  └─ Calls _notify_async(_async_not_full.notify)
   └─ Returns item
```

--------------------------------

### Example of Janus Queue Usage

Source: https://github.com/aio-libs/janus/blob/master/README.rst

Demonstrates how to use Janus Queue for communication between a synchronous thread and an asynchronous coroutine. Ensure the queue is properly closed with aclose() after use.

```python
import asyncio
import janus


def threaded(sync_q: janus.SyncQueue[int]) -> None:
    for i in range(100):
        sync_q.put(i)
    sync_q.join()


async def async_coro(async_q: janus.AsyncQueue[int]) -> None:
    for i in range(100):
        val = await async_q.get()
        assert val == i
        async_q.task_done()


async def main() -> None:
    queue: janus.Queue[int] = janus.Queue()
    loop = asyncio.get_running_loop()
    fut = loop.run_in_executor(None, threaded, queue.sync_q)
    await async_coro(queue.async_q)
    await fut
    await queue.aclose()


asyncio.run(main())
```

--------------------------------

### Shut down the queue

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/sync-queue-proxy.md

Call `shutdown()` to close the queue. Setting `immediate=True` will clear all items, while `immediate=False` (default) will cause `get()` to raise an exception when the queue is empty.

```python
def shutdown(self, immediate: bool = False) -> None
```

```python
sync_q.shutdown(immediate=True)
```

--------------------------------

### Graceful Queue Shutdown with Remaining Items

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/usage-patterns.md

Use `queue.shutdown(immediate=False)` to initiate a graceful shutdown. This allows any items currently in the queue to be consumed before the queue is fully closed. Subsequent `get` operations will raise `asyncio.QueueShutDown` once the queue is empty.

```python
import asyncio
import janus

async def main():
    queue = janus.Queue()
    async_q = queue.async_q
    sync_q = queue.sync_q
    
    # Put items
    sync_q.put("item-1")
    sync_q.put("item-2")
    
    # Graceful shutdown - allows remaining items to be consumed
    queue.shutdown(immediate=False)
    
    # Can still get remaining items
    try:
        item = await async_q.get()
        print(f"Got before shutdown: {item}")
    except asyncio.QueueShutDown:
        pass
    
    # After queue is empty, gets raise
    try:
        item = await async_q.get()
    except asyncio.QueueShutDown:
        print("Queue is shutdown")
    
    await queue.wait_closed()

asyncio.run(main())
```

--------------------------------

### Non-Blocking Queue Operations (Sync)

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/INDEX.md

Demonstrates non-blocking put and get operations for the synchronous side of a Janus Queue. Use `put_nowait` and `get_nowait` to avoid blocking, catching `Full` or `Empty` exceptions.

```python
from queue import Full, Empty

try:
    queue.sync_q.put_nowait(item)
except Full:
    print("Queue full")

try:
    item = queue.sync_q.get_nowait()
except Empty:
    print("Queue empty")
```

--------------------------------

### Creating a Queue

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/public-api.md

Demonstrates how to instantiate different types of Janus queues: a standard FIFO queue, a bounded FIFO queue, a priority queue, and a LIFO queue (stack).

```APIDOC
## Creating a Queue

### Description
Instantiate various types of Janus queues.

### Usage
```python
from janus import Queue, PriorityQueue, LifoQueue

# FIFO queue
fifo = Queue()
fifo_bounded = Queue(maxsize=100)

# Priority queue
pq = PriorityQueue()

# Stack
stack = LifoQueue()
```
```

--------------------------------

### Get current size of async queue

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/async-queue-proxy.md

Use `qsize` to get the number of items currently in the queue.

```python
size = async_q.qsize()
```

--------------------------------

### Get item from async queue

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/async-queue-proxy.md

Use the `get` coroutine to remove and return an item from the queue. It waits if the queue is empty.

```python
item = await async_q.get()
```

--------------------------------

### Getting an item from the Sync Queue Proxy

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/sync-queue-proxy.md

Use the `get` method to retrieve an item from the synchronous side of the queue. This operation will block if the queue is empty.

```python
item = proxy.get()

```

--------------------------------

### Get approximate queue size

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/sync-queue-proxy.md

Use `qsize()` to get an approximate number of items currently in the queue. Note that this method is not reliable in multithreaded environments due to potential race conditions.

```python
def qsize(self) -> int
```

```python
size = sync_q.qsize()
```

--------------------------------

### Create Janus Queues

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/public-api.md

Demonstrates the creation of different types of Janus queues: FIFO, bounded FIFO, priority queue, and LIFO queue (stack).

```python
from janus import Queue, PriorityQueue, LifoQueue

# FIFO queue
fifo = Queue()
fifo_bounded = Queue(maxsize=100)

# Priority queue
pq = PriorityQueue()

# Stack
stack = LifoQueue()
```

--------------------------------

### SyncQueueEmpty Exception

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/public-api.md

Raised when a synchronous get() operation is attempted on an empty queue.

```APIDOC
## SyncQueueEmpty Exception

### Description
Raised when synchronous get() is called on an empty queue.

### Import
`from janus import SyncQueueEmpty`
```

--------------------------------

### get_nowait

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/sync-queue-proxy.md

Removes and returns an item from the queue without blocking. This is a non-blocking equivalent of the `get` method.

```APIDOC
## get_nowait

```python
def get_nowait(self) -> T
```

Remove and return an item from the queue without blocking. Equivalent to `get(block=False)`.

### Returns:

T — The next item from the queue.

### Raises:

| Exception | Condition |
|-----------|-----------|
| SyncQueueEmpty | If the queue is empty |
| SyncQueueShutDown | If the queue is shutdown and empty |

### Example:

```python
item = sync_q.get_nowait()
```
```

--------------------------------

### Queue Initialization

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/queue.md

Creates a mixed sync-async queue with an optional maximum size. If maxsize is less than or equal to 0, the queue is unbounded.

```APIDOC
## Queue

```python
class Queue(Generic[T]):
    def __init__(self, maxsize: int = 0) -> None
```

### Parameters

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| maxsize | int | 0 | Maximum number of items in the queue. If <= 0, the queue is unbounded. |
```

--------------------------------

### AsyncQueueProxy Initialization

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/async-queue-proxy.md

Instances of _AsyncQueueProxy are created by accessing the `async_q` property of a `Queue`. This wrapper provides the asynchronous queue interface.

```python
class _AsyncQueueProxy(AsyncQueue[T]):
    def __init__(self, parent: Queue[T]) -> None:
```

--------------------------------

### SyncQueueEmpty Exception

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/public-api.md

Re-exported from the standard `queue.Empty`. Raised when a synchronous `get()` operation is attempted on an empty queue.

```python
from queue import Empty as SyncQueueEmpty
```

--------------------------------

### get

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/async-queue-proxy.md

Coroutine to remove and return an item from the queue. It waits if the queue is empty until an item becomes available.

```APIDOC
## get

### Description
Coroutine to remove and return an item from the queue. Waits if the queue is empty until an item becomes available.

### Method
`async def get(self) -> T`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
item = await async_q.get()
```

### Response
#### Success Response (200)
- **T** - The next item from the queue.

#### Response Example
None

### Raises
- **AsyncQueueShutDown**: If the queue is shutdown and empty
```

--------------------------------

### Monitor Janus Queue Size

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/usage-patterns.md

This snippet shows how to create a Janus Queue, run a synchronous producer and an asynchronous consumer, and monitor the queue's size and percentage utilization in real-time. It's useful for visualizing queue load and managing backpressure.

```python
import asyncio
import janus
import threading
import time

def sync_producer(sync_q, count=20):
    """Produce items slowly"""
    for i in range(count):
        sync_q.put(f"item-{i}")
        time.sleep(0.05)

async def async_consumer(async_q, rate=0.1):
    """Consume items slowly"""
    try:
        while True:
            item = await async_q.get()
            print(f"Consumed: {item}")
            async_q.task_done()
            await asyncio.sleep(rate)
    except asyncio.QueueShutDown:
        pass

async def monitor_queue(queue):
    """Monitor queue size"""
    while not queue.closed:
        size = queue.async_q.qsize()
        max_size = queue.maxsize
        if max_size > 0:
            percent = (size / max_size) * 100
            print(f"Queue: {size}/{max_size} ({percent:.1f}%)")
        else:
            print(f"Queue: {size} items")
        await asyncio.sleep(0.2)

async def main():
    queue = janus.Queue(maxsize=10)
    loop = asyncio.get_running_loop()
    
    # Start producer in thread
    producer = loop.run_in_executor(
        None,
        sync_producer,
        queue.sync_q,
        20
    )
    
    # Start consumer
    consumer_task = asyncio.create_task(
        async_consumer(queue.async_q, rate=0.1)
    )
    
    # Start monitor
    monitor_task = asyncio.create_task(monitor_queue(queue))
    
    # Wait for producer
    await producer
    
    # Wait for consumer to drain queue
    await queue.async_q.join()
    
    # Shutdown
    queue.close()
    await queue.wait_closed()
    
    # Cancel tasks
    consumer_task.cancel()
    monitor_task.cancel()

asyncio.run(main())

```

--------------------------------

### Import All Janus Modules

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/public-api.md

Shows how to import all public components from the Janus library using a wildcard import. Use with caution to avoid namespace pollution.

```python
from janus import *
```

--------------------------------

### Creating a Sync Queue Proxy

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/sync-queue-proxy.md

Instantiate a SyncQueueProxy to manage a synchronous queue. This is useful for bridging asynchronous and synchronous code.

```python
from janus import SyncQueueProxy

proxy = SyncQueueProxy()

```

--------------------------------

### get

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/sync-queue-proxy.md

Removes and returns an item from the queue. This method can block if the queue is empty, with options to control blocking behavior and timeouts.

```APIDOC
## get

```python
def get(self, block: bool = True, timeout: OptFloat = None) -> T
```

Remove and return an item from the queue. Blocks if the queue is empty and `block=True`.

### Parameters:

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| block | bool | True | If False, raise Empty immediately if queue is empty. If True, block until an item is available or timeout expires. |
| timeout | float or None | None | Maximum time in seconds to block. Only used if block=True. If None, block indefinitely. Must be non-negative if specified. |

### Returns:

T — The next item from the queue.

### Raises:

| Exception | Condition |
|-----------|-----------|
| SyncQueueEmpty | If block=False and queue is empty, or block=True and timeout expires |
| SyncQueueShutDown | If the queue is shutdown and empty |
| ValueError | If timeout is negative |

### Example:

```python
item = sync_q.get()  # Block indefinitely if empty
item = sync_q.get(block=False)  # Raise immediately if empty
item = sync_q.get(timeout=5.0)  # Wait up to 5 seconds
```
```

--------------------------------

### Import Janus Classes with Aliases

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/public-api.md

Illustrates importing Janus classes and renaming them using aliases. This is useful for avoiding naming conflicts or for brevity.

```python
from janus import Queue as JanusQueue
from janus import (
    SyncQueueEmpty as SyncEmpty,
    AsyncQueueEmpty as AsyncEmpty,
)
```

--------------------------------

### Get Maximum Queue Size

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/queue.md

Returns the maximum number of items the queue can hold. Returns 0 for unbounded queues.

```python
@property
def maxsize(self) -> int
```

--------------------------------

### Mark task as done in async queue

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/async-queue-proxy.md

Call `task_done` after processing an item obtained with `get`. This is crucial for `join()` to work correctly.

```python
item = await async_q.get()
try:
    await process(item)
finally:
    async_q.task_done()
```

--------------------------------

### Create Janus LifoQueue (Stack)

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/README.md

Instantiate a LIFO queue, behaving like a stack where the last item added is the first retrieved.

```python
import janus

queue = janus.LifoQueue()
```

--------------------------------

### Implement Timeouts with asyncio.wait_for

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/usage-patterns.md

Use `asyncio.wait_for` to set a timeout for queue operations like `get()`, preventing indefinite blocking and handling `asyncio.TimeoutError`.

```python
try:
    item = await asyncio.wait_for(async_q.get(), timeout=5.0)
except asyncio.TimeoutError:
    print("Timeout")
```

--------------------------------

### Get item from async queue without blocking

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/async-queue-proxy.md

Use `get_nowait` to remove and return an item synchronously. Raises `AsyncQueueEmpty` if the queue is empty.

```python
item = async_q.get_nowait()
```

--------------------------------

### Catching SyncQueueEmpty Exception

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/errors.md

Demonstrates how to catch the SyncQueueEmpty exception when attempting to get an item from an empty synchronous Janus queue without blocking.

```python
from queue import Empty as SyncQueueEmpty
import janus

sync_q = queue.sync_q
try:
    item = sync_q.get(block=False)
except janus.SyncQueueEmpty:
    print("Queue is empty")
```

--------------------------------

### task_done

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/async-queue-proxy.md

Indicate that a formerly enqueued item has been processed. This is a synchronous method. For each `get()` call, a subsequent `task_done()` must be called to mark the item as processed.

```APIDOC
## task_done

### Description
Indicate that a formerly enqueued item has been processed. This is a synchronous method (not a coroutine). For each `get()` call, a subsequent `task_done()` must be called to mark the item as processed.
If a `join()` is blocking, it will resume when all items have been processed.

### Method
`def task_done(self) -> None`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
item = await async_q.get()
try:
    await process(item)
finally:
    async_q.task_done()
```

### Response
#### Success Response (200)
None

#### Response Example
None

### Raises
- **ValueError**: If called more times than items were put() in the queue
```

--------------------------------

### Create Janus Queue (FIFO)

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/README.md

Instantiate a standard FIFO queue with separate synchronous and asynchronous interfaces.

```python
import janus

queue = janus.Queue()
sync_q = queue.sync_q
async_q = queue.async_q
```

--------------------------------

### Asynchronous Queue Get with Timeout

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/INDEX.md

Demonstrates retrieving an item from the asynchronous side of a Janus Queue with a timeout using `asyncio.wait_for`. An `asyncio.TimeoutError` is raised if the timeout occurs.

```python
import asyncio

try:
    item = await asyncio.wait_for(queue.async_q.get(), timeout=5.0)
except asyncio.TimeoutError:
    print("Timeout")
```

--------------------------------

### Initialize Janus Queue

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/queue.md

Creates a new Janus Queue with an optional maximum size. A maxsize of 0 or less indicates an unbounded queue.

```python
class Queue(Generic[T]):
    def __init__(self, maxsize: int = 0) -> None
```

--------------------------------

### Synchronous Queue Operations with Timeout

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/README.md

Demonstrates synchronous queue operations, including blocking put and get with timeouts. Raises SyncQueueFull or SyncQueueEmpty if timeouts expire.

```python
# Example usage for put and get with timeout
# sync_q.put(item, timeout=1.0)
# sync_q.get(timeout=1.0)
```

--------------------------------

### Synchronous Queue Integration

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/INDEX.md

Demonstrates how to use a Janus sync_q with a function designed for Python's standard library `queue.Queue`. This highlights the compatibility of the synchronous interface.

```python
from queue import Queue as StdQueue

def worker(q):
    # Works with any SyncQueue-compatible object
    pass

sync_q = janus_queue.sync_q
worker(sync_q)  # Compatible
```

--------------------------------

### Janus Module Structure

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/README.md

Illustrates the hierarchical organization of the Janus library, detailing the placement of Queue classes and related components.

```text
janus/\n└── __init__.py\n    ├── Queue[T]\n    │   ├── _SyncQueueProxy[T]\n    │   └── _AsyncQueueProxy[T]\n    ├── PriorityQueue[T] (extends Queue)\n    ├── LifoQueue[T] (extends Queue)\n    ├── BaseQueue[T] (Protocol)\n    ├── SyncQueue[T] (Protocol)\n    ├── AsyncQueue[T] (Protocol)\n    └── Exception classes (re-exports + custom)
```

--------------------------------

### Catching AsyncQueueShutDown

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/errors.md

Use this example to handle operations on a Janus asynchronous queue that has been shut down. This involves closing the queue and then using a try-except block to catch AsyncQueueShutDown.

```python
import janus

async_q = queue.async_q
await queue.aclose()

try:
    await async_q.put(item)
except janus.AsyncQueueShutDown:
    print("Queue is shutdown")
```

--------------------------------

### Getting an item from the Async Queue Proxy

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/sync-queue-proxy.md

Use the `get_nowait` method to retrieve an item from the asynchronous side of the queue without blocking. Raises `QueueEmpty` if the queue is empty.

```python
from queue import QueueEmpty

try:
    item = proxy.get_nowait()
except QueueEmpty:
    pass

```

--------------------------------

### Asynchronous Queue Integration

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/INDEX.md

Shows how to use a Janus async_q with an asynchronous function designed for `asyncio.Queue`. This illustrates the compatibility of the asynchronous interface.

```python
import asyncio

async def worker(q):
    # Works with any AsyncQueue-compatible object
    pass

async_q = janus_queue.async_q
await worker(async_q)  # Compatible
```

--------------------------------

### AsyncQueue Protocol

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/types.md

Defines the interface for an asynchronous queue, compatible with Python's `asyncio.Queue`. It supports asynchronous operations for putting items, getting items, and joining the queue.

```APIDOC
## AsyncQueue Protocol

### Description
Defines the asynchronous queue interface, compatible with `asyncio.Queue`. Inherits all methods from `BaseQueue[T]`.

### Methods

#### `put(item: T) -> None`
Asynchronously put an item into the queue.

#### `get() -> T`
Asynchronously get an item from the queue.

#### `join() -> None`
Asynchronously block until all items have been processed.
```

--------------------------------

### Basic Mixed Sync-Async Usage with Janus Queue

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/README.md

Demonstrates how to use a janus.Queue to pass items between a synchronous producer thread and an asynchronous consumer coroutine. Ensure the queue is properly closed after use.

```python
import asyncio
import janus
import threading

def producer_thread(sync_q):
    """Synchronous producer thread"""
    for i in range(5):
        sync_q.put(f"item_{i}")
    sync_q.join()

async def consumer_coro(async_q):
    """Asynchronous consumer"""
    for _ in range(5):
        item = await async_q.get()
        print(f"Got: {item}")
        async_q.task_done()

async def main():
    queue = janus.Queue()
    
    # Start producer in executor (thread pool)
    loop = asyncio.get_running_loop()
    producer = loop.run_in_executor(None, producer_thread, queue.sync_q)
    
    # Run consumer
    await consumer_coro(queue.async_q)
    
    # Wait for producer to finish
    await producer
    
    # Clean up
    await queue.aclose()

asyncio.run(main())
```

--------------------------------

### Synchronous Queue Get with Timeout

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/INDEX.md

Shows how to retrieve an item from the synchronous side of a Janus Queue with a specified timeout. A `janus.SyncQueueEmpty` exception is raised if the timeout is reached before an item is available.

```python
try:
    item = queue.sync_q.get(timeout=5.0)
except janus.SyncQueueEmpty:
    print("Timeout")
```

--------------------------------

### qsize

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/async-queue-proxy.md

Return the number of items currently in the queue.

```APIDOC
## qsize

### Description
Return the number of items in the queue.

### Method
`def qsize(self) -> int`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
size = async_q.qsize()
```

### Response
#### Success Response (200)
- **int** - The number of items in the queue.

#### Response Example
None
```

--------------------------------

### qsize

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/sync-queue-proxy.md

Return the approximate size of the queue. Note: not reliable in multithreaded contexts due to race conditions.

```APIDOC
## qsize

### Description
Return the approximate size of the queue. Note: not reliable in multithreaded contexts due to race conditions.

### Method
```python
def qsize(self) -> int
```

### Returns
int — Approximate number of items in the queue.

### Example
```python
size = sync_q.qsize()
```
```

--------------------------------

### Synchronous Timeout Handling

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/usage-patterns.md

Shows how to use `queue.sync_q.get()` with a timeout to retrieve items from a Janus queue in a synchronous context. Handles `queue.Empty` exceptions when the timeout is reached.

```python
import janus
import time
from queue import Empty

queue = janus.Queue()
sync_q = queue.sync_q

# Producer thread
import threading

def producer():
    time.sleep(2)  # Delay before putting
    sync_q.put("delayed-item")

thread = threading.Thread(target=producer)
thread.start()

# Try to get with timeout
try:
    item = sync_q.get(timeout=1.0)
except Empty:
    print("Timeout waiting for item")

try:
    item = sync_q.get(timeout=3.0)
    print(f"Got: {item}")
except Empty:
    print("Still no item")

thread.join()
```

--------------------------------

### SyncQueue Protocol

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/types.md

Defines the interface for a synchronous queue, compatible with Python's standard library `queue.Queue`. It includes methods for putting items, getting items, and joining the queue.

```APIDOC
## SyncQueue Protocol

### Description
Defines the synchronous queue interface, compatible with `queue.Queue` from the Python standard library. Inherits all methods from `BaseQueue[T]`.

### Methods

#### `put(item: T, block: bool = True, timeout: OptFloat = None) -> None`
Put an item into the queue, optionally blocking.

#### `get(block: bool = True, timeout: OptFloat = None) -> T`
Get an item from the queue, optionally blocking.

#### `join() -> None`
Block until all items have been processed (task_done called for each).
```

--------------------------------

### Catching ValueError for Invalid Synchronous Queue Operations

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/errors.md

Demonstrates how to catch a ValueError when performing an invalid operation on a synchronous Janus queue, such as calling task_done() without a corresponding put() or get().

```python
import janus

import queue
sync_q = queue.Queue()
try:
    sync_q.task_done()  # Without matching put()/get()
except ValueError as e:
    print(f"Invalid operation: {e}")
```

--------------------------------

### Producer-Consumer with Janus Queue

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/README.md

Demonstrates the producer-consumer pattern using a Janus Queue for inter-thread communication between async tasks. Ensure the queue is closed after use.

```python
import asyncio
import janus

async def main():
    queue = janus.Queue()
    
    async def producer():
        for i in range(100):
            await queue.async_q.put(i)
    
    async def consumer():
        for _ in range(100):
            item = await queue.async_q.get()
            queue.async_q.task_done()
    
    await asyncio.gather(producer(), consumer())
    await queue.aclose()

asyncio.run(main())
```

--------------------------------

### Put item into SyncQueue (blocking)

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/configuration.md

Use SyncQueue.put to add an item. By default, it blocks indefinitely until space is available. Can be configured to not block or to use a timeout.

```python
sync_q.put(item, block=True, timeout=None)
```

```python
sync_q.put(item)  # Block indefinitely
```

```python
sync_q.put(item, block=False)  # Raise immediately if full
```

```python
sync_q.put(item, timeout=5.0)  # Block up to 5 seconds
```

--------------------------------

### On Queue Object Methods

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/INDEX.md

Methods available directly on the queue object for managing its lifecycle.

```APIDOC
## On Queue Object Methods

### `close()`

#### Description
Immediately shut down the queue. This is a synchronous operation.

#### Returns
- `None`

### `wait_closed()`

#### Description
Wait for any pending asynchronous tasks related to the queue to complete after it has been closed.

#### Returns
- `Awaitable[None]`

### `aclose()`

#### Description
Asynchronously shut down the queue and wait for any pending tasks to complete. This combines the functionality of `close()` and `wait_closed()`.
```

--------------------------------

### Create Unbounded Janus LifoQueue

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/configuration.md

Use this to create a new mixed sync-async LIFO queue (stack) with no size limit. Items are retrieved in the reverse order they were added. The maxsize parameter defaults to 0 for an unbounded queue.

```python
queue = janus.LifoQueue(maxsize=0)
```

```python
queue = janus.LifoQueue()
```

--------------------------------

### Create Janus PriorityQueue

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/README.md

Instantiate a priority queue where items are retrieved based on their priority.

```python
import janus

queue = janus.PriorityQueue()
```

--------------------------------

### Create Janus Priority Queue with Max Size

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/configuration.md

Use this to create a new mixed sync-async priority queue with a specified maximum number of items. The queue will block if it reaches this size.

```python
queue = janus.PriorityQueue(maxsize=50)
```

--------------------------------

### Import Synchronous Queue Exceptions

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/types.md

Imports standard 'Empty' and 'Full' exceptions from Python's 'queue' module, aliasing them as SyncQueueEmpty and SyncQueueFull for use with Janus synchronous queues.

```python
from queue import Empty as SyncQueueEmpty
from queue import Full as SyncQueueFull
```

--------------------------------

### Close Janus Queue Immediately

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/queue.md

A shortcut for `shutdown(immediate=True)`, closing the queue and discarding any remaining items.

```python
def close(self) -> None
```

--------------------------------

### Import Specific Janus Classes

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/api-reference/public-api.md

Demonstrates how to import specific queue classes and exception types from the Janus library. This is a common pattern for selective imports.

```python
from janus import Queue, PriorityQueue, LifoQueue
from janus import SyncQueueEmpty, AsyncQueueEmpty
```

--------------------------------

### Create Unbounded Janus Priority Queue

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/configuration.md

Use this to create a new mixed sync-async priority queue with no size limit. Items are retrieved by priority. The maxsize parameter defaults to 0 for an unbounded queue.

```python
queue = janus.PriorityQueue(maxsize=0)
```

```python
queue = janus.PriorityQueue()
```

--------------------------------

### Implement PriorityQueue

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/architecture.md

Implements a priority queue using Python's heapq module. Items are expected to be comparable, typically tuples like (priority, data).

```python
def _init(self, maxsize: int) -> None:
    self._heap_queue: list[T] = []
    
    def _qsize(self) -> int:
        return len(self._heap_queue)
    
    def _put(self, item: T) -> None:
        heappush(self._heap_queue, item)
    
    def _get(self) -> T:
        return heappop(self._heap_queue)
```

--------------------------------

### Handle Shutdown Exceptions

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/README.md

Shows how to catch `SyncQueueShutDown` and `AsyncQueueShutDown` exceptions that are raised when attempting operations on a shutdown Janus queue.

```python
import janus

queue.shutdown()

try:
    sync_q.put(item)
except janus.SyncQueueShutDown:
    print("Queue is shutdown")

try:
    await async_q.get()
except janus.AsyncQueueShutDown:
    print("Queue is shutdown")
```

--------------------------------

### Janus Project File Organization

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/README.md

Outlines the directory structure for the Janus project, including the location of documentation files and API references.

```text
/workspace/home/output/\n├── README.md (this file)\n├── types.md\n├── errors.md\n├── configuration.md\n└── api-reference/\n    ├── queue.md\n    ├── sync-queue-proxy.md\n    ├── async-queue-proxy.md\n    ├── priority-queue.md\n    └── lifo-queue.md
```

--------------------------------

### Janus Documentation Navigation Structure

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/MANIFEST.md

Illustrates the hierarchical structure of the Janus project documentation, showing the relationships between different markdown files.

```markdown
INDEX.md ──┬──> README.md
           ├──> usage-patterns.md
           ├──> types.md
           ├──> errors.md
           ├──> configuration.md
           ├──> architecture.md
           └──> api-reference/
                ├──> public-api.md
                ├──> queue.md
                ├──> sync-queue-proxy.md
                ├──> async-queue-proxy.md
                ├──> priority-queue.md
                └──> lifo-queue.md
```

--------------------------------

### Import Asynchronous Queue Exceptions

Source: https://github.com/aio-libs/janus/blob/master/_autodocs/types.md

Imports standard 'QueueEmpty' and 'QueueFull' exceptions from Python's 'asyncio' module, aliasing them as AsyncQueueEmpty and AsyncQueueFull for use with Janus asynchronous queues.

```python
from asyncio import QueueEmpty as AsyncQueueEmpty
from asyncio import QueueFull as AsyncQueueFull
```