Spring-Integration：外部路由单

Question

我想允许呼叫者通过外部routing slip，例如发布：

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

camelCase

应该可以使用给定的路由滑动阵列作为来自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();
}

pojo可以访问routingSlipParam标题，你会认为它可以保持滑动作为内部状态，或者至少这是TestRoutingSlipRoutePojo引导我相信的，所以我建立了这个（有一点点怀疑，因为只有一个实例of 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;
        }
    }

}

事实证明，这只能工作一次，毕竟这并不奇怪 - 索引会针对每个传入的消息递增，并且路由单永远不会更新。

所以我想，当然，我必须保留所有传入消息的内部状态，并想出了这个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;
  }
}

这不起作用，因为策略不仅要求传入消息，还要用于动态路由创建的所有新消息，当然这些消息具有不同的ID。

但它只对原始消息调用两次，因此路由选择永远不会耗尽并且应用程序在无限循环中运行。

如何使弹簧集成使用外部布线单？

更新：

正如Gary Russel所建议的那样，外部路由滑动索引和外部路由滑动本身都不应该存储在Spring bean中，而是可以使用消息头为每个请求单独维护它们：

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)
    )
)

externalRouteStrategy是以下类的实例：

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;
    }
}

作为参考，我在Github上发布了这个例子。

Answer 1

回到你的第一个版本并将i存储在标题扩充器的消息头（AtomicInteger）中。

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

然后

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