// SMSLib for Java v4
// A Java API library for sending and receiving SMS via a GSM modem
// or other supported gateways.
// Web Site: http://www.smslib.org
//
// Copyright (C) 2002-2012, Thanasis Delenikas, Athens/GREECE.
// SMSLib is distributed under the terms of the Apache License version 2.0
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package org.smslib.queues;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.PriorityBlockingQueue;
import org.smslib.OutboundMessage;
import org.smslib.Service;
import org.smslib.helper.Logger;

/**
 * @author Bassam Al-Sarori
 * @since 3.5
 */
public class DefaultQueueManager extends AbstractQueueManager
{
	public static final String MESSAGE_FILE_EXT = ".msg";

	private Map<String, PriorityBlockingQueue<OutboundMessage>> queueMap;

	private DelayQueue<ScheduledOutboundMessage> delayQueue;

	private String queueDirectory;

	private File pendingMessageDir;

	private File delayedMessageDir;

	public DefaultQueueManager()
	{
		super();
	}

	public DefaultQueueManager(String queueDirectory)
	{
		super();
		this.queueDirectory = queueDirectory;
	}

	public DefaultQueueManager(int delay)
	{
		super(delay);
	}

	public DefaultQueueManager(int delay, String queueDirectory)
	{
		super(delay);
		this.queueDirectory = queueDirectory;
	}

	@Override
	protected void init()
	{
		super.init();
		queueMap = new HashMap<String, PriorityBlockingQueue<OutboundMessage>>();
		delayQueue = new DelayQueue<ScheduledOutboundMessage>();
		if (queueDirectory == null)
		{
			queueDirectory = Service.getInstance().getSettings().QUEUE_DIRECTORY;
			if (queueDirectory == null)
			{
				Logger.getInstance().logInfo("Queue directory not defined. Queued messages will not be saved to filesystem.", null, null);
				return;
			}
		}
		File baseDir = new File(queueDirectory, "queue");
		pendingMessageDir = new File(baseDir, "pending");
		if (!pendingMessageDir.exists())
		{
			if (!pendingMessageDir.mkdirs())
			{
				Logger.getInstance().logError("Could not create directory for pending messages queue at " + pendingMessageDir.getPath(), null, null);
			}
		}
		else
		{
			Logger.getInstance().logDebug("loading pending messages..", null, null);
			loadPendingMessages();
		}
		delayedMessageDir = new File(baseDir, "delayed");
		if (!delayedMessageDir.exists())
		{
			if (!delayedMessageDir.mkdirs())
			{
				Logger.getInstance().logError("Could not create directory for delayed messages queue at " + delayedMessageDir.getPath(), null, null);
			}
		}
		else
		{
			Logger.getInstance().logDebug("loading delayed messages..", null, null);
			loadDelayedMessages();
		}
	}

	/* (non-Javadoc)
	 * @see org.smslib.queues.AbstractQueueManager#queueMessage(org.smslib.OutboundMessage)
	 */
	@Override
	public boolean queueMessage(OutboundMessage message)
	{
		if (message.getDeliveryDelay() > 0)
		{
			return addToDelayedQueue(message, true);
		}
		else
		{
			return addToGatewayQueue(message, true);
		}
	}

	/* (non-Javadoc)
	 * @see org.smslib.queues.AbstractQueueManager#removePendingMessage(org.smslib.OutboundMessage)
	 */
	@Override
	public boolean removePendingMessage(OutboundMessage message)
	{
		for (PriorityBlockingQueue<OutboundMessage> q : queueMap.values())
		{
			if (q.remove(message))
			{
				deletePendingMessage(message.getGatewayId(), message.getUuid());
				return true;
			}
		}
		return false;
	}

	/* (non-Javadoc)
	 * @see org.smslib.queues.AbstractQueueManager#removePendingMessage(java.lang.String)
	 */
	@Override
	public boolean removePendingMessage(String messageUUID)
	{
		for (PriorityBlockingQueue<OutboundMessage> q : queueMap.values())
		{
			for (OutboundMessage m : q)
			{
				if (m.getId().equalsIgnoreCase(messageUUID))
				{
					if (q.remove(m))
					{
						deletePendingMessage(m.getGatewayId(), messageUUID);
						return true;
					}
				}
			}
		}
		return false;
	}

	private boolean addToGatewayQueue(OutboundMessage message, boolean store)
	{
		PriorityBlockingQueue<OutboundMessage> queue = queueMap.get(message.getGatewayId());
		if (queue == null)
		{
			queue = new PriorityBlockingQueue<OutboundMessage>(50, new PriorityComparator());
			queueMap.put(message.getGatewayId(), queue);
		}
		boolean queued = queue.add(message);
		if (store && queued) storePendingMessage(message);
		return queued;
	}

	private boolean addToDelayedQueue(OutboundMessage message, boolean store)
	{
		boolean queued = delayQueue.add(new ScheduledOutboundMessage(message));
		if (store && queued) storeDelayedMessage(message);
		return queued;
	}

	private boolean storePendingMessage(OutboundMessage message)
	{
		if (queueDirectory == null) { return true; }
		File gatewayDir = new File(pendingMessageDir, message.getGatewayId().replace("/", "."));
		if (!gatewayDir.exists())
		{
			if (!gatewayDir.mkdir())
			{
				Logger.getInstance().logError("Queue directory could be created for gateway " + message.getGatewayId() + ". Could not create directory .." + gatewayDir.getPath(), null, null);
				return false;
			}
		}
		return serializeMessage(message, new File(gatewayDir, message.getUuid() + MESSAGE_FILE_EXT));
	}

	private boolean deletePendingMessage(String gatewayId, String messageUUID)
	{
		if (queueDirectory == null) { return true; }
		return new File(new File(pendingMessageDir, gatewayId), messageUUID + MESSAGE_FILE_EXT).delete();
	}

	private boolean deletePendingMessages(String gatewayId)
	{
		if (queueDirectory == null) { return true; }
		if (gatewayId == null) return emptyDirectory(pendingMessageDir, false);
		else return emptyDirectory(new File(pendingMessageDir, gatewayId), true);
	}

	private boolean storeDelayedMessage(OutboundMessage message)
	{
		if (queueDirectory == null) { return true; }
		return serializeMessage(message, new File(delayedMessageDir, message.getUuid() + MESSAGE_FILE_EXT));
	}

	private boolean deleteDelayedMessage(String messageUUID)
	{
		if (queueDirectory == null) { return true; }
		return new File(delayedMessageDir, messageUUID + MESSAGE_FILE_EXT).delete();
	}

	@Override
	public OutboundMessage pollDelayedMessage()
	{
		try
		{
			OutboundMessage message = delayQueue.take().getMessage();
			deleteDelayedMessage(message.getUuid());
			return message;
		}
		catch (InterruptedException e)
		{
			//ignored
		}
		return null;
	}

	@Override
	public OutboundMessage pollPendingMessage(String gatewayId)
	{
		PriorityBlockingQueue<OutboundMessage> queue = queueMap.get(gatewayId);
		if (queue == null) return null;
		OutboundMessage message = queue.poll();
		if (message != null) deletePendingMessage(gatewayId, message.getUuid());
		return message;
	}

	private boolean serializeMessage(OutboundMessage message, File toFile)
	{
		if (queueDirectory == null) { return true; }
		if (toFile.exists())
		{
			Logger.getInstance().logError("Cannot save Message " + message.getUuid() + " File already exist.", null, null);
			return false;
		}
		ObjectOutputStream out = null;
		try
		{
			out = new ObjectOutputStream(new FileOutputStream(toFile));
			out.writeObject(message);
			out.close();
		}
		catch (IOException e)
		{
			Logger.getInstance().logError("Cannot save Message " + message.getUuid(), e, null);
			return false;
		}
		return true;
	}

	private OutboundMessage deserializeMessage(File fromFile)
	{
		if (!fromFile.exists())
		{
			Logger.getInstance().logError("File of queued message doesn't exist " + fromFile.getPath(), null, null);
			return null;
		}
		ObjectInputStream in = null;
		OutboundMessage message = null;
		try
		{
			in = new ObjectInputStream(new FileInputStream(fromFile));
			message = (OutboundMessage) in.readObject();
			in.close();
		}
		catch (IOException e)
		{
			Logger.getInstance().logError("Could not read queued message from file " + fromFile.getPath(), e, null);
			return null;
		}
		catch (ClassNotFoundException e)
		{
			Logger.getInstance().logError("Could not read queued message from file " + fromFile.getPath(), e, null);
			return null;
		}
		return message;
	}

	private void loadPendingMessages()
	{
		File[] pendingDirs = pendingMessageDir.listFiles();
		for (File pendingDir : pendingDirs)
		{
			if (pendingDir.isDirectory())
			{
				File[] pendingFiles = pendingDir.listFiles();
				for (File pendingFile : pendingFiles)
				{
					if (pendingFile.getName().endsWith(MESSAGE_FILE_EXT))
					{
						addToGatewayQueue(deserializeMessage(pendingFile), false);
					}
				}
			}
		}
	}

	private void loadDelayedMessages()
	{
		File[] delayedFiles = delayedMessageDir.listFiles();
		for (File delayedFile : delayedFiles)
		{
			if (delayedFile.getName().endsWith(MESSAGE_FILE_EXT))
			{
				OutboundMessage message = deserializeMessage(delayedFile);
				if (message.getDeliveryDelay() > 0) addToDelayedQueue(message, false);
				else
				{
					addToGatewayQueue(message, true);
					deleteDelayedMessage(message.getUuid());
				}
			}
		}
	}

	class PriorityComparator implements Comparator<OutboundMessage>
	{
		@Override
		public int compare(OutboundMessage x, OutboundMessage y)
		{
			int comp = y.getPriority() - x.getPriority();
			if (comp == 0) comp = x.getDate().compareTo(y.getDate());
			return comp;
		}
	}

	@Override
	public int delayedQueueSize(String gatewayId)
	{
		return delayQueue.size();
	}

	@Override
	public Collection<OutboundMessage> getDelayedMessages()
	{
		List<OutboundMessage> messages = new ArrayList<OutboundMessage>();
		for (ScheduledOutboundMessage scheduledOutboundMessage : delayQueue)
		{
			messages.add(scheduledOutboundMessage.getMessage());
		}
		return messages;
	}

	@Override
	public Collection<OutboundMessage> getPendingMessages(String gatewayId)
	{
		PriorityBlockingQueue<OutboundMessage> queue = queueMap.get(gatewayId);
		if (queue == null) return new ArrayList<OutboundMessage>();
		return new ArrayList<OutboundMessage>(queue);
	}

	@Override
	public int pendingQueueSize(String gatewayId)
	{
		PriorityBlockingQueue<OutboundMessage> queue = queueMap.get(gatewayId);
		if (queue == null) return 0;
		return queue.size();
	}

	@Override
	public boolean removeDelayedMessage(OutboundMessage message)
	{
		for (ScheduledOutboundMessage scheduledOutboundMessage : delayQueue)
		{
			if (message.equals(scheduledOutboundMessage.getMessage()))
			{
				if (delayQueue.remove(scheduledOutboundMessage))
				{
					deleteDelayedMessage(message.getUuid());
					return true;
				}
			}
		}
		return false;
	}

	@Override
	public boolean removeDelayedMessage(String messageUUID)
	{
		for (ScheduledOutboundMessage scheduledOutboundMessage : delayQueue)
		{
			if (messageUUID.equals(scheduledOutboundMessage.getMessage().getUuid()))
			{
				if (delayQueue.remove(scheduledOutboundMessage))
				{
					deleteDelayedMessage(messageUUID);
					return true;
				}
			}
		}
		return false;
	}

	@Override
	public boolean removeAllDelayedMessages()
	{
		delayQueue.clear();
		if (queueDirectory == null) { return true; }
		return emptyDirectory(delayedMessageDir, false);
	}

	@Override
	public boolean removeAllPendingMessages(String gatewayId)
	{
		PriorityBlockingQueue<OutboundMessage> queue = queueMap.get(gatewayId);
		if (queue != null)
		{
			queue.clear();
			queueMap.remove(queue);
			deletePendingMessages(gatewayId);
			return true;
		}
		return false;
	}

	@Override
	public boolean removeAllPendingMessages()
	{
		queueMap.clear();
		deletePendingMessages(null);
		return false;
	}

	private boolean emptyDirectory(File dir, boolean removeDir)
	{
		File[] pendingDirs = dir.listFiles();
		for (File file : pendingDirs)
		{
			if (file.isDirectory())
			{
				emptyDirectory(file, true);
			}
			else if (!file.delete()) { return false; }
		}
		if (removeDir) { return dir.delete(); }
		return true;
	}
}
