Spring Integration: connection to multiple MQ servers by config

Question

I do have a Spring Boot 5 application and I also have it running against one IBM MQ server.

Now we want it to connect to three or more MQ servers. My intention is now to just add XY connection infos to the environment and then I get XY MQConnectionFactory beans and al the other beans that are needed for processing.

At the moment this is what I have:

    @Bean
    @Qualifier(value="MQConnection")
    public MQConnectionFactory getIbmConnectionFactory() throws JMSException {
        MQConnectionFactory factory = new MQConnectionFactory();
// seeting all the parameters here

        return factory;
    }

But this is quite static. Is there an elegant way of doing this?

I stumbled about IntegrationFlow. Is this a possibly working solution?

Thanks for all your tipps!

KR

Solution

Based on Artem Bilan's response I built this class.

@Configuration
public class ConnectionWithIntegrationFlowMulti {
    protected static final Logger LOG = Logger.create();
    
    @Value("${mq.queue.jms.sources.queue.queue-manager}")
    private String queueManager;


    @Autowired
    private ConnectionConfig connectionConfig;


    @Autowired
    private SSLSocketFactory sslSocketFactory;

    @Bean
    public MessageChannel queureader() {
        return new DirectChannel();
    }

    @Autowired
    private IntegrationFlowContext flowContext;
    
    @PostConstruct
    public void processBeanDefinitionRegistry() throws BeansException {

        Assert.notEmpty(connectionConfig.getTab().getLocations(), "At least one CCDT file locations must be provided.");
        for (String tabLocation : connectionConfig.getTab().getLocations()) {
            try {
                IntegrationFlowRegistration theFlow = this.flowContext.registration(createFlow(tabLocation)).register();
                LOG.info("Registered bean flow for %s with id = %s", queueManager, theFlow.getId());
            } catch (JMSException e) {
                LOG.error(e);
            }
        }
    }
    
    public IntegrationFlow createFlow(String tabLocation) throws JMSException {
        LOG.info("creating ibmInbound");
        return IntegrationFlows.from(Jms.messageDrivenChannelAdapter(getConnection(tabLocation)).destination(createDestinationBean()))
                .handle(m -> LOG.info("received payload: " + m.getPayload().toString()))
                .get();
    }

    
    public MQConnectionFactory getConnection(String tabLocation) throws JMSException {
        MQConnectionFactory factory = new MQConnectionFactory();
        
        // doing stuff
        
        return factory;
    }

    @Bean
    public MQQueue createDestinationBean() {
        LOG.info("creating destination bean");
        MQQueue queue = new MQQueue();

        try {
            queue.setBaseQueueManagerName(queueManager);
            queue.setBaseQueueName(queueName);

        } catch (Exception e) {
            LOG.error(e, "destination bean: Error for integration flow");
        }
        return queue;
    }


}

score 1 · Answer 1 · answered Feb 04 '20 at 18:50

If you are fine with creating them statically, you can create the beans as you are now (each having a unique qualifier), but you can access them all dynamically in your services / components by having an @Autowired List<MQConnectionFactory> field or @Autowired Map<String, MQConnectionFactory> field. Spring will automatically populate the fields with all of the beans of type MQConnectionFactory

In the the Map implementation, the String will be the qualifier value.

If you also want to create the beans dynamically based on some properties, etc, it gets a little more complicated. You will need to look into something along the lines of instantiating beans at runtime

score 1 · Accepted Answer · answered Feb 04 '20 at 19:34

With Spring Integration you can create IntegrationFlow instances dynamically at runtime. For that purpose there is an IntegrationFlowContext with its registration() API. The returned IntegrationFlowRegistrationBuilder as a callback like:

/**
     * Add an object which will be registered as an {@link IntegrationFlow} dependant bean in the
     * application context. Usually it is some support component, which needs an application context.
     * For example dynamically created connection factories or header mappers for AMQP, JMS, TCP etc.
     * @param bean an additional arbitrary bean to register into the application context.
     * @return the current builder instance
     */
    IntegrationFlowRegistrationBuilder addBean(Object bean);

So, your MQConnectionFactory instances can be populated alongside with the other flow, used as references in the particular JMS components and registered as beans, too.

See more info in docs: https://docs.spring.io/spring-integration/docs/5.2.3.RELEASE/reference/html/dsl.html#java-dsl-runtime-flows

Thanks. I'm working on it and try to implement this. But where is the right place to register these beans programatically? I thought about `implements BeanDefinitionRegistryPostProcessor` but there I do not have the environment variables for the connections. — Tobias Schnupp, Feb 05 '20 at 13:09
`@PostConstruct` in some your `@Configuration` with injected `IntegrationFlowContext` should be enough to iterate your environment and register dynamic flows with an y possible support beans. — Artem Bilan, Feb 05 '20 at 14:43

Spring Integration: connection to multiple MQ servers by config

Solution

2 Answers2