Merge branch 'enable-topic' into 'master'

pub sub model implemented

See merge request !48

src/main/java/org/opengroup/osdu/core/ibm/messagebus/ActiveMQFactory.java

@@ -8,16 +8,19 @@ import java.util.HashSet;
 import java.util.Set;
 
 import javax.annotation.PostConstruct;
+import javax.inject.Inject;
 import javax.jms.Connection;
 import javax.jms.ConnectionFactory;
 import javax.jms.Session;
 
 import org.apache.http.HttpStatus;
 import org.opengroup.osdu.core.common.model.http.AppException;
+import org.opengroup.osdu.core.common.model.http.DpsHeaders;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.context.annotation.Lazy;
 import org.springframework.jms.JmsException;
 import org.springframework.jms.core.JmsTemplate;
@@ -35,6 +38,9 @@ public class ActiveMQFactory implements IMessageFactory {
 	@Autowired
 	public JmsTemplate jmsTemplate;
 	
+	@Autowired
+	DpsHeaders header;
+	
 	private Set<String> existingQueues = new HashSet<String>();
 	
 	@PostConstruct
@@ -62,6 +68,9 @@ public class ActiveMQFactory implements IMessageFactory {
 	public void sendMessage(String msg) {
 		sendMessage(DEFAULT_QUEUE_NAME, msg, -1L);
 		sendMessage(DEFAULT_QUEUE_COPY_NAME, msg, -1L); // For WKE 
+		//sending Message to topic
+		String topicName = prefix+ "-" +header.getPartitionId() + "-" + INDEXER_QUEUE_NAME + "-topic";
+		sendMessageToTopic(topicName, msg, -1L);
 	}
 
 	@Override
@@ -122,5 +131,19 @@ public class ActiveMQFactory implements IMessageFactory {
 		return queueNameWithPrefix;
 	}
 	
+	
+	public void sendMessageToTopic(String topicName, String msg, Long delay) {
+		try {
+			jmsTemplate.setDeliveryDelay(delay);
+			jmsTemplate.setPubSubDomain(true);
+			jmsTemplate.convertAndSend(topicName, msg);
+			logger.info("[xx] Sent '" + msg + "' to topic [" + topicName + "] with delay " + delay);
+		} catch (JmsException e) {
+			logger.error("Failed to publish status." +" topic name [" + topicName + "]");
+			logger.error(e.getMessage(), e);
+			e.printStackTrace();
+		}
+	}
+	
 
 }
\ No newline at end of file