romeways-memory-queue 0.2.0

No project description provided

Romeways memory queue

This is an extra package for romeways for more details access the romeways Github page.

Configs

Queue

The only difference between the romeways.GenericQueueConfig and romeways_memory_queue.MemoryQueueConfig is the multiprocessing.Queue instance.

from dataclasses import dataclass
from multiprocessing import Queue

from romeways import GenericQueueConfig


@dataclass(slots=True, frozen=True)
class MemoryQueueConfig(GenericQueueConfig):
    connector_name: str
    frequency: float
    max_chunk_size: int
    sequential: bool
    queue: Queue

Connector

from dataclasses import dataclass

from romeways import GenericConnectorConfig

@dataclass(slots=True, frozen=True)
class MemoryConnectorConfig(GenericConnectorConfig):
    connector_name: str

Use case

from multiprocessing import Queue
import asyncio

import romeways

# Config the connector
queue = Queue()

# Create a queue config
config_q = romeways.MemoryQueueConfig(
    connector_name="memory-dev1",
    frequency=1,
    max_chunk_size=10,
    sequential=False,
    queue=queue
)

# Register a controller/consumer for the queue name
@romeways.queue_consumer(queue_name="queue.payment.done", config=config_q)
async def controller(message: romeways.Message):
    print(message)



config_p = romeways.MemoryConnectorConfig(connector_name="memory-dev1")

# Register a connector
romeways.connector_register(
    connector=romeways.MemoryQueueConnector, config=config_p, spawn_process=True
)

asyncio.run(romeways.start())