package org.eclipse.scanning.event;

import java.lang.reflect.Method;
import java.net.URI;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Comparator;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TreeSet;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.QueueBrowser;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;

import org.eclipse.scanning.api.event.IEventConnectorService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Reads a Queue of json beans from the activemq queue and deserializes the 
 * json beans to a specific class.
 * 
 * @author Matthew Gerring
 *
 * @param <T>
 */
class QueueReader<T> {
	
	private static final Logger logger = LoggerFactory.getLogger(QueueReader.class);
	
	private Comparator<? super T> comparator;

	private IEventConnectorService service;

	public QueueReader(IEventConnectorService service) {
		this(service, null);
	}
	
	public QueueReader(IEventConnectorService service, Comparator<? super T> comparator) {
		this.service    = service;
		this.comparator = comparator;
	}

	
	/**
	 * Read the status beans from any queue.
	 * Returns a list of optionally date-ordered beans in the queue.
	 * 
	 * @param uri
	 * @param queueName
	 * @param clazz
	 * @param monitor
	 * @return
	 * @throws Exception
	 */
	public List<T> getBeans(final URI uri, final String queueName) throws Exception {
		
		QueueConnection qCon = null;
		try {	        
			QueueConnectionFactory connectionFactory = (QueueConnectionFactory)service.createConnectionFactory(uri);
			qCon  = connectionFactory.createQueueConnection(); // This times out when the server is not there.
			QueueSession    qSes  = qCon.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
			Queue queue   = qSes.createQueue(queueName);
			qCon.start();

			QueueBrowser qb = qSes.createBrowser(queue);
			@SuppressWarnings("rawtypes")
			Enumeration  e  = qb.getEnumeration();


			final Collection<T> list;
			if (comparator!=null) {
				list = new TreeSet<T>(comparator);
			} else {
				list = new ArrayList<T>(17);
			}

			while(e.hasMoreElements()) {
				Message m = (Message)e.nextElement();
				if (m==null) continue;
				if (m instanceof TextMessage) {
					TextMessage t = (TextMessage)m;
					String json   = t.getText();
					@SuppressWarnings("unchecked")
					final T bean = (T)service.unmarshal(json, null);
					list.add(bean);
				}
			}
			return list instanceof List ? (List<T>)list : new ArrayList<T>(list);

		} finally {
			if (qCon!=null) qCon.close();
		}

	}

	/**
	 * 
	 * @param uri
	 * @param topicName
	 * @param clazz
	 * @param monitorTime
	 * @return
	 */
	public Map<String, T> getHeartbeats(final URI uri, final String topicName, final Class<T> clazz, final long monitorTime) throws Exception {
		
		final Map<String, T> ret = new HashMap<String, T>(3);
		Connection topicConnection = null;
		try {
			ConnectionFactory connectionFactory  = (ConnectionFactory)service.createConnectionFactory(uri);
			topicConnection = connectionFactory.createConnection();
			topicConnection.start();

			Session session = topicConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);

			final Topic           topic    = session.createTopic(topicName);
			final MessageConsumer consumer = session.createConsumer(topic);


			MessageListener listener = new MessageListener() {
				public void onMessage(Message message) {		            	
					try {
						if (message instanceof TextMessage) {
							TextMessage t = (TextMessage) message;
							final T bean = (T)service.unmarshal(t.getText(), clazz);
							Method nameMethod = bean.getClass().getMethod("getName");
							ret.put((String)nameMethod.invoke(bean), bean);
						}
					} catch (Exception e) {
						logger.error("Updating changed bean from topic", e);
					}
				}
			};
			consumer.setMessageListener(listener);
			Thread.sleep(monitorTime);
			
			return ret;

		} catch (Exception ne) {
			logger.error("Cannot listen to topic changes because command server is not there", ne);
			return null;
		} finally {
			topicConnection.close();
		}

	}
}

@@ -761,16 +761,73 @@
 ervice;%0A
+import org.eclipse.scanning.api.event.status.StatusBean;%0A
 import o
@@ -1063,16 +1063,35 @@
 Reader%3CT
+ extends StatusBean
 %3E %7B%0A%09%0A%09p
@@ -2781,20 +2781,32 @@
 l(json, 
-null
+StatusBean.class
 );%0A%09%09%09%09%09

Modify QueueReader to expect all beans to be StatusBeans