Skip to content

bovine_process

handle_outbox_item(item, actor) async

Synchronous part of processing an outbox item. In particular it contains the logic to assign ids to the item

Source code in bovine_process/bovine_process/__init__.py 
41
42
43
44
45
46
47
async def handle_outbox_item(
    item: ProcessingItem, actor: BovineStoreActor
) -> ProcessingItem | None:
    """Synchronous part of processing an outbox item. In particular
    it contains the logic to assign ids to the item"""

    return await default_outbox_process(item, actor)

process_inbox_item(item, actor, retry=False) async

Asynchronous processing of an item arriving at the inbox

Source code in bovine_process/bovine_process/__init__.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
async def process_inbox_item(
    item: ProcessingItem, actor: BovineStoreActor, retry=False
):
    """Asynchronous processing of an item arriving at the inbox"""

    try:
        logger.debug("Processing inbox request")
        result = await default_inbox_process(item, actor)
        return result
    except Exception:
        logger.error(">>>>> SOMETHING WENT WRONG IN PROCESSING <<<<<<")

        if item:
            item.dump()

        if not retry:
            wait_time = random.randint(3, 33)
            logger.error("Retrying in %d seconds", wait_time)
            await asyncio.sleep(wait_time)
            await process_inbox_item(item, actor, retry=True)

process_outbox_item(item, actor) async

Asynchronous part of processing an outbox item. In particular it contains the logic of sending the message to all followers

Source code in bovine_process/bovine_process/__init__.py 
50
51
52
53
54
55
56
async def process_outbox_item(item: ProcessingItem, actor: BovineStoreActor):
    """Asynchronous part of processing an outbox item. In particular
    it contains the logic of sending the message to all followers"""

    logger.debug("Processing outbox request")
    await default_async_outbox_process(item, actor)
    await send_outbox_item(item, actor)