package example.nocem.livestreaming;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.TreeSet;
import java.util.function.Predicate;

import peersim.cdsim.CDProtocol;
import peersim.config.Configuration;
import peersim.config.FastConfig;
import peersim.core.CommonState;
import peersim.core.Linkable;
import peersim.core.Network;
import peersim.core.Node;
import racoon.rsim.RSProtocol;
import racoon.rsim.accountability.AccountabilitySetting;

/**
 * This class is an implementation of a simple <em>Live Streaming</em> (LS) protocol for P2P systems, 
 * based on a standard gossip-based information dissemination protocol.
 * 
 * In this implementation of the LS, the time is split in rounds. Every round, a special node called
 * the Source creates a set of {@link Chunk} objects, which disseminates to a fanout of other nodes.
 * Also, every round, each node i proposes its unexpired chunks to a fanout of neighbours. With each
 * neighbour j, i establishes a three-phase protocol: (1) i proposes its chunks to j; (2) j requests
 * the chunks that it does not have; (3) i delivers the requested chunks to j.
 * 
 * @author Guido Lena Cota
 */
public class LiveStreaming extends RSProtocol implements CDProtocol {
	
	// CONSTANTS
	// ------------------------------------------------------------------------------------------------
	
	// TODO teoria: funziona meglio perché riesce ad arrivare a più peers, quindi 
	// diffonde meglio i chunks
	
	/**
	 * The round at which starts the monitoring.
	 */
	public static final int START_MONITORING = 600; // 10 minutes
	
	/**
	 * The number of gossip protocol executions per second (i.e., 1 cycle).
	 * Default value [5, which corresponds to 200 ms] is taken from "Stretching Gossip with Live Streaming" 
	 * by Frey et al.
	 */
	private static final byte GOSSIP_PER_SEC = 1;
	
	/** 
	 * The number of {@link Chunk}s sent every 1 second of video streaming. 
	 * In our setting, a chunk includes 5 video-frames. Considering that the
	 * default value of FPS is 25, then the default number of chunks-per-second
	 * is 5.
	 */
	public static final byte CPS = 5;

	/** 
	 * The bits of video that can be transmitted in a second, in bps. 
	 * The default value is taken from the YouTube recommandations for upload encoding which can be found
	 * here: https://support.google.com/youtube/answer/1722171 , https://support.google.com/youtube/answer/2853702 .
	 * The recommended video bit rate for a video with resolution 480 p (854x480) is between 500 Kbps and 2000 Kbps.
	 * Default value: 2 Mbps.
	 * (value in "Stretching Gossip with Live Streaming": 600 Kbps)
	 */
	private static final int BITRATE = 2000000;
	
	/** 
	 * The size of a video chunk, in bytes.
	 * As in "Experimental comparison of neighborhood filtering strategies in unstructured P2P-TV systems"
	 * by Traverso et al., we assume that a video chunk corresponds to a singe video frame.
	 * The CHUNK_SIZE value can be thus derived from the {@link BITRATE} and {@link FPS} values.
	 */
	private static final int CHUNK_SIZE = BITRATE / CPS / 8;
	
	/**	
	 * The size of the video stream, stored as number of chunks, and calculated as:
	 * ["satisfaction_deadline"] * {@link FPS}.
	 */
	public static final int TOT_CHUNKS_NUMBER = (int) (CommonState.getEndTime() * CPS);
	
	// Message types
	
	/** 
	 * Specifies a STREAMING message.
	 */
	private static final byte MSG_STREAMING = 0;

	/** 
	 * Specifies a PROPOSE message.
	 */
	private static final byte MSG_PROPOSE = 1;
	
	/** 
	 * Specifies a REQUEST message.
	 */
	private static final byte MSG_REQUEST = 2;
	
	/** 
	 * Specifies a SERVE message.
	 */
	private static final byte MSG_SERVE = 3;
	
	// PARAMETERS
	// ------------------------------------------------------------------------------------------

	/**
	 * The fraction of peers in the network that are directly connected with the video source.
	 */
	private static final String PAR_SOURCE_FANOUT = "fanout_source";

	/** 
	 * The playback deadline, i.e., the maximum allowed time (in cycles) between the creation of a 
	 * video chunk and its playback. Chunks received after their playback deadline are discarded.
	 */
	private static final String PAR_DEADLINE = "deadline";
	

	// MESSAGES
	// ------------------------------------------------------------------------------------------------
		
	/**
	 * The instance used for all the SHARING messages.
	 */
	private Message sharingMessage;

	/**
	 * The instance used for all the PROPOSE messages.
	 */
	private Message proposeMessage;
	
	/**
	 * The instance used for all the REQUEST messages.
	 */
	private Message requestMessage;

	/**
	 * The instance used for all the SERVE messages.
	 */
	private Message serveMessage;
	
	// FIELDS
	// ------------------------------------------------------------------------------------------
    	
	/** 
	 * The playback deadline of media {@link Chunk}s, as specified in {@link PAR_DEADLINE}.
	 * The default value is taken from "Experimental comparison of neighborhood filtering strategies
	 * in unstructured P2P-TV systems" by Traverso et al.
	 */
	private final byte deadline;
	
	/* Attributes of the source node */
	
	/** 
	 * The fraction of the network connected with the source, as specified in {@link PAR_SOURCE_FANOUT}.
	 * The default value is taken from "BAR Gossip" by Li et al.
	 */
	private final double source_fanout;
	
	/* Attributes of all nodes */
	
	/**	
	 * The size of the media stream, stored as number of {@link Chunk}, and set by {@link StreamingInit}.
	 */
	private int nChunks;
	
	/**
	 * The partners period.
	 */
	private int partnerPeriod;
	
	/**
	 * The set of unexpired {@link Chunk}s that can be exchanged by peers.
	 */
	private TreeSet<Chunk> transmissibleChunks;
	
	/** 
	 * The number of communication partners, as specified in {@link PAR_FANOUT}. 
	 * The default value is taken from "Probabilistic Reliable Dissemination in Large-Scale Systems" 
	 * by Kermarrec et al.
	 */
	private final byte fanout;
	
	/**
	 * The set of communication partners.
	 */
	private List<LiveStreaming> partners;
	
	/** 
	 * The id of the linkable protocol that defines the overlay network. 
	 */
	private static byte linkableId;
	
	/**
	 * The per-cycle upload bandwidth capacity of the local node, set by {@link StreamingInit}
	 */
	private long uploadBandwidthCapacity;

	/**
	 * The upload bandwidth utilisation in the current cycle.
	 */
	private long uploadBandwidthLeft;
	
	/**
	 * True if this node is the source node.
	 */
	private boolean isSource;
	
	// SUPPORT VARIABLES
	// ------------------------------------------------------------------------------------------------
	
	/**	
	 * Support lists for randomising the communication partners selection. 
	 */
	// the size of the list below depends on the neighbours degree assigned by the overlay protocol
	private static List<Short> peer_potentialPartnersIDs;
	
	// INITIALIZATION
	// ------------------------------------------------------------------------------------------
   
    /** 
     * Standard constructor invoked by the simulation engine.
     * @param prefix: the configuration prefix for this class. */
    public LiveStreaming(String prefix) {
    	super(prefix);
    	
    	deadline = (byte) Configuration.getInt(prefix + "." + PAR_DEADLINE, 6);
    	transmissibleChunks = new TreeSet<Chunk>(new ChunkById());
    	partners = new ArrayList<LiveStreaming>();

    	/* source's parameters */
    	source_fanout = Configuration.getDouble(prefix + "." + PAR_SOURCE_FANOUT, 0.1);
    	
    	/* peers' parameters */
    	fanout = (byte) (Math.round(Math.log(Network.size())) + (byte) (Math.round(Math.log(Network.size()) * 1.2)));
		linkableId = (byte) FastConfig.getLinkable(CommonState.getPid());
    }
    
    @Override
	public LiveStreaming clone(){
    	LiveStreaming ls = null;
    	ls = (LiveStreaming) super.clone(); 
    	ls.partners = new ArrayList<LiveStreaming>();
    	ls.transmissibleChunks = new TreeSet<Chunk>(new ChunkById());
    	return ls;
    }
    
    // METHODS implementing the live streaming functionalities
 	// ------------------------------------------------------------------------------------------------
 	
 	/**
     * Implement the live streaming protocol.
     * 
     * @param node: 		the node on which this component is run.
     * @param protocolID: 	the id of this protocol in the protocol array.
     */
    @Override
	public void nextCycle(Node node, int protocolID) {
    	cyclesSinceLastUpdate++;
    	long currentCycle = CommonState.getTime();
    	
    	// reset the bandwidth left
    	if(currentCycle > 0 && !isSource)
    	
		uploadBandwidthLeft = uploadBandwidthCapacity;
		
		/* the node is the source */
		if(isSource){
			if (currentCycle > 0 && currentCycle % partnerPeriod == 0)
				selectSourcePartners(protocolID);
			
			doStreaming();
			return;
		}
		/* if the node is a peer */
		else if(currentCycle > 0){
			if (currentCycle % partnerPeriod == 0)
				selectPartners(protocolID);
			
			// update the set of transmissible chunks
			updateTransmissibleChunks();
			// execute the gossip protocol
			startProtocol();
		}
    }
    
    /* SOURCE METHODS */
    
    /**
	 * Simulate the streaming of the {@link StreamingInit.FPS} chunks during the current cycle.
	 */
	private void doStreaming(){
		// get the set of chunks to stream at this cycle
		TreeSet<Chunk> chunksToStream = getStreamableChunks(transmissibleChunks, (int) CommonState.getTime());
		
		for(LiveStreaming partner : partners){
			// create the message
			sharingMessage = new Message(this, MSG_STREAMING, chunksToStream);
			// send a message to the selected peer with the set of chunks to stream
			sendMessage(sharingMessage, partner);
		}
		chunksToStream = null;
	}
	
	/* PEERS METHODS */
    
	/**
	 * Simulate the execution of the gossip protocol for {@value StreamingControl.GOSSIP_PER_SEC} times
	 * at every simulation cycle.
	 */
	private void startProtocol(){
		stopGossip:
		for(int i = 0; i < GOSSIP_PER_SEC; i++)
			// send a PROPOSE message to each communication partner
			if(!doPropose(fanout))
				break stopGossip;
	}
	
	/**
     * Implement the Propose transition of the LS protocol, invoked by the responder role.
     */
    private boolean doPropose(int fanout){
    	// skip if the node doesn't have any chunk to share
		if(transmissibleChunks.isEmpty())
			return false;
    			
    	boolean result = false;
    	switch(getDeviationType("Propose")){
    		case NONE:
    			result = doPropose_legal(fanout);
    			break;
    		case TIMEOUT: 
    			break;
//    		case SUBSET:
//    			result = doPropose_subset(node,protocolID);
//    			break;
    		case MULTICAST:
    			result = doPropose_multicast(fanout);
    			break;
			default: 
    	}
    	
    	return result;
    }
	
	/**
	 * Simulate the preparation and sending of a PROPOSE message to the communication partners.
	 * 
	 * @param fanout the number of communication partner of the peer
	 * @return true if the peer has still some upload bandwidth left for another gossip process
	 */
	private boolean doPropose_legal(int fanout){
		return propose(fanout, true);
	}
	
	/**
	 * Multicast deviation from the Propose transition of the LS protocol.
	 */
	private boolean doPropose_multicast(int fanout){
		return propose(fanout, false);
	}
	
	private boolean propose(int fanout, boolean isLegal){
		TreeSet<Chunk> chunksToPropose = new TreeSet<Chunk>(transmissibleChunks);
		Collections.shuffle(partners);
		
		int nPropose = 0;
		int cont = 0;
		// for each communication partner
		while(nPropose < fanout && cont < partners.size()){
			LiveStreaming p = partners.get(cont);
			if((isLegal) || 
				(!isLegal && CommonState.r.nextDouble() < getDeviationDegree("Propose"))){
				/* Prepare the message to send to the partner:
				 * - message value: the set of completed chunks that can be still played out
				 */
				// check how many chunks are virtually transmissible
				int nTransmissibleChunks = howManyTransmissibleChunks(chunksToPropose);
				// if no chunks can be transmitted, quit the propose step for this cycle
				if(nTransmissibleChunks == 0)
					return false;
				
				// create the message
				proposeMessage = new Message(this, MSG_PROPOSE, chunksToPropose, nTransmissibleChunks);
				// send a message to the selected peer with the set of chunks to propose
				sendMessage(proposeMessage, p);
				
				processRMessage("g0", chunksToPropose.size(), this, "r0", p, "R1");
				processRTransition("Propose", this, "r0", p, "R1");
				
				nPropose++;
			}
			cont++;
		}
		chunksToPropose = null;
		return true;
	}
	
	/**
	 * Simulate the preparation and sending of a REQUEST message.
	 * 
	 * @param msg the PROPOSE message received from the communication partner
	 */
	private void doRequest(Message msg){
		/* Prepare the message to send to the partner:
		 * - message value: the set of chunks to require among the one proposed
		 */
		// calculate the chunks to request among the ones that were proposed in the PROPOSE message
		TreeSet<Chunk> chunksToRequest = new TreeSet<Chunk>(msg.getValue());
		chunksToRequest.removeAll(transmissibleChunks);
		
		int sizeRequestMessage = (chunksToRequest.size() * Integer.SIZE) + AccountabilitySetting.MESSAGE_HEADER_BYTES;
		
		
		if(chunksToRequest.isEmpty() || uploadBandwidthLeft < sizeRequestMessage)
			return;
		
		if(chunksToRequest.size() > msg.getMaxTransmissibleChunks()){
			// if there is no enough bandwidth left to send the whole set of chunks
			int chunksToRemove = chunksToRequest.size() - msg.getMaxTransmissibleChunks();
			Iterator<Chunk> iter = chunksToRequest.descendingIterator();
			while(chunksToRemove > 0 && iter.hasNext()){
				iter.next();
				iter.remove();
				chunksToRemove--;
			}
		}

		requestMessage = new Message(this, MSG_REQUEST, chunksToRequest);
		// send a message to the selected peer with the set of chunks to request
		sendMessage(requestMessage, msg.getSender());
		
		processRMessage("g1", chunksToRequest.size(), this, "R1", msg.getSender(), "r0");
		
		chunksToRequest = null;
	}
	
	
	/**
     * Implement the Serve transition of the LS protocol, invoked by the responder role.
     */
    private void doServe(Message msg, int nChunksToServe){
    	switch(getDeviationType("Serve")){
    		case NONE:
    			serve(msg, nChunksToServe);
    			break;
    		case TIMEOUT: 
    			break;
    		case SUBSET:
    			// calculate the chunks to serve among the ones that were requested in the REQUEST message
    			nChunksToServe = (int) (nChunksToServe * getDeviationDegree("Serve"));
    			serve(msg, nChunksToServe);
    			break;
//    		case MULTICAST:
//    			doServe_multicast(msg, nChunksToServe);
//    			break;
			default: 
    	}
	}
	
	/**
	 * Simulate the preparation and sending of a SERVE message.
	 * 
	 * @param msg the REQUEST message received from the communication partner
	 * @param nChunksToServe the number of chunks that are requested in the REQUEST message
	 */
	private void serve(Message msg, int nChunksToServe){
		/* Prepare the message to send to the partner:
		 * - message value: the set of chunks to serve
		 */
		// calculate the chunks to serve among the ones that were requested in the REQUEST message
		TreeSet<Chunk> chunksToServe = new TreeSet<Chunk>(msg.getValue());
		int chunksToRemove = chunksToServe.size() - nChunksToServe;
		Iterator<Chunk> iter = chunksToServe.descendingIterator();
		while(chunksToRemove > 0 && iter.hasNext()){
			iter.next();
			iter.remove();
			chunksToRemove--;
		}
		
		serveMessage = new Message(this, MSG_SERVE, chunksToServe);
		// send a message to the selected peer with the set of chunks to request
		sendMessage(serveMessage, msg.getSender());
		
		processRMessage("g2", chunksToServe.size(), this, "r0", msg.getSender(), "R1");
		
		chunksToServe = null;
	}
	
	/* MESSAGE HANDLING */
	
	/**
	 * Simulate the sending of a {@link Message} to a communication partner, as well as the processing of
	 * the message by the recipient.
	 * 
	 * @param msg the message to send
	 * @param recipient the communication partner 
	 */
	private void sendMessage(Message msg, LiveStreaming recipient){
		// update the bandwidth that is left, according to the message type
		switch(msg.getType()){
			case MSG_STREAMING:
			{	
				/* sending */
				// send a set of Chunk objects
				uploadBandwidthLeft -= AccountabilitySetting.MESSAGE_HEADER_BYTES + msg.getValue().size() * CHUNK_SIZE;
				/* processing */
				// update the set of completed chunks
				if(CommonState.getTime() > START_MONITORING) recipient.nChunks += msg.getValue().size();
				recipient.transmissibleChunks.addAll(msg.getValue());
				
				break;
			}
			case MSG_PROPOSE: 
			{
				/* sending */
				// send a set of integers (identifiers of the Chunk objects to propose)
				// a primitive field in Java has size of 32 bits
				uploadBandwidthLeft -= AccountabilitySetting.MESSAGE_HEADER_BYTES + msg.getValue().size() * Integer.SIZE;
				/* processing */
				// prepare a request message
				recipient.doRequest(msg);
				break;
			}
			case MSG_REQUEST:
			{
				/* sending */
				// send a set of integers (identifiers of the Chunk objects to request)
				uploadBandwidthLeft -= AccountabilitySetting.MESSAGE_HEADER_BYTES + msg.getValue().size() * Integer.SIZE;
				/* processing */
				// prepare a serve message
				recipient.doServe(msg, msg.getValue().size());
				break;
			}
			case MSG_SERVE:
			{
				/* sending */
				// send a set of Chunk objects
				uploadBandwidthLeft -= AccountabilitySetting.MESSAGE_HEADER_BYTES + msg.getValue().size() * CHUNK_SIZE;
				/* processing */
				// update the set of completed chunks
				if(CommonState.getTime() > START_MONITORING) recipient.nChunks += msg.getValue().size();
				recipient.transmissibleChunks.addAll(msg.getValue());
				break;
			}
			default:
		}
	}
	
	// METHODS for supporting the live streaming functionalities
	// ------------------------------------------------------------------------------------------------
	
	/* INITALISE THE SOURCE AND THE PEERS */
	
	/**
	 * Initialise the source of the video streaming (by default, the node with ID = 0). 
	 * In particular, this method creates the video source's chunks.
	 */
	public void initialiseSource(long uploadBandwidthCapacity, int partnerPeriod){
		isSource = true;
		
		this.nChunks = TOT_CHUNKS_NUMBER;
		this.uploadBandwidthCapacity = uploadBandwidthCapacity;
		this.partnerPeriod = partnerPeriod;
		
		int currentCycle = 0;
		// create the video chunks
		for(int i = 0 ; i < nChunks ; i++) {
			transmissibleChunks.add(new Chunk(i, currentCycle));
			// update the currenyCycle value every FPS chunks
			if((i+1) % CPS == 0)
				currentCycle++;
		}
	}
	
	/**
	 * Initialise the peer (any node with ID different from 0). 
	 */
	public void initialisePeer(long uploadBandwidthCapacity, int partnerPeriod){
		isSource = false;
		
		this.nChunks = 0;
		this.uploadBandwidthCapacity = uploadBandwidthCapacity;
		this.partnerPeriod = partnerPeriod;
	}
	
	/* UPDATE THE PARTNERS OF SOURCE AND PEERS */
	
	/**
	 * Randomly select a number of {@link fanout} partners from the neighbourhood.
	 */
	public void selectPartners(int protocolId){
		Linkable neighbours = (Linkable) CommonState.getNode().getProtocol(linkableId);
		List<LiveStreaming> oldPartners = (!partners.isEmpty()) ? new ArrayList<LiveStreaming>(partners) : null;
		
		// reset the set of communication partners
		int ndegree = neighbours.degree();
		partners = new ArrayList<LiveStreaming>(ndegree);
		
		// shuffle the peers' support variable for randomising the partner selection
		Collections.shuffle(peer_potentialPartnersIDs, CommonState.r);
		
		// number of partners from the random network
		int partnersNw = Math.max(1, (int) (Math.round(fanout * 0.2)));
		
		int i = 0;
		while (partners.size() < (ndegree - partnersNw) && i < peer_potentialPartnersIDs.size())	{
			// pick a random neighbour
			Node n = neighbours.getNeighbor(peer_potentialPartnersIDs.get(i));
			// get the LiveStreaming instance
			LiveStreaming gp = (LiveStreaming) n.getProtocol(protocolId);

			// if gp is the source, or this node, or an old partner, then skip to the next partner
			if(gp.isSource || gp == this || (oldPartners != null && oldPartners.contains(gp))) {
        		i++;
        		continue;
			}
		
//			if(ProtocolEGT.checkStartInteraction(gp))
				partners.add(gp);
			i++;
		}
		
		int j = 0;
		int nwSize = Network.size();
		while (partners.size() < ndegree && j < nwSize){
			// pick a random neighbour from the network
			Node n = Network.get(CommonState.r.nextInt(nwSize));
			// get the LiveStreaming instance
			LiveStreaming gp = (LiveStreaming) n.getProtocol(protocolId);
			
			// if gp is the source, or this node, or an old partner, or is already a partner, then consider another node
			if(gp.isSource || gp == this || (oldPartners != null && oldPartners.contains(gp)) || partners.contains(gp)) {	
				j++;
        		continue;
			}
			
//			if(ProtocolEGT.checkStartInteraction(gp))
				partners.add(gp);
			j++;
		}
		
		oldPartners = null;
	}
	
	/**
	 * Randomly select a number of {@link source_fanout} of peers for the source node.
	 */
	public void selectSourcePartners(int protocolId){
		final int sourcePartnersSize = (int) (source_fanout * (Network.size() - 1));
		// reset the set of communication partners
		partners = new ArrayList<LiveStreaming>(sourcePartnersSize);
		int partnersSize = 0;
		int nwSize = Network.size();
		while (partnersSize < sourcePartnersSize){
			// pick a random neighbour
			Node n = Network.get(CommonState.r.nextInt(nwSize));
			
			// get the LiveStreaming instance
			LiveStreaming gp = (LiveStreaming) n.getProtocol(protocolId);
			if(gp.isSource || partners.contains(gp)) continue; // check if n is the source node or if the partner was already added
			
			partners.add(gp);
			partnersSize++;
		}
	}
	
	/**
	 * Support lists for randomising the selection of the source and peers' communication partners. 
	 */
	public static void initialisePotentialPartnersSupports(){
		int overlayNeighboursSize = ((Linkable)Network.get(0).getProtocol(linkableId)).degree();
		peer_potentialPartnersIDs = new ArrayList<Short>(overlayNeighboursSize);
		
		for(short i = 1; i < overlayNeighboursSize; i++){
			peer_potentialPartnersIDs.add(i);
		}
	}
	
	/* CHUNKS RETRIEVAL */

	private int howManyTransmissibleChunks(Set<Chunk> chunksSet){
		int result = 0;
		long sizeProposeMessage = AccountabilitySetting.MESSAGE_HEADER_BYTES + (chunksSet.size() * Integer.SIZE);
		result = Math.min(chunksSet.size(), (int) ((uploadBandwidthLeft - sizeProposeMessage - AccountabilitySetting.MESSAGE_HEADER_BYTES) / CHUNK_SIZE));
		
		if(result > 3){
			int perPartnerTChks = (int) (uploadBandwidthLeft * 0.6 - sizeProposeMessage - AccountabilitySetting.MESSAGE_HEADER_BYTES) / (CHUNK_SIZE);
			result = Math.min(perPartnerTChks, result);
		}
		return (result < 0) ? 0 : result;
	}
	
	/**
	 * Return the set of {@link Chunk}s that satisfy the passed condition.
	 */
	private TreeSet<Chunk> getStreamableChunks(TreeSet<Chunk> chunksSet, int value){
		TreeSet<Chunk> result = new TreeSet<Chunk>(chunksSet);
		Predicate<Chunk> predicate = new Predicate<Chunk>() {

			@Override
			public boolean test(Chunk c) {
				return c.creationTime != value;
			}
		};
		
		result.removeIf(predicate);
		transmissibleChunks.removeAll(result);
		return result;
	}
	
	/**
	 * Update the set of {@link Chunk}s that can be exchanged (i.e., the not expired yet).
	 */
	private void updateTransmissibleChunks(){
		/* get the set of chunks to propose at this cycle, i.e., the set of completed chunks
		 * that can be still played out */
		int lastValidCreationTime = Math.max(0, (int) CommonState.getTime() - deadline);
		
		Predicate<Chunk> predicate = new Predicate<Chunk>() {
			@Override
			public boolean test(Chunk c) {
				return c.creationTime < lastValidCreationTime;
			}
		};
		
		transmissibleChunks.removeIf(predicate);
	}
	
	/**
	 * Return true if this node is the source.
	 */
	public boolean isSource(){
		return isSource;
	}
	
	/* MONITOR */

	public long getCompletedChunks(){
		return nChunks;
	}
}

/**
 * Define attributes and operations for menaging the message exchange in the gossip protocol.
 */
class Message {
	
	/** 
	 * The sender of the message.
	 */
	private final LiveStreaming sender;
	
	/**
	 * The type of the message, with reference to the codes written in the GossipProtocol class.
	 */
	private final int type;
	
	/**
	 * The data value of the message.
	 */
	private TreeSet<Chunk> value;
	
	/**
	 * In PROPOSE messages, specifies the max number of chunks that can be actually transmitted.
	 */
	private final int maxTransmissibleChunks;
	
	public Message(LiveStreaming sender, int type, TreeSet<Chunk> value, int maxTransmissibleChunks){
		this.sender = sender;
		this.type = type;
		this.value = value;
		this.maxTransmissibleChunks = maxTransmissibleChunks;
	}

	public Message(LiveStreaming sender, int type, TreeSet<Chunk> value){
		this.sender = sender;
		this.type = type;
		this.value = value;
		this.maxTransmissibleChunks = value.size();
	}

	public LiveStreaming getSender() {
		return sender;
	}

	public int getType() {
		return type;
	}
	
	public TreeSet<Chunk> getValue() {
		return value;
	}
	
	public int getMaxTransmissibleChunks() {
		return maxTransmissibleChunks;
	}
}
