Batch nodes

A batch node is the node type that turns one input message into many output messages, each routed independently to the next step. It is the right shape for paginated APIs, for splitting a collection into items, and for any fan-out where downstream parallelism matters.

For background on Batch as a platform primitive, see Concepts: Batch.

Class to extend #

Two kinds of batch node #

A batch node can either:

• Split a collection that already lives in the input message (no external call), or
• Iterate over a paginated API until the cursor is exhausted, emitting items as it goes.

The platform persists the iterator state, so a batch node that crashes mid-page resumes from the same cursor and does not duplicate items.

Splitting a collection #

// worker/src/Hubspot/Batch/SplitContacts.ts
import ABatchNode from '@orchesty/nodejs-sdk/dist/lib/Batch/ABatchNode';
import BatchProcessDto from '@orchesty/nodejs-sdk/dist/lib/Utils/BatchProcessDto';

export default class SplitContacts extends ABatchNode {
    public getName(): string { return 'split-contacts'; }

    public async processAction(dto: BatchProcessDto): Promise<BatchProcessDto> {
        const input = dto.getJsonData() as { contacts: object[] };
        for (const contact of input.contacts) {
            dto.addItemFromArray(contact);
        }
        return dto;
    }
}

Each call to addItemFromArray() produces one output message. The downstream node sees them as independent inputs.

Iterating a paginated API #

The platform passes the cursor to your node through the message metadata so you can resume from the right page. See the full pattern (with end-of-pages handling and the iterate-only flag) in Patterns: Pagination and batch. Per-node throughput on the batch consumer and the ordering trade-off are covered in Operations: Prefetch.

The skeleton looks like this:

public async processAction(dto: BatchProcessDto): Promise<BatchProcessDto> {
    const cursor = dto.getBatchCursor() ?? '';

    const response = await this.callApi(cursor);
    for (const item of response.items) {
        dto.addItemFromArray(item);
    }

    if (response.next) {
        dto.setBatchCursor(response.next);
    }

    return dto;
}

The platform calls processAction() again with the new cursor until you stop setting one. From the topology's point of view, every emitted item is a separate message and rides through the rest of the topology in parallel.

Retry policy #

A Batch node is a connector — the same retry rules apply. If a page fetch fails transiently, throw OnRepeatException(seconds, maxHops, message); the platform retries the same page (cursor included) without losing iterator state. For deterministic failures (4xx that won't fix itself), let the message stop and move to Trash; iteration continues with the next page.

import OnRepeatException from '@orchesty/nodejs-sdk/dist/lib/Exception/OnRepeatException';

if (response.status >= 500) {
    throw new OnRepeatException(60, 10, `Upstream ${response.status}`);
}

See Error handling and retries for the full mechanism.

Register the batch node #

container.setBatch(new SplitContacts());

See also #

• Concepts: Batch
• Patterns: Pagination and batch
• Custom nodes
• Connectors
• Error handling and retries
• Reference: Node.js / ABatchNode
• Reference: PHP / BatchAbstract