/*
 * Receiver.java Created on June 24, 2004, 9:31 AM
 */

package gov.va.med.esr.vitria;

// Java Classes
import java.io.File;
import java.io.FileWriter;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.Hashtable;
import java.util.List;
import java.util.MissingResourceException;
import java.util.ResourceBundle;

import javax.jms.BytesMessage;
import javax.jms.Message;
import javax.jms.MessageEOFException;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueReceiver;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.rmi.PortableRemoteObject;

/**
 * @author DNS   LEV
 */
public class Receiver implements MessageListener {
	private static boolean onlyTestConfig = false;

	// Lock object
	private static Object lock = new Object();

	// JMS resources - maintain to properly shutdown
	private static QueueConnectionFactory factory = null;

	private static InitialContext jndi = null;

	private QueueConnection connection = null;

	private QueueSession session = null;

	private QueueReceiver receiver = null;

	// Root directory to store messages
	private String msgDirectory = null;
	
	private boolean shouldWriteLog;
	
	private boolean shouldWriteMsg;		
	
	private boolean shouldValidateMessages;
	
	private int invalidMessageCount = 0;
	
	private int invalidMessageLimit = 100;
	
	public Receiver(String context, String factoryName, String serverName,
			String queueName, String msgDirectory, boolean shouldWriteLog, boolean shouldWriteMsg,
			boolean shouldValidateMessages) throws Exception {
		super();

		this.msgDirectory = msgDirectory;
		
		this.shouldWriteLog = shouldWriteLog;
		this.shouldWriteMsg = shouldWriteMsg;
		this.shouldValidateMessages = shouldValidateMessages;		
		
		if(!onlyTestConfig) {		
			// This is to emulate Vitria caching initial context and connection
			// factory
			if (jndi == null) {
				// Create a context
				Hashtable data = new Hashtable();
				data.put(Context.INITIAL_CONTEXT_FACTORY, context);
				data.put(Context.PROVIDER_URL, serverName);
				jndi = new InitialContext(data);
			}
			if (factory == null) {
				// Look up JMS connection factory
				factory = (QueueConnectionFactory) PortableRemoteObject.narrow(jndi
						.lookup(factoryName), QueueConnectionFactory.class);
			}
	
			// Look up JMS connection factory and queue
			Queue queue = (Queue) PortableRemoteObject.narrow(jndi
					.lookup(queueName), Queue.class);
	
			// Create all JMS resources to isolate the time it takes to send
			// messages
			connection = factory.createQueueConnection();
			session = connection
					.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
			receiver = session.createReceiver(queue);
	
			receiver.setMessageListener(this);
			connection.start();
		}
	}

	public void onMessage(Message message) {
		try {
			if(this.shouldWriteLog) {
				System.out.println("JMS Message consumed - " + message
						.getStringProperty(MockVitriaConstants.MESSAGE_ID));
			}
			
			boolean validMessage = true;
			
			if(this.shouldValidateMessages) {
				if(!isValidMessage(message)) {
					validMessage = false;
					if(this.shouldWriteLog) {
						System.out.println("Invalid message");
					}
				}
			}
			
			if (validMessage) {
				if(shouldWriteMsg) {
					writeToAFile(this.getMessageAsText(message), message
							.getJMSDestination().toString(), message
							.getStringProperty(MockVitriaConstants.MESSAGE_ID), this
							.formatDateTime(Calendar.getInstance().getTime()));
				}
			} else {
				this.invalidMessageCount++;
				if(this.invalidMessageCount > this.invalidMessageLimit) {
					synchronized (lock) {
						lock.notifyAll();
					}
				}
			}			
		} catch (Exception e) {
			System.out.println("Failed to receive a message" + e);
		}
	}

	public void close() {

		if (connection != null && session != null && receiver != null) {
			try {
				connection.close();
				session.close();
				receiver.close();
			} catch (Exception e) {
				System.out.println("Failed to close JMS resources ");
			}
		}
		connection = null;
		session = null;
		receiver = null;
	}

	private String getMessageAsText(Message message) throws Exception {

		String content = null;
		if (message instanceof TextMessage) {
			TextMessage text = (TextMessage) message;
			content = text.getText();
		} else if (message instanceof BytesMessage) {
			BytesMessage byteMessage = (BytesMessage) message;

			ArrayList bytes = new ArrayList();
			while (Boolean.valueOf(true).booleanValue()) {
				try {
					byte b = byteMessage.readByte();
					bytes.add(new Byte(b));
				} catch (MessageEOFException e) {
					break;
				}
			}
			bytes.trimToSize();
			byte[] data = new byte[bytes.size()];
			for (int i = 0; i < data.length; i++) {
				data[i] = ((Byte) bytes.get(i)).byteValue();
			}

			content = new String(data);
		}
		return content;
	}

	private boolean isValidMessage(Message message) throws Exception {

		/**
		 * text.setStringProperty("Message_Id", this.message_id);
		 * text.setStringProperty("Message_Type", this.messageType);
		 * text.setStringProperty("Sending_Application",
		 * this.sending_Application); text.setStringProperty("Sending_Site",
		 * this.sending_Site);
		 */

		String messageId = message.getStringProperty(MockVitriaConstants.MESSAGE_ID);
		return messageId != null;
	}

	private void writeToAFile(String messageText, String queueName,
			String messageId, String arrivalTime) throws Exception {

		if(this.shouldWriteLog) {
			System.out.println("Time of arrival for incoming message "
					+ arrivalTime);
			System.out.println("Message ID: " + messageId);
		}
		
		StringBuffer path = new StringBuffer(this.msgDirectory);
		path.append(File.separator).append(queueName);

		// Create a directory for each queue
		File directory = new File(path.toString());
		if (!directory.exists()) {
			directory.mkdirs();
		}

		// Create a file writer for each message
		path.append(File.separator).append(messageId).append(arrivalTime)
				.append(".txt");

		FileWriter fw = new FileWriter(path.toString());
		fw.write(messageText);
		fw.flush();
		fw.close();
	}

	private String formatDateTime(Date dateParam) {
		SimpleDateFormat simpleFormat = new SimpleDateFormat("yyyyMMddHHmmss");
		return dateParam != null ? simpleFormat.format(dateParam) : null;
	}

	public static void main(String[] args) throws Exception {

		if (args.length != 1) {
			displayHelp();
			System.exit(-1);
		}

		// Read properties from config.properties
		ResourceBundle bundle = getResourceBundle(args[0]);
		String context = null;
		String factoryName = null;
		String serverName = null;
		String msgDirectory = null;
		String serverNameOverride = null;
		String[] queues = null;
		
		
		String receiverMode = "outbound";
		try {
			receiverMode = bundle.getString(MockVitriaConstants.RECEIVER_MODE);
		} catch(Exception e) { }
		
		String jndiQueueSuffix = "";
		try {
			jndiQueueSuffix = bundle.getString(MockVitriaConstants.JNDI_QUEUE_SUFFIX_KEY);
		} catch(Exception e) { }

		// check for env variable override (eg, -D option)
		serverNameOverride = System.getProperty(MockVitriaConstants.SERVER_URL_KEY);
		
		if("outbound".equals(receiverMode)) {
			context = bundle.getString(MockVitriaConstants.JNDI_CTX_KEY);
			factoryName = bundle.getString(MockVitriaConstants.OUTBOUND_FACTORY_KEY);
			serverName = bundle.getString(MockVitriaConstants.OUTBOUND_URL_KEY);
			msgDirectory = bundle
					.getString(MockVitriaConstants.OUTBOUNDDIRETCORYKEY);
			queues = MockVitriaConstants.OUTBOUND_QUEUES_JNDI;
			
		} else if("inbound".equals(receiverMode)) {
			context = bundle.getString(MockVitriaConstants.JNDI_CTX_KEY);
			factoryName = bundle.getString(MockVitriaConstants.INBOUND_FACTORY_KEY);
			serverName = bundle.getString(MockVitriaConstants.INBOUND_URL_KEY);
			msgDirectory = bundle
					.getString(MockVitriaConstants.INBOUNDMESSAGEDIRECTORY);
			queues = MockVitriaConstants.INBOUND_QUEUES_JNDI;
		}
		
		if (queues != null) {
			String shouldWriteLogStr = bundle.getString(MockVitriaConstants.WRITE_LOG_KEY);
			String shouldWriteMsgsStr = bundle.getString(MockVitriaConstants.WRITE_FILE_KEY);
			String shouldValidateMsgsStr = bundle.getString(MockVitriaConstants.VALIDATE_MSGS_KEY);
			
			if(serverNameOverride != null) {
				serverName = serverNameOverride;
			}
			
			System.out.println(serverName + " - Number of candidate JMS Queues: "
					+ queues.length);
			List receivers = new ArrayList();
			Receiver receiver = null;
	
			String targetQueueJndiName = null;
			for (int counter = 0; counter < queues.length; counter++) {
				targetQueueJndiName = queues[counter] + jndiQueueSuffix;
				if (shouldConsumeFromQueue(queues[counter],	bundle)) {
					receiver = new Receiver(context, factoryName, serverName,
							targetQueueJndiName,
							msgDirectory,
							Boolean.valueOf(shouldWriteLogStr).booleanValue(),
							Boolean.valueOf(shouldWriteMsgsStr).booleanValue(),
							Boolean.valueOf(shouldValidateMsgsStr).booleanValue());
					receivers.add(receiver);
					System.out.println("Listener Started for Queue : " + targetQueueJndiName);				
				} else {
					System.out.println("....Listener NOT Started for Queue : " + targetQueueJndiName);
				}			
			}
			
			if(onlyTestConfig) System.exit(-1);
	
			// Register a shutdown hook to clean up resources
			Runtime.getRuntime().addShutdownHook(
					new ReceiverShutdownHandler(receivers));
	
			// Allow receivers to receive messages
			synchronized (lock) {
				lock.wait();
			}
		}
	}

	private static boolean shouldConsumeFromQueue(String queueId,
			ResourceBundle bundle) {
		boolean shouldConsume = true;
		// convention is the key for lookup is last element of queueId
		// (tokenized by ".")
		String key = queueId;
		String[] tokens = queueId.split("\\.");
		if (tokens.length > 0)
			key = tokens[tokens.length - 1];

		// look in config file
		try {
			shouldConsume = Boolean.valueOf(bundle.getString(key)).booleanValue();
		} catch(MissingResourceException e) {
			// ok, absence of an entry, assume true (ie, ok to consume)
			shouldConsume = true;
		}

		return shouldConsume;
	}

	private static void displayHelp() {
		System.out.println("\n\n");
		System.out.println("Usage: java [-DServerURL=XXX] gov.va.med.esr.Receiver <propertyFile>");
		System.out
				.println("Where: <propertyFile>.properties contains JMS configuration information");
	}

	private static ResourceBundle getResourceBundle(String fileName)
			throws Exception {

		if (fileName == null || fileName.length() == 0) {
			throw new IllegalArgumentException("Invalid file name");
		}
		// throw MissingResourceException if failed to locate a resource
		return ResourceBundle.getBundle(fileName);
	}

	public static class ReceiverShutdownHandler extends Thread {

		private List receivers = null;

		private ReceiverShutdownHandler(List receivers) {
			this.receivers = receivers;
		}

		public void run() {

			for (int i = 0; i < receivers.size(); i++) {
				Receiver receiver = (Receiver) receivers.get(i);
				receiver.close();
			}
		}
	}
}