Spring-Integration: external routing slip

Question

I would like to allow callers to pass an external routing slip, e.g. by posting:

POST http://localhost:8080/transform?routing-slip=capitalize&routing-slip=lowercase
Content-Type: text/plain

camelCase

It should be possible to use the given routing-slip array as external routing slip from a pojo:

@Bean
public IntegrationFlow transformerChain(RoutingSlipRouteStrategy routeStrategy) {
    return IntegrationFlows.from(
        Http.inboundGateway("/transform")
            .headerExpression("routingSlipParam", 
                    "#requestParams['routing-slip']")
            .requestPayloadType(String.class))
        .enrichHeaders(spec -> spec.header(
            IntegrationMessageHeaderAccessor.ROUTING_SLIP,
              new RoutingSlipHeaderValueMessageProcessor(
                 "@routePojo.get(request, reply)")
              )
        )
        .logAndReply();
}

The pojo can access the routingSlipParam header and you would think it can then hold the slip as internal state, or at least that is what TestRoutingSlipRoutePojo lead me to believe, so I built this (with a slight doubt, given that there is only one instance of the pojo):

public class ExternalRoutingSlipRoutePojo {

    private List<String> routingSlip;
    private int i = 0;

    public String get(Message<?> requestMessage, Object reply) {
        if (routingSlip == null) {
            routingSlip = (LinkedList)requestMessage.getHeaders()
                .get("routingSlipParam");
        }
        try {
            return this.routingSlip.get(i++);
        } catch (Exception e) {
            return null;
        }
    }

}

It turns out that this only works exactly once, which is not surprising after all - the index is incremented for every incoming message and the routing slip is never updated.

So I thought, sure, I have to hold the internal status for all incoming messages and came up with this RouteStrategy:

public class ExternalRoutingSlipRouteStrategy implements RoutingSlipRouteStrategy {

  private Map<UUID, LinkedList<String>> routingSlips = new WeakHashMap<>();
  private static final LinkedList EMPTY_ROUTINGSLIP = new LinkedList<>();

  @Override
  public Object getNextPath(Message<?> requestMessage,Object reply) {
    MessageHeaders headers = requestMessage.getHeaders();

    UUID id = headers.getId();
    if (!routingSlips.containsKey(id)) {
        @SuppressWarnings("unchecked")
        List<String> routingSlipParam = 
            headers.get("routingSlipParam", List.class);
        if (routingSlipParam != null) {
            routingSlips.put(id, 
                new LinkedList<>(routingSlipParam));
        }
    }
    LinkedList<String> routingSlip = routingSlips.getOrDefault(id,
        EMPTY_ROUTINGSLIP);
    String nextPath = routingSlip.poll();
    if (nextPath == null) {
        routingSlips.remove(id);
    }
    return nextPath;
  }
}

That does not work either because the strategy is not only called for the incoming message but also for all the new messages which are created by the dynamic routing, which of course have different IDs.

But it is only called twice for the original message, so the routing slip never gets exhausted and the application runs in an endless loop.

How can I make spring-integration use an external routing slip?

UPDATE:

As suggested by Gary Russel, neither the external routing slip index nor the external routing slip itself should be stored in the Spring bean, rather one can use message headers to maintain them separately for each request:

Http.inboundGateway("/transform")
    .headerExpression("routingSlipParam",
            "#requestParams['routing-slip']")
    .requestPayloadType(String.class))
.enrichHeaders(spec -> spec
    .headerFunction("counter",h -> new AtomicInteger())
    .header(IntegrationMessageHeaderAccessor.ROUTING_SLIP,
        new RoutingSlipHeaderValueMessageProcessor(externalRouteStrategy)
    )
)

The externalRouteStrategy is an instance of the following class:

public class ExternalRoutingSlipRouteStrategy implements 
        RoutingSlipRouteStrategy {

    @Override
    public Object getNextPath(Message<?> requestMessage, Object reply) {

        List<String> routingSlip = (List<String>) 
            requestMessage.getHeaders().get("routingSlipParam");

        int routingSlipIndex = requestMessage.getHeaders()
            .get("counter", AtomicInteger.class)
            .getAndIncrement();

        String routingSlipEntry;
        if (routingSlip != null 
            && routingSlipIndex < routingSlip.size()) {

            routingSlipEntry = routingSlip.get(routingSlipIndex);
        } else {
            routingSlipEntry = null;
        }
        return routingSlipEntry;
    }
}

For reference, I have published the example in Github.

Gary Russell Gary Russell · Accepted Answer · 2019-02-18T18:13:49

Go back to your first version and store i in a message header (AtomicInteger) in the header enricher.

.headerExpression("counter", "new java.util.concurrent.atomic.AtomicInteger()")

then

int i = requestMessage.getHeaders().get("counter", AtomicInteger.class).getAndIncrement();

Spring-Integration: external routing slip

1 Answers