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

import convFramework.domain.*;
import convFramework.exceptions.*;
import convFramework.traffic.event.*;

/**
 * Ricevitore: va incapsulato in un thread e lancato come daemon
 * @author Lompa
 */
public class Receiver implements Runnable {
	
	/* campi */
	private InGate inGate = null;
	private PacketDispatcher dispatcher = null;
	private ConversationManager convMan = null;
	private ReceptionEventListenerManager eventListenerMan = null;
	private ReTransmissionManager reTransMan = null;
	private AutoAckManager ackMan = null;
	
	/**
	 * Creo una nuova istanza di Receviver 
	 * @param inGate il gate da cui ottenre i pacchetti in ingresso
	 * @param dispatcher il dipatcher in cui riporre i pacchetti ricevuti
	 * @param convMan il gestore delle conversazioni
	 * @param eventListenerMan il manager su cui si registrano gli oggetti che devono essere noditificati degli eventi
	 * @param reTransMan il manager delle ritrasmissioni, per poter scartare gli ack a messaggi non trasmessi
	 */
	public Receiver(InGate inGate, PacketDispatcher dispatcher, ConversationManager convMan, ReceptionEventListenerManager eventListenerMan, ReTransmissionManager reTransMan, AutoAckManager ackMan){
		this.inGate = inGate;
		this.dispatcher = dispatcher;
		this.convMan = convMan;
		this.eventListenerMan = eventListenerMan;
		this.reTransMan = reTransMan;
		this.ackMan = ackMan;
	}
	
	/* (non-Javadoc)
	 * @see java.lang.Runnable#run()
	 */
	public void run() {
		/* il servant è un daemon => esegue a ciclo infinito */
		while(true){
			/* ricezione pacchetto */
			Packet packet = null;
			boolean isAutoAck = false;
			boolean isFalseAck = false;
			
			try {
				//System.out.println("servant - waiting for packets");
				packet = inGate.receive();
				//System.out.println("\nreceiver - packet received");
			}
			catch(MarshallingException e){
				notifyReceptionError("unmarshalling error: "+ e.getMessage() +". packet lost");
			}
			catch(ReceptionErrorException e){
				notifyReceptionError("socket error; packet lost");
			}
			
			/* controllo il tipo del pacchetto */
			if (packet.getType().equals(AutoAckPacket.getTypeDescription())){
				isAutoAck = true;
			}
			/* controllo che non sia una risposta ad un pacchetto mai inviato */
			if (packet.isAck())
				if(!reTransMan.isBeingReTransmitted(packet.getAcknowledgedPacket()))
					isFalseAck = true;
				
			/* se rischiede auto ack invio subito una conferma */
			if (packet.isAutoAckRequired()){
				ackMan.ackPacket(packet);
			}
				
			
			/* ripongo il pacchetto nel dispatcher (se non è un ack automatico) */
			Packet insertedPackets[] = new Packet[0];
			if (!isAutoAck && !isFalseAck){
				try { insertedPackets = dispatcher.insertPacket((ContentPacket) packet); }
				catch(LostPacketException e){
					notifyReceptionError("Packet dispatcher error: "+e.getMessage());
				}
			}
			
			/* notifica di un ack automatico: genera solo eventi di errore di trasmissione o 
			 * di missing response */
			if (isAutoAck){
				/* si informa che è arrivata una risposta */
				notifyResponseArrived(packet);
			}
			
			/* notifica standard, non si attiva per i messaggi di ack automatico */
			for(int i = 0; i< insertedPackets.length; i++){
				/* verifico se il pacchetto inizia una nuova conversazione */
				boolean newConversation = false;
				if (!convMan.existsConversation(packet.getConversationID())){
					/* aggiunta della conversazione e della relativa mailbox */
					this.convMan.addConversation(new StateLessConversation(packet.getConversationID()));
					newConversation = true;
				}
				
				/* controllo se il paccehtto è una risposta o fa parte di una sequenza */
				if (insertedPackets[i].isAck() && insertedPackets[i].getAcknowledgedPacket() == null)
					notifyReceptionError("ack without referred message. packet lost");
				
				/* notifica dell'inizio di una nuova conversazione */
				if (newConversation)
					notifyNewConversation(insertedPackets[i]);
				
				/* verifico s il pacchetto è una risposta */
				if (insertedPackets[i].isAck()){
					notifyResponseArrived(insertedPackets[i]);
				}
				
				/* gestione dell'arrivo del pacchetto */
				notifyPacketArrived(insertedPackets[i]);
			}
		}
	}
	
	/* REAZIONI AD EVENTI */
	/* errore di ricezione: pacchetto perduto */
	private void notifyReceptionError(String message){
		ReceptionErrorEvent event = new ReceptionErrorEvent(this, message);
		
		IReceptionErrorListener[] listeners = this.eventListenerMan.getReceptionErrorListeners();
		
		for(int i = 0; i < listeners.length; i++)
			listeners[i].receptionError(event);
	}
	
	/* ricezione di un pacchetto */
	private void notifyPacketArrived(Packet packet){
		ContentPacketArrivedEvent event = new ContentPacketArrivedEvent(this, packet);
		
		IContentPacketArrivedListener[] listeners = this.eventListenerMan.getContentPacketArrivedListeners();
		
		for(int i = 0; i < listeners.length; i++)
			listeners[i].contentPacketArrived(event);
	}
	
	/* ricezione di una risposta */
	private void notifyResponseArrived(Packet packet){
		AckArrivedEvent event = null;
		try {
			event = new AckArrivedEvent(this, packet);
		}
		catch(BadInvocationArgumentException e){
			throw new CFKError("Error handling response arrival: packet must be a response");
		}
		
		IAckArrivedListener[] listeners = this.eventListenerMan.getAckArrivedListeners();
		
		for(int i = 0; i < listeners.length; i++)
			listeners[i].ackArrived(event);
	}
	
	/* nuova conversazione */
	private void notifyNewConversation(Packet packet){
		/* preparazione dell'evento */
		NewConversationEvent event = new NewConversationEvent(this, packet);
		/* notifica delll'evento */
		INewConversationListener[] listeners = this.eventListenerMan.getNewConversationListeners();
		
		for(int i = 0; i < listeners.length; i++)
			listeners[i].newConversationStarted(event);
	}
	
}
