Skip to main content

Processor

A Processor runs business logic against a message. It can enrich the message with derived data, validate it, route it sideways via the RouteContext, or drop it. Processors are the most common place to put per-message logic that does not fit format conversion (Transformer) or routing predicates (Condition).

PropertyValue
Java interfacecom.connamara.sdk.v1.adapter.components.Processor
MethodMessage<?> process(Message<?> message, RouteContext routeContext)
Pipeline slotAfter Transformers, before the target Connector.
Return contractA Message<?> to continue the chain (often the input reference, sometimes a derived message), or null to drop the message.
Side effectsAllowed and expected. The RouteContext exposes side-channel sends and metadata access.

Complete example​

package com.example.myplugin;

import com.connamara.sdk.v1.adapter.RouteContext;
import com.connamara.sdk.v1.adapter.components.Processor;
import com.connamara.sdk.v1.common.component.ManifestResource;
import com.connamara.sdk.v1.common.message.DerivedMessage;
import com.connamara.sdk.v1.common.message.Message;

import java.util.Map;

@ManifestResource("""
    {
      "id": "tag-count-processor",
      "pluginId": "my-plugin",
      "functionalType": "tag-count-processor",
      "displayName": "Tag Count Processor",
      "category": "PROCESSOR",
      "configuration": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "type": "object",
        "title": "Tag Count Settings",
        "x-order": ["headerName"],
        "properties": {
          "headerName": {
            "type": "string",
            "title": "Output Header Name",
            "description": "The header key where the tag count will be stored.",
            "default": "x-tag-count"
          }
        }
      }
    }
    """)
public class TagCountProcessor implements Processor {

    private final String headerName;

    public TagCountProcessor(Map<String, Object> configuration) {
        this.headerName = (String) configuration.getOrDefault("headerName", "x-tag-count");
    }

    @Override
    public Message<?> process(Message<?> message, RouteContext routeContext) {
        int count = message.getHeaders().toMap().size();
        return DerivedMessage.derive(message, message.getPayload(), Map.of(headerName, count));
    }
}

Key rules​

  • Return value drives the pipeline. Returning the input message reference passes it through unchanged. Returning a derived message replaces it. Returning null drops the message; downstream Processors and the target are skipped, and the source Connector receives the Ack.
  • Use RouteContext for side-channel work. The context exposes the route's metadata and provides hooks for sending the message to the configured target (routeContext.sendToTarget(message)) ahead of pipeline completion, when a Processor wants to commit to delivery before optional follow-up work runs.
  • Never mutate the input message. All changes flow through DerivedMessage.derive(...). Mutating in place corrupts upstream and downstream observers.
  • Throwing is a route-level failure. Any exception escapes the Processor and is handed to the route's ErrorProcessor chain, if present, or surfaces as a Nack to the source Connector.

Test pattern​

@Test
@DisplayName("GIVEN a message with 3 headers WHEN processed THEN output header contains count 3")
void process_countsHeaders() {
    TagCountProcessor processor = new TagCountProcessor(Map.of("headerName", "x-tag-count"));

    Map<String, Object> headerMap = new HashMap<>();
    headerMap.put("h1", "v1");
    headerMap.put("h2", "v2");
    headerMap.put("h3", "v3");

    Message<String> message = mock(Message.class);
    when(message.getPayload()).thenReturn("test");
    when(message.getHeaders()).thenReturn(new MessageHeaders(headerMap));

    RouteContext routeContext = mock(RouteContext.class);

    Message<?> result = processor.process(message, routeContext);

    assertEquals(3, result.getHeaders().get("x-tag-count"));
}

When to reach for a Processor​

NeedReach for
Enrich the message with a derived field, audit trail, or computed metadata.Processor.
Validate the message; drop it on failure.Processor (return null to drop).
Convert the payload between two wire formats.Transformer.
Decide whether the route should run at all for this message.Condition.
React to an exception thrown from a previous Processor.ErrorProcessor.

See also​