<commit_before>/*
 * Created on Jan 20, 2005
 *
 */
package fi.csc.microarray.messaging;

import java.io.InputStream;
import java.io.OutputStream;

import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TemporaryTopic;
import javax.jms.Topic;

import org.apache.log4j.Logger;

import fi.csc.microarray.messaging.message.NamiMessage;

/**
 * One topic ("channel") in the messaging fabric. Topics are publish-subscribe
 * type of communication. Used for sending and receiving messages.
 *  
 * @author akallio
 */
public class MessagingTopic {
	/**
	 * Logger for this class
	 */
	private static final Logger logger = Logger.getLogger(MessagingTopic.class);

	public static enum Type {
		NORMAL,
		TEMPORARY,
		NAMED_TEMPORARY;
	}
	
	public static enum AccessMode {
		READ,
		WRITE,
		READ_WRITE;
	}
	
	
	private MessageConsumer consumer;
	private Session session;
	private Topic topic;
	private MessagingEndpoint endpoint;
	
	public MessagingTopic(Session session, String topicName, Type type, AccessMode accessMode, MessagingEndpoint endpoint) throws JMSException {
		this.session = session;
		
		switch (type) {
		case NORMAL:
			this.topic = session.createTopic(topicName);
			break;
		case TEMPORARY:
			this.topic = session.createTemporaryTopic();
			break;
		default:
			throw new RuntimeException("unknown type " + type);
		}
		
		// create consumer only if we need to read from the topic
		if (accessMode == AccessMode.READ || accessMode == AccessMode.READ_WRITE) {
			this.consumer = session.createConsumer(topic);	
		}
		this.endpoint = endpoint;
	}

	
	protected void sendReplyableMessage(NamiMessage message, TempTopicMessagingListener replyListener, MessagingListener authenticationListener) throws JMSException {
		MessagingTopic tempTopic = new MessagingTopic(session, null, Type.TEMPORARY, AccessMode.READ_WRITE, endpoint);
		
		MultiplexingMessagingListener plexer = new MultiplexingMessagingListener();
		plexer.addChannel(Topics.MultiplexName.REPLY_TO.toString(), replyListener);
		if (authenticationListener != null) {
			plexer.addChannel(Topics.MultiplexName.AUTHORISE_TO.toString(), authenticationListener);
		}
		tempTopic.setListener(plexer);
		
		replyListener.setTempTopic(tempTopic);
		message.setReplyTo(tempTopic.topic);
		sendMessage(message);
		
	}
	
	/**
	 * Sends the message and creates a temporary topic for replying.
	 * 
	 * @param replyListener receives replies (if any) through hidden temporary topic
	 */
	public void sendReplyableMessage(NamiMessage message, TempTopicMessagingListener replyListener) throws JMSException {
		sendReplyableMessage(message, replyListener, null);
	}
	
	/**
	 * The basic message sending method. Sends a message without reply possibility.
	 */
	public void sendMessage(NamiMessage message) throws JMSException {

		// log
		logger.debug("sending " + message);
		
		// marshal message to MapMessage
		MapMessage mapMessage = session.createMapMessage();
		message.marshal(mapMessage);
		
		session.createProducer(topic).send(mapMessage);
	}

	/**
	 * Returns the JMS-name of the topic.
	 */
	public String getName() throws JMSException {
		return topic.getTopicName();
	}
<commit_msg>Close producer after send message.<commit_after>/*
 * Created on Jan 20, 2005
 *
 */
package fi.csc.microarray.messaging;

import java.io.InputStream;
import java.io.OutputStream;

import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TemporaryTopic;
import javax.jms.Topic;

import org.apache.log4j.Logger;

import fi.csc.microarray.messaging.message.NamiMessage;

/**
 * One topic ("channel") in the messaging fabric. Topics are publish-subscribe
 * type of communication. Used for sending and receiving messages.
 *  
 * @author akallio
 */
public class MessagingTopic {
	/**
	 * Logger for this class
	 */
	private static final Logger logger = Logger.getLogger(MessagingTopic.class);

	public static enum Type {
		NORMAL,
		TEMPORARY,
		NAMED_TEMPORARY;
	}
	
	public static enum AccessMode {
		READ,
		WRITE,
		READ_WRITE;
	}
	
	
	private MessageConsumer consumer;
	private Session session;
	private Topic topic;
	private MessagingEndpoint endpoint;
	
	public MessagingTopic(Session session, String topicName, Type type, AccessMode accessMode, MessagingEndpoint endpoint) throws JMSException {
		this.session = session;
		
		switch (type) {
		case NORMAL:
			this.topic = session.createTopic(topicName);
			break;
		case TEMPORARY:
			this.topic = session.createTemporaryTopic();
			break;
		default:
			throw new RuntimeException("unknown type " + type);
		}
		
		// create consumer only if we need to read from the topic
		if (accessMode == AccessMode.READ || accessMode == AccessMode.READ_WRITE) {
			this.consumer = session.createConsumer(topic);	
		}
		this.endpoint = endpoint;
	}

	
	protected void sendReplyableMessage(NamiMessage message, TempTopicMessagingListener replyListener, MessagingListener authenticationListener) throws JMSException {
		MessagingTopic tempTopic = new MessagingTopic(session, null, Type.TEMPORARY, AccessMode.READ_WRITE, endpoint);
		
		MultiplexingMessagingListener plexer = new MultiplexingMessagingListener();
		plexer.addChannel(Topics.MultiplexName.REPLY_TO.toString(), replyListener);
		if (authenticationListener != null) {
			plexer.addChannel(Topics.MultiplexName.AUTHORISE_TO.toString(), authenticationListener);
		}
		tempTopic.setListener(plexer);
		
		replyListener.setTempTopic(tempTopic);
		message.setReplyTo(tempTopic.topic);
		sendMessage(message);
		
	}
	
	/**
	 * Sends the message and creates a temporary topic for replying.
	 * 
	 * @param replyListener receives replies (if any) through hidden temporary topic
	 */
	public void sendReplyableMessage(NamiMessage message, TempTopicMessagingListener replyListener) throws JMSException {
		sendReplyableMessage(message, replyListener, null);
	}
	
	/**
	 * The basic message sending method. Sends a message without reply possibility.
	 */
	public void sendMessage(NamiMessage message) throws JMSException {

		// log
		logger.debug("sending " + message);
		
		// marshal message to MapMessage
		MapMessage mapMessage = session.createMapMessage();
		message.marshal(mapMessage);
		
		MessageProducer producer = null;
		try {
			producer = session.createProducer(topic);
			producer.send(mapMessage);
		} finally {
			try {
				producer.close();
			} catch (Exception e) {
			}
		}
	}

	/**
	 * Returns the JMS-name of the topic.
	 */
	public String getName() throws JMSException {
		return topic.getTopicName();
	}

{'fraction_non_alphanumeric': 0.09112709832134293, 'fraction_numerical': 0.0020983213429256594, 'mean_word_length': 4.451797385620915, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6344615', 'n_tokens_mistral': 2156, 'n_tokens_neox': 1903, 'n_words': 512}