Create Solace Output Adapter Spring

Create Solace Output Adapter Spring

Solace Output Adapter Introduction
A custom built output adapter called Solace Output adapter will be developed in Java using Spring framework that can be used to publish messages on Solace compliant messaging bus. It can be embedded to any java based project to send message in JMS Text Message formats.

Connection to Solace Messaging server
By default adapter will be configured to connect Solace Messaging server one time only before throwing an exception, if it fails to connect messaging server, will throw an exception called “com.solacesystems.jcsmp.JCSMPErrorResponseException”. Client can configure this default property to retry connection to certain number of times at set interval before throwing exception. To do this, client can set SERVER_NUM_RETRIES and SERVER_RECONNECT_INTERVAL in seconds attributes in properties file. The server-num-retries attribute controls the number of times the adapter will try to connect before throwing an exception. The default value will be set to 0 which specified no retry will be attempted; maximum value is MAX_INT. The SERVER_RECONNECT_INTERVAL in seconds attribute will control time duration between next connection attempts will be made. The default value to this attribute will be 30 seconds.

Message Routing
Destination routing of Solace messaging is configurable through routing section in the properties file with below attributes:

• HOST
• VPN_NAME
• USERNAME
• PASSWORD
• QUEUE_NAME
• DELIVERY_MODE

All fields are required. If any of the fields are missing the adapter will throw an exception “com.solacesystems.jcsmp.JCSMPErrorResponseException” with a detail message. If more than one queue names needs to be specifies then this should be separated with a comma. It’s also possible to specify multiple servers for clustered environment; they should also separate by comma.

Solace Message Translation
This adapter will accept message only as a String format. It will convert the String to JMS Text Message then publish the message to the Solace bus.

Solace output writer and Delivery Mode
Adapter will create Solace message to JMS Text Message type and deliver them to the Solace Messaging bus. Solace message can be delivered with either of two delivery mode: DeliveryMode.PERSISTENT or DeliveryMode.NON_PERSISTENT and by default DeliveryMode.PERSISTENT will be set by the adapter and it can be change to DeliveryMode.NON_PERSISTENT in configuration properties file with attribute name: DeliveryMode. This could be finding in delivery mode section in the properties file.

Type checking and Error Handling

Solace output adapter will perform two checks during process start. It will ensure that a value for its configuration which is required and mentioned in the document has been specified. Default configuration will be set for the field which is not required. If any of the required filed has not been set, “com.solacesystems.jcsmp.JCSMPErrorResponseException” will be thrown with proper message.

Solace adapter logging

By default logging is enabled and it’s configurable through Solace.properties file. If user disables logging then it will only disable logging for info and debug. Note: All logging error level will be enabled and it’s not configurable.

Now let’s dive into its implementation using Spring framework. Below are needed:

• Eclipse 4.0 or up
• JDK 1.8
• Maven 3.2
• Solace version 6.2.0.64

Note: Its assumed that you have access to Solace server and Output queue has been created.

• Create maven project name: SolaceOutputAdapter belwo is final project structure:

• Please add below dependency in your pom.xml:

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>

	<groupId>com.wfc.otc.solaceadapter</groupId>
	<version>0.0.1-SNAPSHOT</version>
	<packaging>jar</packaging>

	<name>SolaceOutputAdapter</name>
	<url>http://maven.apache.org</url>

	<properties>
		<org.springframework.version>4.1.5.RELEASE</org.springframework.version>
		<log4j.version>1.2.16</log4j.version>
		<org.apache.log4j.version>2.1</org.apache.log4j.version>
		<com.tibco.version>5.1.2</com.tibco.version>
		<jms.version>1.1</jms.version>
	</properties>

	<dependencies>

		<!-- Application Context (depends on spring-core, spring-expression, spring-aop, 
			spring-beans) This is the central artifact for Spring's Dependency Injection 
			Container and is generally always defined -->
		<dependency>
			<groupId>org.springframework</groupId>
			<artifactId>spring-context</artifactId>
			<version>${org.springframework.version}</version>
		</dependency>

		<!-- Log4j -->
		<dependency>
			<groupId>org.apache.logging.log4j</groupId>
			<artifactId>log4j-api</artifactId>
			<version>${org.apache.log4j.version}</version>
		</dependency>

		<dependency>
			<groupId>org.apache.logging.log4j</groupId>
			<artifactId>log4j-core</artifactId>
			<version>${org.apache.log4j.version}</version>
		</dependency>

		<!--TIBCO -->
		<dependency>
			<groupId>com.tibco</groupId>
			<artifactId>tibjms</artifactId>
			<version>${com.tibco.version}</version>
		</dependency>

		<dependency>
			<groupId>javax.jms</groupId>
			<artifactId>jms</artifactId>
			<version>${jms.version}</version>
		</dependency>

		<dependency>
			<groupId>org.springframework</groupId>
			<artifactId>spring-jms</artifactId>
			<version>${org.springframework.version}</version>
		</dependency>


		<dependency>
			<groupId>com.solacesystems.jms</groupId>
			<artifactId>sol-jms</artifactId>
			<version>6.2.0.64</version>
		</dependency>

		<dependency>
			<groupId>com.solacesystems.jcsmp</groupId>
			<artifactId>sol-jcsmp</artifactId>
			<version>6.2.0.64</version>
		</dependency>

		<dependency>
			<groupId>com.solacesystems.common</groupId>
			<artifactId>sol-common</artifactId>
			<version>6.2.0.64</version>
		</dependency>

		<dependency>
			<groupId>solace</groupId>
			<artifactId>sol-commons-lang</artifactId>
			<version>2.2</version>
		</dependency>

	</dependencies>
	
	<build>				
 		
		<resources>
			<resource>
				<directory>src/main/config</directory>
				<excludes>
					<exclude>**/*.xml</exclude>
					<exclude>**/*.properties</exclude>
				</excludes>
			</resource>
		    <resource>
				<directory>src/main/resources</directory>
				<excludes>
					<exclude>**/*.xml</exclude>
					<exclude>**/*.properties</exclude>
				</excludes>
			</resource>
		</resources> 
		
		<plugins>
			<plugin>
				<groupId>org.apache.maven.plugins</groupId>
				<artifactId>maven-dependency-plugin</artifactId>
				<version>2.8</version>
				<executions>
					<execution>
						<id>copy-dependencies</id>
						<phase>package</phase>
						<goals>
							<goal>copy-dependencies</goal>
						</goals>
						<configuration>
							<outputDirectory>${project.build.directory}/lib</outputDirectory>
							<overWriteReleases>false</overWriteReleases>
							<overWriteSnapshots>false</overWriteSnapshots>
							<overWriteIfNewer>true</overWriteIfNewer>
						</configuration>
					</execution>
				</executions>
			</plugin>
						
		</plugins> 
		
	</build>


	<artifactId>SolaceOutputAdapter</artifactId>
</project>

• log4j2.xml:

<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="INFO">

	<Properties>
		<Property name="envrionment.target">DEV</Property>
	</Properties>

	<Properties>
		<Property name="logging.dir">./</Property>
	</Properties>

	<Appenders>
		<Console name="Console" target="SYSTEM_OUT">
			<PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n" />
		</Console>

		<RollingFile name="RollingFile"
			fileName="./log/rolling-file.log"	filePattern="${sys:logging.dir}/logs/rolling-file-%d{yyyy-MM-dd}-%i.log">
			<PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n" />
			<!-- TODO:Change to time based policy -->
			<Policies>
				<TimeBasedTriggeringPolicy interval="1"	modulate="true" />
				<SizeBasedTriggeringPolicy size="100 MB" />
			</Policies>
			<DefaultRolloverStrategy max="4" />
		</RollingFile>
	</Appenders>

	<Loggers>
		<Root level="info">
			<AppenderRef ref="Console" />
			<!-- <AppenderRef ref="file" /> -->
			<AppenderRef ref="RollingFile" />
		</Root>
	</Loggers>
</Configuration>

• Solace.properties:

#Solace Message Routing

SOLACE_HOST=HOST
SOLACE_VPN=VPN
SOLACE_USERNAME=USER NAME
SOLACE_PASSWORD=Password
SOLACE_OUTPUT_QUEUE_NAME=QUEUE NAME

#Solace output writer and Delivery Mode

DELIVERY_MODE=PERSISTENT
#DELIVERY_MODE=NON_PERSISTENT

#Connection settings:

SERVER_NUM_RETRIES=0

#In seconds

SERVER_RECONNECT_INTERVAL=30

#Logging

LOG_DETAILS=false

• solace-context.xml:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:util="http://www.springframework.org/schema/util"
	xmlns:tx="http://www.springframework.org/schema/tx" xmlns:aop="http://www.springframework.org/schema/aop"
	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.1.xsd
                           http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-4.1.xsd
                           http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.1.xsd
                           http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.1.xsd">

	<bean id="solaceOutputAdapter" class="com.javahonk.solace.messaging.SolaceOutputAdapter">
		<constructor-arg value="${SOLACE_HOST}" />
		<constructor-arg value="${SOLACE_VPN}" />
		<constructor-arg value="${SOLACE_USERNAME}" />
		<constructor-arg value="${SOLACE_PASSWORD}" />
		<constructor-arg value="${SOLACE_OUTPUT_QUEUE_NAME}" />
		<constructor-arg value="${DELIVERY_MODE}" />
	</bean>
	
	<bean id="solacePublisherImpl" class="com.javahonk.solace.messaging.SolacePublisherImpl" />

</beans>

• spring-context.xml:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xmlns:context="http://www.springframework.org/schema/context"
	xmlns:jdbc="http://www.springframework.org/schema/jdbc"
	xmlns:tx="http://www.springframework.org/schema/tx"
	xmlns:util="http://www.springframework.org/schema/util"
	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.1.xsd
		http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc-4.1.xsd
		http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.1.xsd
		http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-4.1.xsd">
	
	<context:annotation-config />
	<context:component-scan base-package="com.javahonk.solace, com.javahonk.solace.messaging" />
	
	<bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
		<property name="locations">
			<list>
				<value>classpath:messaging/solace.properties</value>							
			</list>
		</property>
		<property name="ignoreUnresolvablePlaceholders" value="true"/>
	</bean>	

	<import resource="solace-context.xml"/>
		
</beans>

• SolaceOutputAdapter.java: Main output adapter class which connect to Solace and provide all supported method to send message to Solace messaging bus:

package com.javahonk.solace.messaging;

import java.util.HashMap;
import java.util.Map;

import javax.annotation.PostConstruct;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Value;

import com.solacesystems.jcsmp.DeliveryMode;
import com.solacesystems.jcsmp.EndpointProperties;
import com.solacesystems.jcsmp.InvalidPropertiesException;
import com.solacesystems.jcsmp.JCSMPException;
import com.solacesystems.jcsmp.JCSMPFactory;
import com.solacesystems.jcsmp.JCSMPProperties;
import com.solacesystems.jcsmp.JCSMPSession;
import com.solacesystems.jcsmp.JCSMPStreamingPublishEventHandler;
import com.solacesystems.jcsmp.Queue;
import com.solacesystems.jcsmp.TextMessage;
import com.solacesystems.jcsmp.XMLMessageProducer;

public class SolaceOutputAdapter {
	
	private static final Logger logger = LogManager.getLogger(SolaceOutputAdapter.class.getName());
	
	private String host;
	private String vpn;
	private String userName;
	private String userPassword;
	private String solaceOutputQueueName;
	private String deliveryMode;
	private  XMLMessageProducer publisher;	
	private Queue queue;
	
	@Value("${SERVER_NUM_RETRIES}")
	private int serverNumRetries;
	
	@Value("${SERVER_RECONNECT_INTERVAL}")
	private int serverReconnectInterval;
	
	@Value("${LOG_DETAILS}")
	private boolean logDetails;
	
	private int retries = 0;
	
	public SolaceOutputAdapter(String host, String vpn, String userName,
			String userPassword, String solaceOutputQueueName,
			String deliveryMode) {
		this.host = host;
		this.vpn = vpn;
		this.userName = userName;
		this.userPassword = userPassword;
		this.solaceOutputQueueName = solaceOutputQueueName;
		this.deliveryMode = deliveryMode;
	}
	
	@PostConstruct
	public void init() {
		
		try {
			
			if (logDetails)
			logger.info("Connection info:\n Host: {} \n VPN: {}\n User Name: {}\n Password: {}\n Queue Name: {}\n Delivery Mode: {}\n",
					host, vpn, userName, userPassword, solaceOutputQueueName,
					deliveryMode);

			final EndpointProperties endpointProps = new EndpointProperties();	        
	        endpointProps.setPermission(EndpointProperties.PERMISSION_CONSUME);
	        endpointProps.setAccessType(EndpointProperties.ACCESSTYPE_EXCLUSIVE);
			queue = JCSMPFactory.onlyInstance().createQueue(" "+solaceOutputQueueName);	        
	        
			getSession().provision(queue, endpointProps, JCSMPSession.FLAG_IGNORE_ALREADY_EXISTS);
	        
	        publisher = getSession().getMessageProducer(
            new JCSMPStreamingPublishEventHandler() {
                public void responseReceived(String messageID) {
                	logger.info("Producer received response for msg ID {}",messageID);
                }
                public void handleError(String messageID, JCSMPException e, long timestamp) {
                	logger.info("Producer received error for msg ID {}", messageID,timestamp,e);
                }
            });
	        
		} catch (Exception e) {
			
			retries++;	
			logger.error("Number of retries to Solace: {}", retries);
			logger.error("Error creating connection to Solace:", e);						
			while (retries < serverNumRetries) {
				try {					
					Thread.sleep(serverReconnectInterval*1000);
					init();
				} catch (Exception e1) {
					logger.error("Error creating connection to Solace", e);
				}								
			}
		} 
	}
	
	public JCSMPSession getSession() {
		
		JCSMPSession session = null;
		
		try {
			
			JCSMPProperties properties = new JCSMPProperties();
	        properties.setProperty(JCSMPProperties.HOST, host);
	        properties.setProperty(JCSMPProperties.VPN_NAME, vpn);
	        properties.setProperty(JCSMPProperties.USERNAME, userName);
	        properties.setProperty(JCSMPProperties.PASSWORD, userPassword);
	        session = JCSMPFactory.onlyInstance().createSession(properties);
	        
		} catch (InvalidPropertiesException e) {
			logger.error("Failed to create Solace session", e);
		} 
		
		return session;		
	}
	
	public TextMessage getTextMessage(String message) {
		
		TextMessage msg = JCSMPFactory.onlyInstance().createMessage(TextMessage.class);
		if (deliveryMode.trim().equalsIgnoreCase("NON_PERSISTENT")) {
			msg.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
		}else {
			msg.setDeliveryMode(DeliveryMode.PERSISTENT);
		}        
        msg.setText(message);
        
        return msg;
	}
	
	public Queue getQueue() {
		
		return queue;		
	}
	
	public XMLMessageProducer getPublisher() {
		
		return publisher;		
	}
	
	public void sendMessage(String message) {
		
		try {
			
			logger.info("Sending message: {}", message);
			getPublisher().send(getTextMessage(message), getQueue());
			
		} catch (JCSMPException e) {
			logger.error("Failed to send message", e);
		}	
	}
	
	public boolean isConnected() {
		
		return !getSession().isClosed();		
	}
	
	public Map<String, String> getConnectionInfo() {
		
		Map<String, String> connectionInfo = new HashMap<String, String>();
		
		connectionInfo.put("host", host);
		connectionInfo.put("vpn", vpn);
		connectionInfo.put("userName", userName);
		connectionInfo.put("solaceOutputQueueName", solaceOutputQueueName);
		connectionInfo.put("deliveryMode", deliveryMode);
		
		return connectionInfo;		
	}
	
	

}

• SolacePublisher.java: An interface which will be exposed to the client to send message to Solace messaging server:

package com.javahonk.solace.messaging;

public interface SolacePublisher {
	void publishMessage(String message);
}

• SolacePublisherImpl.java: This is real class which implements SolacePublisher and send message to Solace messaging server:

package com.javahonk.solace.messaging;

import javax.annotation.Resource;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

public class SolacePublisherImpl implements SolacePublisher{

	@Resource(name="solaceOutputAdapter")
	private SolaceOutputAdapter solaceOutputAdapter;
	
	private static final Logger logger = LogManager.getLogger(SolacePublisherImpl.class.getName());	
	
	@Override
	public void publishMessage(String message) {
		
		logger.info("Connected. About to send message to queue {}",solaceOutputAdapter.getQueue().getName());			
		
		solaceOutputAdapter.sendMessage(message);		
        
		logger.info("Message sent. Exiting.");
		
	}

}

• Now its time to run our application. Please use class SolaceApplicationTest.java which will load Spring context and send message to Solace server:

package com.javahonk.solace;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import com.javahonk.solace.messaging.SolaceOutputAdapter;
import com.javahonk.solace.messaging.SolacePublisherImpl;

public class SolaceApplicationTest {
	
	private static final Logger logger = LogManager.getLogger(SolaceApplicationTest.class.getName());
	
	public static void main(String[] args) {
		
		logger.info("Solace Messaging Middleware started.");
		
		ApplicationContext context = new ClassPathXmlApplicationContext("spring-context.xml");
		
		SolaceOutputAdapter solaceOutputAdapter = (SolaceOutputAdapter) context.getBean(SolaceOutputAdapter.class);
		
		System.out.println(solaceOutputAdapter.isConnected());
		
		System.out.println("Connection info:"+solaceOutputAdapter.getConnectionInfo());
		
		solaceOutputAdapter.sendMessage("Hello from Java Honk");
		
		//By using interface 
		
		SolacePublisherImpl solaceMessagePublisherImpl = (SolacePublisherImpl) context.getBean(SolacePublisherImpl.class);

		solaceMessagePublisherImpl.publishMessage("Hello from Java Honk");		
		
		
		((AbstractApplicationContext) context).close();			
		
	}
}

• Download source code: SolaceOutputAdapter

Reference:

• Solace Integration