gpt4 book ai didi

java - Spring Integration 从 REST 服务获取分页结果

转载 作者:行者123 更新时间:2023-11-30 01:55:45 25 4
gpt4 key购买 nike

我正在研究与 REST 服务的集成,其想法是由出站网关 marketingCategoryOutboundGateway 进行轮询。由 HttpRequestExecutingMessageHandler 实现。网关向 REST 服务发出请求并将其响应推送到 marketingCategory channel 。网关本身由 marketingCategoryPollerMessageSource 创建的消息触发。使用makeTriggeringMessage工厂方法。

问题在于该服务返回分页结果。我一些东西可以在 marketingCategory 上监听 channel ,除了我已经拥有的服务激活器之外,检查是否响应并推送一条由 makeTriggeringMessage 创建的递增页码的新消息。到marketingCategoryPoller channel ,以便代码会循环旋转,直到从 REST 服务获取所有页面。

Spring Integration是否允许创建这样的过滤器:在输入 channel 上接收一条消息,根据条件对其进行测试,并在条件为真时将新消息推送到输出 channel ?

代码:

//Responses from the REST service go to this channel
@Bean("marketingCategory")
MessageChannel marketingCategory() { return new PublishSubscribeChannel();}

//This channel is used to trigger the outbound gateway which makes a request to the REST service
@Bean
MessageChannel marketingCategoryPoller() {return new DirectChannel();}

//An adapter creating triggering messages for the gateway
@Bean
@InboundChannelAdapter(channel = "marketingCategoryPoller", poller = @Poller(fixedDelay = "15000"))
public MessageSource<String> marketingCategoryPollerMessageSource() { return () -> makeTriggeringMessage(1);}

//A factory for producing messages which trigger the gateway
private Message<String> makeTriggeringMessage(int page) {
//make a message for triggering marketingCategoryOutboundGateway
return MessageBuilder.withPayload("")
.setHeader("Host", "eclinic")
.setHeader("page", page)
.build();
}

//An outbound gateway, makes a request to the REST service and returns the response to marketingCategory channel
@Bean
@ServiceActivator(inputChannel = "marketingCategoryPoller")
public MessageHandler marketingCategoryOutboundGateway(@Qualifier("marketingCategory") MessageChannel channel) {
//make a request to the REST service and push the response to the marketingCategory channel
}

//handler for REST service responses
@Bean
@ServiceActivator(inputChannel = "marketingCategory")
public MessageHandler marketingCategoryHandler() {
return (msg) -> {
//process the categories returned by marketingCategoryOutboundGateway
};
}

最佳答案

我根据这篇文章找到了解决方案Read and download from a paginated REST-Services with spring integration :

  1. 触发与 REST 服务通信的出站网关,并使用带有轮询器的入站 channel 适配器将响应推送到 channel 。入站 channel 适配器是一个消息源,它最初生成一条消息,该消息的 header 指示要从 REST API 获取的页码。出站网关使用页面消息头来生成指定所需页面的 url

  2. 出站网关推送 REST 服务响应的 channel 有 2 个订阅者:

    2.1。一个服务激活器,它对获取的数据执行某些操作

    2.2。一个过滤器,它检查这是否是最后一页,如果不是,它将消息进一步发送到 header 丰富器使用的另一个 channel

  3. 收到消息后, header 丰富器会增加其页面 header ,并将消息进一步推送到触发出站网关的 channel ,网关读取递增的页头并从 REST 服务获取下一页

  4. 循环不断旋转,直到 REST 服务返回最后一页。过滤器不会让此消息传递到 header 丰富器,从而打破循环。

完整代码:

@Configuration
public class IntegrationConfiguration {

private final ApiGateConfig apiGateConfig;

IntegrationConfiguration(ApiGateConfig apiGateConfig) {
this.apiGateConfig = apiGateConfig;
}

@Bean("marketingCategory")
MessageChannel marketingCategory() {
return new PublishSubscribeChannel();
}

@Bean
MessageChannel marketingCategoryPoller() {
return new DirectChannel();
}

@Bean
MessageChannel marketingCategoryPollerNextPage() {
return new DirectChannel();
}

@Bean
@InboundChannelAdapter(channel = "marketingCategoryPoller", poller = @Poller(fixedDelay = "15000"))
public MessageSource<RestPageImpl<MarketingCategory>> marketingCategoryPollerMessageSource() {
return () -> makeTriggeringMessage(0);
}

/**
* Build a gateway triggering message
*/
private Message<RestPageImpl<MarketingCategory>> makeTriggeringMessage(int page) {
return MessageBuilder.withPayload(new RestPageImpl<MarketingCategory>())
.setHeader("Host", "eclinic")
.setHeader("page", page)
.build();
}

@Bean
@ServiceActivator(inputChannel = "marketingCategoryPoller")
public MessageHandler marketingCategoryOutboundGateway(@Qualifier("marketingCategory") MessageChannel channel) {

String uri = apiGateConfig.getUri() + "/marketingCategories?page={page}";

//the type of the payload
ParameterizedTypeReference<RestPageImpl<MarketingCategory>> type = new ParameterizedTypeReference<>() {
};

//page number comes from the message
SpelExpressionParser expressionParser = new SpelExpressionParser();
var uriVariables = new HashMap<String, Expression>();
uriVariables.put("page", expressionParser.parseExpression("headers.page"));

HttpRequestExecutingMessageHandler handler = new HttpRequestExecutingMessageHandler(uri);
handler.setHttpMethod(HttpMethod.GET);
handler.setExpectedResponseTypeExpression(new ValueExpression<>(type));
handler.setOutputChannel(channel);
handler.setUriVariableExpressions(uriVariables);

return handler;
}

@Bean
@ServiceActivator(inputChannel = "marketingCategory")
public MessageHandler marketingCategoryHandler() {
return (msg) -> {
var page = (RestPageImpl<MarketingCategory>) msg.getPayload();

System.out.println("Page #" + page.getNumber());

page.getContent().forEach(c -> System.out.println(c.getMarketingCategory()));

};
}

@Filter(inputChannel = "marketingCategory", outputChannel = "marketingCategoryPollerNextPage")
public boolean marketingCategoryPaginationFilter(RestPageImpl<MarketingCategory> page) {
return !page.isLast();
}

@Bean
@Transformer(inputChannel = "marketingCategoryPollerNextPage", outputChannel = "marketingCategoryPoller")
HeaderEnricher incrementPage() {
Map<String, HeaderValueMessageProcessor<?>> headersToAdd = new HashMap<>();
Expression expression = new SpelExpressionParser().parseExpression("headers.page+1");

var valueProcessor = new ExpressionEvaluatingHeaderValueMessageProcessor<>(expression, Integer.class);
valueProcessor.setOverwrite(true);

headersToAdd.put("page", valueProcessor);
return new HeaderEnricher(headersToAdd);
}
}

关于java - Spring Integration 从 REST 服务获取分页结果,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/54591622/

25 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com