About Callback mechanism in the JCSMP

When a message is published to the incorrect topic or encounters a permission issue, no exception is raised. Although JCSMPStreamingPublishCorrelatingEventHandler has been implemented, the handleErrorEx and responseReceivedEx methods are not being invoked, making it difficult to track missing messages.

import java.io.IOException;
import java.util.LinkedList;
import java.util.concurrent.CountDownLatch;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.ibsplc.iloyal.plugins.projectspecific.config.EventsConfiguration;
import com.ibsplc.iloyal.plugins.projectspecific.dto.ExternalConfigurationDTO;
import com.ibsplc.iloyal.plugins.projectspecific.dto.JMSHeaderDTO;
import com.ibsplc.xibase.util.log.Log;
import com.ibsplc.xibase.util.log.factory.LogFactory;
import com.solacesystems.jcsmp.BytesXMLMessage;
import com.solacesystems.jcsmp.DeliveryMode;
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.JCSMPStreamingPublishCorrelatingEventHandler;
import com.solacesystems.jcsmp.JCSMPTransportException;
import com.solacesystems.jcsmp.SDTException;
import com.solacesystems.jcsmp.SDTMap;
import com.solacesystems.jcsmp.TextMessage;
import com.solacesystems.jcsmp.Topic;
import com.solacesystems.jcsmp.XMLMessageProducer;
@
public class EventsConfirmedPublish {

private static final Log LOGGER = LogFactory.getLogger(EventsConfirmedPublish.class.getName());

@ 
private EventsConfiguration eventsConfiguration;


final int count = 5;

final CountDownLatch latch = new CountDownLatch(count); // used for Synchronizing b/w threads


/**

 * A correlation structure. This structure is passed back to the publisher

 * callback when the message is acknowledged or rejected

 */

class MsgInfo {

	public volatile boolean acked = false;

	public volatile boolean publishedSuccessfully = false;

	public BytesXMLMessage sessionIndependentMsg = null;

	public final long id;


	public MsgInfo(long id) {

		this.id = id;

		LOGGER.log(Log.FINE, "[FRM] MsgInfo acked from constructor - {}", acked);

	}


	@Override

	public String toString() {

		return String.format("Message ID: %d, Published Confirmation: %b, Published Successfully: %b", id, acked,

				publishedSuccessfully);

	}

}


class PubCallback implements JCSMPStreamingPublishCorrelatingEventHandler {


	@Override

	public void handleErrorEx(Object arg0, JCSMPException arg1, long arg2) {

		if (arg0 instanceof MsgInfo) {

			MsgInfo i = (MsgInfo) arg0;

			i.acked = true;

			LOGGER.log(Log.SEVERE, "Message response (rejected) received for {}, error was {} \n", i, arg1 );

		} else {

	        LOGGER.log(Log.FINE, "Received error but correlation object is not of type MsgInfo");

	    }

		latch.countDown();

	}


	@Override

	public void responseReceivedEx(Object arg0) {

		if (arg0 instanceof MsgInfo) {

			MsgInfo i = (MsgInfo) arg0;

			i.acked = true;

			i.publishedSuccessfully = true;

			LOGGER.log(Log.FINE, "Message response (accepted) received for {} \n", i);

		} else {

	        LOGGER.log(Log.FINE, "Received response but correlation object is not of type MsgInfo");

	    }

		latch.countDown();

	}

}


public void sendMessage(JMSHeaderDTO jmsHeaderDTO, String outGoingMsg)

		throws JCSMPTransportException, JCSMPException, IOException {


	ExternalConfigurationDTO externalConfigs = eventsConfiguration.getConfigurations();

	final LinkedList<MsgInfo> msgList = new LinkedList<MsgInfo>();

	JCSMPSession session = null;

	try {

	LOGGER.log(Log.FINE, "ConfirmedPublish initializing...");

	// Create a JCSMP Session

	final JCSMPProperties properties = new JCSMPProperties();

	properties.setProperty(JCSMPProperties.USERNAME, externalConfigs.getSolaceConfigs().getSolaceUsername());

	properties.setProperty(JCSMPProperties.PASSWORD, externalConfigs.getSolaceConfigs().getSolacePassword());

	properties.setProperty(JCSMPProperties.HOST, externalConfigs.getSolaceConfigs().getSolaceHost());

	properties.setProperty(JCSMPProperties.VPN_NAME, externalConfigs.getSolaceConfigs().getSolaceVpn());

	session = JCSMPFactory.onlyInstance().createSession(properties);

	session.connect();

	LOGGER.log(Log.FINE, "JCSMP session connected");

	final Topic solaceTopic = JCSMPFactory.onlyInstance().createTopic(jmsHeaderDTO.getDestination());

	final XMLMessageProducer producer = session.getMessageProducer(new PubCallback());

	LOGGER.log(Log.FINE, "About to send message to Topic - {}", solaceTopic.getName());

	TextMessage message = createTextMessage(jmsHeaderDTO, outGoingMsg, externalConfigs);

	// The application will wait and confirm the message is published

    // successfully.

    // In this case, wrap the message in a MsgInfo instance, and

    // use it as a correlation key.

	for (int i=1; i<=count; i++) {

		final MsgInfo msgCorrelationInfo = new MsgInfo(count);

		msgCorrelationInfo.sessionIndependentMsg = message;

		msgList.add(msgCorrelationInfo);


		// Set the message's correlation key. This reference

        // is used when calling back to responseReceivedEx().

		message.setCorrelationKey(msgCorrelationInfo);

		// **Important** Set delivery mode to PERSISTENT for guaranteed messaging

        message.setDeliveryMode(DeliveryMode.PERSISTENT);

        message.setAckImmediately(true);

        LOGGER.log(Log.FINE, "Message - count {} - {}", i, message.toString());

		producer.send(message, solaceTopic);

	}


	LOGGER.log(Log.FINE, "Message Send. Processing replies");

	try {

		latch.await();

	} catch (InterruptedException ex) {

		LOGGER.log(Log.SEVERE, "[FRM] I was awoken while awaiting");

	}

	//Process the replies

	while (msgList.peek() != null) {

		final MsgInfo ackedMsgInfo = msgList.poll();

		LOGGER.log(Log.FINE, "Removing acknowledged message {} from application list.\n", ackedMsgInfo);

    }

	LOGGER.log(Log.FINE, "Message successfully published to Solace topic : {}", solaceTopic);

	}

	catch (InvalidPropertiesException ex) {

		LOGGER.log(Log.SEVERE, "InvalidPropertiesException occured while connecting to JCSMPSession session. {}",

				ex.getMessage());

		throw new InvalidPropertiesException(

				"InvalidPropertiesException occured while connecting to JCSMPSession session. {}", ex);

	} catch (JCSMPException ex) {

		LOGGER.log(Log.SEVERE, "JCSMPException occured while publishing event data. {}", ex.getMessage());

		throw new JCSMPException("JCSMPException occured while publishing event data. {}", ex);

	} finally {

		if (session != null) {

			session.closeSession();

		}

	}

}


private TextMessage createTextMessage(JMSHeaderDTO jmsHeaderDTO, String outGoingMsg,

		ExternalConfigurationDTO externalConfigs) throws SDTException {

	TextMessage message = JCSMPFactory.onlyInstance().createMessage(TextMessage.class);

	// Set message headers

    SDTMap map = JCSMPFactory.onlyInstance().createMap();

    map.putString(externalConfigs.getJmsHeaderKeys().getJmsCorrelation(), jmsHeaderDTO.getCorrelationID());

    map.putString(externalConfigs.getJmsHeaderKeys().getJmsContentType(), jmsHeaderDTO.getDataContentType());

    map.putString(externalConfigs.getJmsHeaderKeys().getJmsDestination(), jmsHeaderDTO.getDestination());

    map.putString(externalConfigs.getJmsHeaderKeys().getJmsEvent(), jmsHeaderDTO.getEvent());

    map.putString(externalConfigs.getJmsHeaderKeys().getJmsEventId(), jmsHeaderDTO.getEventId());



    message.setProperties(map);

    message.setText(outGoingMsg);

    message.setHTTPContentType(jmsHeaderDTO.getDataContentType());

    return message;

}

}

Please advise