content: /*
   *  ProtocolLib - Bukkit server library that allows access to the Minecraft protocol.
   *  Copyright (C) 2012 Kristian S. Stangeland
   *
   *  This program is free software; you can redistribute it and/or modify it under the terms of the 
   *  GNU General Public License as published by the Free Software Foundation; either version 2 of 
   *  the License, or (at your option) any later version.
   *
   *  This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; 
   *  without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. 
   *  See the GNU General Public License for more details.
   *
   *  You should have received a copy of the GNU General Public License along with this program; 
   *  if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 
   *  02111-1307 USA
   */
  
  package com.comphenix.protocol.async;
  
  import java.util.ArrayList;
  import java.util.List;
  import java.util.concurrent.ConcurrentMap;
  import java.util.concurrent.Executor;
  import java.util.concurrent.Executors;
  import java.util.concurrent.ThreadFactory;
  
  import com.comphenix.protocol.PacketType;
  import com.comphenix.protocol.concurrency.ConcurrentPlayerMap;
  import com.comphenix.protocol.error.ErrorReporter;
  import com.comphenix.protocol.events.PacketEvent;
  import com.comphenix.protocol.injector.SortedPacketListenerList;
  import com.google.common.util.concurrent.ThreadFactoryBuilder;
  
  import org.bukkit.entity.Player;
  
  /**
   * Contains every sending queue for every player.
   * 
   * @author Kristian
   */
  class PlayerSendingHandler {
  	private final ErrorReporter reporter;
  	private final ConcurrentMap<Player, QueueContainer> playerSendingQueues;
  	
  	// Timeout listeners
  	private final SortedPacketListenerList serverTimeoutListeners;
  	private final SortedPacketListenerList clientTimeoutListeners;
  	
  	// Asynchronous packet sending
  	private Executor asynchronousSender;
  	
  	// Whether or not we're currently cleaning up
  	private volatile boolean cleaningUp;
  	
  	/**
  	 * Sending queues for a given player.
  	 * 
  	 * @author Kristian
  	 */
  	private class QueueContainer {
  		private final PacketSendingQueue serverQueue;
  		private final PacketSendingQueue clientQueue;
  		
  		public QueueContainer() {
  			// Server packets can be sent concurrently
  			serverQueue = new PacketSendingQueue(false, asynchronousSender) {
  				@Override
  				protected void onPacketTimeout(PacketEvent event) {
  					if (!cleaningUp) {
  						serverTimeoutListeners.invokePacketSending(reporter, event);
  					}
  				}
  			};
  			
  			// Client packets must be synchronized
  			clientQueue = new PacketSendingQueue(true, asynchronousSender) {
  				@Override
  				protected void onPacketTimeout(PacketEvent event) {
  					if (!cleaningUp) {
  						clientTimeoutListeners.invokePacketSending(reporter, event);
  					}
  				}
  			};
  		}
  
  		public PacketSendingQueue getServerQueue() {
  			return serverQueue;
  		}
  
  		public PacketSendingQueue getClientQueue() {
  			return clientQueue;
  		}
  	}
  	
  	/**
  	 * Initialize a packet sending handler.
  	 * @param reporter - error reporter.
  	 * @param serverTimeoutListeners - set of server timeout listeners.
  	 * @param clientTimeoutListeners - set of client timeout listeners.
  	 */
  	public PlayerSendingHandler(ErrorReporter reporter, 
  			SortedPacketListenerList serverTimeoutListeners, SortedPacketListenerList clientTimeoutListeners) {
  		
  		this.reporter = reporter;
  		this.serverTimeoutListeners = serverTimeoutListeners;
  		this.clientTimeoutListeners = clientTimeoutListeners;
  		
  		// Initialize storage of queues
  		this.playerSendingQueues = ConcurrentPlayerMap.usingAddress();
  	}
  	
  	/**
  	 * Start the asynchronous packet sender.
  	 */
  	public synchronized void initializeScheduler() {
  		if (asynchronousSender == null) {
  			ThreadFactory factory = new ThreadFactoryBuilder().
  				setDaemon(true).
  				setNameFormat("ProtocolLib-AsyncSender %s").
  				build();
  			asynchronousSender = Executors.newSingleThreadExecutor(factory);
  		}
  	}
  
  	/**
  	 * Retrieve the sending queue this packet belongs to.
  	 * @param packet - the packet.
  	 * @return The server or client sending queue the packet belongs to.
  	 */
  	public PacketSendingQueue getSendingQueue(PacketEvent packet) {
  		return getSendingQueue(packet, true);
  	}
  	
  	/**
  	 * Retrieve the sending queue this packet belongs to.
  	 * @param packet - the packet.
  	 * @param createNew - if TRUE, create a new queue if it hasn't already been created.
  	 * @return The server or client sending queue the packet belongs to.
  	 */
  	public PacketSendingQueue getSendingQueue(PacketEvent packet, boolean createNew) {
  		QueueContainer queues = playerSendingQueues.get(packet.getPlayer());
  		
  		// Safe concurrent initialization
  		if (queues == null && createNew) {
  			final QueueContainer newContainer = new QueueContainer();
  
  			// Attempt to map the queue
  			queues = playerSendingQueues.putIfAbsent(packet.getPlayer(), newContainer);
  			
  			if (queues == null) {
  				queues = newContainer;
  			}
  		}
  		
  		// Check for NULL again
  		if (queues != null)
  			return packet.isServerPacket() ? queues.getServerQueue() : queues.getClientQueue();
  		else
  			return null;
  	}
  
  	/**
  	 * Send all pending packets.
  	 */
  	public void sendAllPackets() {
  		if (!cleaningUp) {
  			for (QueueContainer queues : playerSendingQueues.values()) {
  				queues.getClientQueue().cleanupAll();
  				queues.getServerQueue().cleanupAll();
  			}
  		}
  	}
  	
  	/**
  	 * Immediately send every server packet with the given list of IDs.
  	 * @param types - types of every packet to send immediately.
  	 * @param synchronusOK - whether we're running on the main thread.
  	 */
  	public void sendServerPackets(List<PacketType> types, boolean synchronusOK) {
  		if (!cleaningUp) {
  			for (QueueContainer queue : playerSendingQueues.values()) {
  				queue.getServerQueue().signalPacketUpdate(types, synchronusOK);
  			}
  		}
  	}
  	
  	/**
  	 * Immediately send every client packet with the given list of IDs.
  	 * @param synchronusOK - whether we're running on the main thread.
  	 */
  	public void sendClientPackets(List<PacketType> types, boolean synchronusOK) {
  		if (!cleaningUp) {
  			for (QueueContainer queue : playerSendingQueues.values()) {
  				queue.getClientQueue().signalPacketUpdate(types, synchronusOK);
  			}
  		}
  	}
  	
  	/**
  	 * Send any outstanding server packets.
  	 * @param onMainThread - whether this is occurring on the main thread.
  	 */
  	public void trySendServerPackets(boolean onMainThread) {
  		for (QueueContainer queue : playerSendingQueues.values()) {
  			queue.getServerQueue().trySendPackets(onMainThread);
  		}
  	}
  	
  	/**
  	 * Send any outstanding server packets.
  	 * @param onMainThread - whether this is occurring on the main thread.
  	 */
  	public void trySendClientPackets(boolean onMainThread) {
  		for (QueueContainer queue : playerSendingQueues.values()) {
  			queue.getClientQueue().trySendPackets(onMainThread);
  		}
  	}
  	
  	/**
  	 * Retrieve every server packet queue for every player.
  	 * @return Every sever packet queue.
  	 */
  	public List<PacketSendingQueue> getServerQueues() {
  		List<PacketSendingQueue> result = new ArrayList<>();
  		
  		for (QueueContainer queue : playerSendingQueues.values())
  			result.add(queue.getServerQueue());
  		return result;
  	}
  	
  	/**
  	 * Retrieve every client packet queue for every player.
  	 * @return Every client packet queue.
  	 */
  	public List<PacketSendingQueue> getClientQueues() {
  		List<PacketSendingQueue> result = new ArrayList<>();
  		
  		for (QueueContainer queue : playerSendingQueues.values())
  			result.add(queue.getClientQueue());
  		return result;
  	}
  	
  	/**
  	 * Send all pending packets and clean up queues.
  	 */
  	public void cleanupAll() {
  		if (!cleaningUp) {
  			cleaningUp = true;
  			
  			sendAllPackets();
  			playerSendingQueues.clear();
  		}
  	}
  
  	/**
  	 * Invoked when a player has just logged out.
  	 * @param player - the player that just logged out.
  	 */
  	public void removePlayer(Player player) {
  		// Every packet will be dropped - there's nothing we can do
  		playerSendingQueues.remove(player);
  	}
  }
  