/*
 * Created on 20-apr-2004
 */
package convFramework.traffic;

import convFramework.domain.*;
import convFramework.exceptions.*;

import java.util.*;

/**
 * Classe che mantiene una serie di "mailbox" (una per ogni conversazione) in cui inserisce i pacchetti
 * che le vengono passati. E' possible opi recuperare i pacchetti a partire da un id di conversazione.
 * @author Lompa
 */
public class PacketDispatcher {
	
	/* campi */
	private TreeMap mailBoxes = new TreeMap();
	
	/* coda attesa */
	//private Queue queue = new Queue();
	
	
	public PacketDispatcher() throws InstantiationRuleBrokenException{
	}
	
	/**
	 * Inserisce un pacchetto nella mailbox associata alla conversazione cui il pacchetto appartiene;
	 * se questa non esiste la chiamata è bloccante in attesa che la mailbox venga creata.
	 * Se il pacchetto inserito è già arrivato lo si scarta, se è il seguito di un pacchetto non ancora
	 * arrivato lo si insersce in una coda in attesa che arrivi il pacchetto che lo precede.
	 * @param packet il paccheto
	 * @return l'array dei pacchetti inseriti
	 */
	public ContentPacket[] insertPacket(ContentPacket packet) throws LostPacketException{
		MailBox box = null;
		synchronized(this){
			if (!existsMailBox(packet.getConversationID())){
				addMailBox(packet.getConversationID());
			}
			box = getMailBox(packet.getConversationID());
		}
		ContentPacket[] insertedPacket = box.insertPacket(packet);
		//this.queue.qNotify();
		synchronized(this){
			this.notify();
		}
		return insertedPacket;
	}
	
	/**
	 * Restituisce il primo pacchetto preso dalla mailbox associata alla conversazione specificata;
	 * se questa non esiste restituisce null
	 * @param convID l'id della conversazione
	 * @return il pacchetto estratto
	 */
	public ContentPacket extractPacket(IConversationID convID) throws InterruptedException{
		MailBox box = null;
		synchronized(this){
			if (!existsMailBox(convID)){
				//this.queue.qWait();
				this.wait();
			}
			box = getMailBox(convID);
		}
		return box.extractMessage();
	}
	
	/**
	 * Aggiunge una mailBox (se non è già presente)
	 * @param convID l'id della conversazione cui la mailBox è associata
	 */
	public synchronized void addMailBox(IConversationID convID){
		if (!existsMailBox(convID))
			newMailBox(convID);
	}
	
	/**
	 * Eliina una mailbox, se è presente
	 * @param convID id della conversazione cui la mailbox è associata
	 */
	public synchronized void removeMailBox(IConversationID convID){
		if (existsMailBox(convID))
			deleteMailBox(convID);
	}
	
	/* metodi per l'accesso alle mail boxes */
	/* per ottenre una mailbox */
	private synchronized MailBox getMailBox(IConversationID convID){
		return (MailBox) mailBoxes.get(convID.encode());
	}
	/* per aggiungere una mailbox */
	private synchronized MailBox newMailBox(IConversationID convID){
		MailBox box = new MailBox();
		mailBoxes.put(convID.encode(), box);
		return box;
	}
	/* per eliminare una mailbox */
	private synchronized void deleteMailBox(IConversationID convID){
		mailBoxes.remove(convID.encode());
	}
	/* per sapere se una mailbox esiste */
	private synchronized boolean existsMailBox(IConversationID convID){
		return mailBoxes.containsKey(convID.encode());
	}
	
	/* classe mailbox: scarta i pacchetti duplicati e ordina le sequenze di pacchetti */
	private class MailBox{
		
		private LinkedList messages = new LinkedList();
		private Vector arrivedMessagesID = new Vector();
		private TreeMap pendentPackets = new TreeMap(); 
		
		public MailBox(){}
		
		public synchronized ContentPacket extractMessage() throws InterruptedException{
			if (messages.isEmpty()) this.wait();
			return (ContentPacket) messages.removeFirst();
		}
		
		public synchronized ContentPacket[] insertPacket(ContentPacket packet){
			/* se il pacchetto è gà arrivato lo si scarta */
			Vector insertedPackets = new Vector();
			if (!arrivedMessagesID.contains(packet.getPacketID().encode())){
				/* se il pacchetto è una sequenza e non è ancora arrivato il pacchetto di riferimento
				 * lo si sblocca */
				if (packet.isSequence() && packet.getPreviousPacket() != null && !arrivedMessagesID.contains(packet.getPreviousPacket().encode())){
					pendentPackets.put(packet.getPreviousPacket().encode(), packet);
				}
				else {
					/* altrimenti lo si aggiunge*/
					messages.add(packet);
					/* lo si imposta come pacchetto inserito */
					insertedPackets.add(packet);
					/* ... e si controlla se questo pacchetto ne sblocca uno pendente */
					ContentPacket currentPacket = packet;
					while (currentPacket != null){
						if (pendentPackets.containsKey(currentPacket.getPacketID().encode())){
							ContentPacket unblockedPacket = (ContentPacket) pendentPackets.remove(packet.getPacketID().encode());
							messages.add(unblockedPacket);
							insertedPackets.add(unblockedPacket);
							currentPacket = unblockedPacket;
						}
						else {
							currentPacket = null;
						}
					}
					/* si notifica un processo in attesa */
					this.notify();
				}
				/* in ogni caso si aggiorna l'elenco dei pacchetti arrivati */
				arrivedMessagesID.add(packet.getPacketID().encode());
			}
			/* alla fine sis restituisce il pacchetto inserito (che può anche essere null) */
			return (ContentPacket[]) insertedPackets.toArray(new ContentPacket[insertedPackets.size()]);
		}
		
		public synchronized boolean isEmpty(){
			return messages.isEmpty();
		}
		
	}
	
	private class Queue {
		
		public synchronized void qWait() throws InterruptedException{
			this.wait();
		}
		
		public synchronized void qNotify(){
			this.notify();
		}
		
	}
	
}
