Spring Integration Exception Handling

Here's another exercise where there were just no examples. I spent a half day misunderstanding that the phrase "in that channel" in THe Spring Integration Guide was referring to the ErrorChannel and not the channel that was processing the message. So, here is an example of configuring an exception handler for exceptions being thrown by a Transformer. Since the handler is listening on the errorChannel, the exception still ends up being logged. I assume that you would need to create your own errorChannel bean to get a different behavior with this strategy. Of course, you can also just use an errorChannel header which is probably the right thing to do anyway. In the configuration file:

    @Bean
    public IntegrationFlow exceptionFlow() {
        return IntegrationFlows
                .from(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME)
                .route(transformerExceptionRouter())
                .get();
    }
    
    @Bean
    public ErrorMessageExceptionTypeRouter transformerExceptionRouter() {
        ErrorMessageExceptionTypeRouter router = new ErrorMessageExceptionTypeRouter();
        router.setChannelMapping(CustomException.class.getCanonicalName(), "TransformerExceptionFlow");
        return router;
    }

    @Bean
    public IntegrationFlow transformerException() {
        return IntegrationFlows
                .from("TransformerExceptionFlow")
                .handle(MessageTransformationException.class, (p, h) -> {
                    EquipmentUpdateTransformerException e = (CustomException)(p.getCause().getCause());
                    System.out.println("Could process xml: " + e.getSource()); return e;
                 })
                .get();
    }

I created a CustomException to store the original xml document that was causing the exception:

@SuppressWarnings("serial")
public class CustomException extends RuntimeException {
    private final String source;

    public CustomException(Throwable cause, String source) {
        super(cause);
        this.source = source;
    }

    public String getSource() { return source; }
}

Subscribing to an IBM MQ JMS topic using a jndi .bindings file

Another task with not enough documentation and examples. I did find this stackoverflow post helpful. This post was also helpful but not quite correct. You don't need the custom JndiTemplate and the url should be the path to the directory where the bindings file is installed not the path to the file itself. This assumes that you have installed the .bindings file in /var/mqm. Essentially, you are making the filesystem your jndi store. It is also important to use the correct version of the IBM MQ libraries. We started using the 8.0.0.4 version of the libraries to connect to a 7.1 server and found the the connection name was being set improperly on the mq server. Getting the right libraries fixed the problem.

    @Bean
    public JndiTemplate jndiTemplate() {
        Properties jndiProps = new Properties();
        jndiProps.setProperty("java.naming.factory.initial", "com.sun.jndi.fscontext.RefFSContextFactory");
        jndiProps.setProperty("java.naming.provider.url", "file:///var/mqm/");

        JndiTemplate jndiTemplate = new JndiTemplate();
        jndiTemplate.setEnvironment(jndiProps);
        return jndiTemplate;
    }

    @Bean
    public JndiObjectFactoryBean jmsConnectionFactory() {
        JndiObjectFactoryBean jndiObjectFactoryBean = new JndiObjectFactoryBean();
        jndiObjectFactoryBean.setJndiTemplate(jndiTemplate());
        jndiObjectFactoryBean.setJndiName("CONNECTIONFACTORY");
        
        return jndiObjectFactoryBean;
    }
    
    @Bean
    public JndiObjectFactoryBean jmsTopicName() {
        JndiObjectFactoryBean jndiObjectFactoryBean = new JndiObjectFactoryBean();
        jndiObjectFactoryBean.setJndiTemplate(jndiTemplate());
        jndiObjectFactoryBean.setJndiName("YOURTOPICNAME");
        return jndiObjectFactoryBean;
    }

    @Bean
    public IntegrationFlow jmsInboundFlow() throws Exception{
        return IntegrationFlows
            .from(Jms.messageDrivenChannelAdapter((ConnectionFactory)jmsConnectionFactory().getObject())
                    .destination((Destination) jmsTopicName().getObject())
                    .configureListenerContainer(c -> { c.pubSubDomain(true); c.subscriptionDurable(true);  }))
             .handle(CharacterStreamWritingMessageHandler.stdout())
            .get();
    }

Subscribing to a JMS Topic with Spring Integration Java DSL

I couldn't find this on google or stack overflow so here is my working solution for connecting to an activemq JMS topic on localhost:

package com.twcable.smp.spring.integration;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.jms.Jms;
import org.springframework.integration.stream.CharacterStreamWritingMessageHandler;

import javax.jms.ConnectionFactory;

@Configuration
@EnableIntegration
@ComponentScan
@IntegrationComponentScan
public class IntegrationConfiguration {
    @Bean
    ConnectionFactory connectionFactory() {
        return new ActiveMQConnectionFactory("tcp://localhost:61616");
    }

    @Bean
    public IntegrationFlow jmsInboundFlow() {
        return IntegrationFlows.from(Jms.messageDrivenChannelAdapter(connectionFactory())
                                        .destination("TopicName")
                                        .configureListenerContainer( c -> {c.pubSubDomain(true); }))
                               .handle(CharacterStreamWritingMessageHandler.stdout())
                               .get();
    }
}