How can I make my Spring integration flow generic to process different requests?

Question

I'm using spring integration in my current project. I want to have a reusable flow for different requests.

Config file

@Configuration
    public class IntegrationConfiguration {
      @Autowired LionsServiceImpl lionsService;
    
      long dbId = new SequenceGenerator().nextId();
  //   Main flow
  @Bean
  public IntegrationFlow flow() {
    return flow ->
        flow.handle(
            (payload, header) -> {
              lionService.validateRequest((LionRequest) payload);
              return payload;
            })
            .split()
            .channel(c -> c.executor(Executors.newCachedThreadPool()))
            .convert(LionRequest.class)
            .scatterGather(
                scatterer ->
                    scatterer
                        .applySequence(true)
                        .recipientFlow(flow1())
                        .recipientFlow(flow2())
                        .recipientFlow(flow3()),
                gatherer -> gatherer.releaseLockBeforeSend(true))
            .log()
            .aggregate(a -> a.outputProcessor(MessageGroup::getMessages))
            .channel("output-flow");
  }
      //   flow1
      @Bean
  public IntegrationFlow flow1() {
    return integrationFlowDefinition ->
        integrationFlowDefinition
            .channel(c -> c.executor(Executors.newCachedThreadPool()))
            .handle(
                (payload, header) -> {
                  try {
                    return lionService.saveRequest(
                        payload,
                        String.valueOf(dbId),
                        Integer.valueOf(
                            ((LionRequest) payload)
                                .getDetails()
                                .getId()),
                        ((SourceSystem) Objects.requireNonNull(header.get("sourceSystem")))
                            .getSourceSystemCode());
                  } catch (JsonProcessingException e) {
                    return e.getMessage();
                  }
                })
            .nullChannel();
  }
    
  //   flow2
  @Bean
  public IntegrationFlow flow2() {
    return integrationFlowDefination ->
        integrationFlowDefination
            .channel(c -> c.executor(Executors.newCachedThreadPool()))
            .handle(
                message ->
                    lionService.getData(
                        (LionRequest) message.getPayload(), SourceSystem.ONE))
            .log();
  }

  //  flow3
  @Bean
  public IntegrationFlow flow3() {
    return integrationFlowDefination ->
        integrationFlowDefination
            .channel(c -> c.executor(Executors.newCachedThreadPool()))
            .handle(
                message ->
                    lionService.prepareCDRequest(
                        (LionRequest) message));
  }

  @Bean
  public MessageChannel replyChannel() {
    return MessageChannels.executor("output-flow", outputExecutor()).get();
  }

  @Bean
  public ThreadPoolTaskExecutor outputExecutor() {
    ThreadPoolTaskExecutor pool = new ThreadPoolTaskExecutor();
    pool.setCorePoolSize(4);
    pool.setMaxPoolSize(4);
    return pool;
  }
}

Gateway

  @Gateway(requestChannel = "flow.input")
  void processLionRequest(
      @Payload LionRequest lionRequest,
      @Header("sourceSystem") SourceSystem sourceSystem);
}

I'm casting the payload with LionRequest in few of the places but for other requests I want it to work. If I pass CatRequest and my flow is similar then how do I overcome the Object casting that I'm doing in few places.

user_2756547 · Answer

Such a casting must go away from your flow definitions. Move it (or remove) altogether into the service method calls level.

You may have in your service overloaded methods for particular types - the framework determines the target method to call automatically by the payload in the message.

For example if we take your validation part, it could look like this:

class MyValidationService {

    LionRequest validateRequest(LionRequest request) {
        
    }

    CatRequest validateRequest(CatRequest request) {
        
    }
}

Then in that flow you just do:

flow.handle(myValidationService).

The framework will wrap this into a ServiceActivatingHandler which is going to delegate to this or that method according the payload type you send to the input channel of this flow.

Something similar could be done for any other endpoints you use in your flows.

Probably even that .convert(LoanProvisionRequest.class) could be delegated to some POJO via transform() with similar overloaded method signatures. Or... those methods can have a different names - the framework doesn't care what is that: it just takes a payload type and tries to find a candidate in the methods of the POJO you provide for this endpoint.

There is some explanation in docs: https://docs.spring.io/spring-integration/docs/current/reference/html/messaging-endpoints.html#service-activator-namespace

How can I make my Spring integration flow generic to process different requests?

1 Answers