弹簧-集成:外部路由选择

Question

我希望允许来电者传递外部选路滑，例如通过张贴：

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引导我所相信的，所以我构建了它(考虑到只有一个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所建议的，外部路由选择索引和外部路由选择本身都不应该存储在Spring中，而是可以使用消息头来为每个请求分别维护它们：

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