package io.openvidu.server.kurento.core;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.kurento.client.Continuation;
import org.kurento.client.ErrorEvent;
import org.kurento.client.EventListener;
import org.kurento.client.IceCandidate;
import org.kurento.client.KurentoClient;
import org.kurento.client.MediaPipeline;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.openvidu.client.OpenViduException;
import io.openvidu.client.OpenViduException.Code;
import io.openvidu.java.client.SessionProperties;
import io.openvidu.server.core.Participant;
import io.openvidu.server.core.Session;

/**
 * @author Pablo Fuente (pablofuenteperez@gmail.com)
 */
public class KurentoSession implements Session {

	private final static Logger log = LoggerFactory.getLogger(Session.class);
	public static final int ASYNC_LATCH_TIMEOUT = 30;

	private final ConcurrentMap<String, KurentoParticipant> participants = new ConcurrentHashMap<>();
	private String sessionId;
	private SessionProperties sessionProperties;

	private MediaPipeline pipeline;
	private CountDownLatch pipelineLatch = new CountDownLatch(1);

	private KurentoClient kurentoClient;
	private KurentoSessionEventsHandler kurentoSessionHandler;

	private volatile boolean closed = false;

	private final ConcurrentHashMap<String, String> filterStates = new ConcurrentHashMap<>();
	private AtomicInteger activePublishers = new AtomicInteger(0);

	private Object pipelineCreateLock = new Object();
	private Object pipelineReleaseLock = new Object();
	private volatile boolean pipelineReleased = false;
	private boolean destroyKurentoClient;

	public KurentoSession(String sessionId, SessionProperties sessionProperties, KurentoClient kurentoClient, KurentoSessionEventsHandler kurentoSessionHandler,
			boolean destroyKurentoClient) {
		this.sessionId = sessionId;
		this.sessionProperties = sessionProperties;
		this.kurentoClient = kurentoClient;
		this.destroyKurentoClient = destroyKurentoClient;
		this.kurentoSessionHandler = kurentoSessionHandler;
		log.debug("New SESSION instance with id '{}'", sessionId);
	}

	@Override
	public String getSessionId() {
		return this.sessionId;
	}
	
	@Override
	public SessionProperties getSessionProperties() {
		return this.sessionProperties;
	}

	@Override
	public void join(Participant participant) {
		checkClosed();
		createPipeline();

		KurentoParticipant kurentoParticipant = new KurentoParticipant(participant, this, getPipeline(), kurentoSessionHandler.getInfoHandler());
		participants.put(participant.getParticipantPrivateId(), kurentoParticipant);

		filterStates.forEach((filterId, state) -> {
			log.info("Adding filter {}", filterId);
			kurentoSessionHandler.updateFilter(sessionId, participant, filterId, state);
		});

		log.info("SESSION {}: Added participant {}", sessionId, participant);
	}

	public void newPublisher(Participant participant) {
		registerPublisher();

		// pre-load endpoints to recv video from the new publisher
		for (KurentoParticipant p : participants.values()) {
			if (participant.equals(p)) {
				continue;
			}
			p.getNewOrExistingSubscriber(participant.getParticipantPublicId());
		}

		log.debug("SESSION {}: Virtually subscribed other participants {} to new publisher {}", sessionId,
				participants.values(), participant.getParticipantPublicId());
	}

	public void cancelPublisher(Participant participant) {
		deregisterPublisher();

		// cancel recv video from this publisher
		for (KurentoParticipant subscriber : participants.values()) {
			if (participant.equals(subscriber)) {
				continue;
			}
			subscriber.cancelReceivingMedia(participant.getParticipantPublicId());
			
		}

		log.debug("SESSION {}: Unsubscribed other participants {} from the publisher {}", sessionId, participants.values(),
				participant.getParticipantPublicId());

	}
	
	@Override
	public void leave(String participantPrivateId) throws OpenViduException {

		checkClosed();

		KurentoParticipant participant = participants.get(participantPrivateId);
		if (participant == null) {
			throw new OpenViduException(Code.USER_NOT_FOUND_ERROR_CODE,
					"Participant with private id " + participantPrivateId + " not found in session '" + sessionId + "'");
		}
		participant.releaseAllFilters();

		log.info("PARTICIPANT {}: Leaving session {}", participant.getParticipantPublicId(), this.sessionId);
		if (participant.isStreaming()) {
			this.deregisterPublisher();
		}
		this.removeParticipant(participant);
		participant.close();
	}
	
	@Override
	public Set<Participant> getParticipants() {
		checkClosed();
		return new HashSet<Participant>(this.participants.values());
	}
	
	@Override
	public Participant getParticipantByPrivateId(String participantPrivateId) {
		checkClosed();
		return participants.get(participantPrivateId);
	}
	
	@Override
	public Participant getParticipantByPublicId(String participantPublicId) {
		checkClosed();
		for (Participant p : participants.values()) {
			if (p.getParticipantPublicId().equals(participantPublicId)) {
				return p;
			}
		}
		return null;
	}
	
	@Override
	public boolean close() {
		if (!closed) {

			for (KurentoParticipant participant : participants.values()) {
				participant.releaseAllFilters();
				participant.close();
			}

			participants.clear();

			closePipeline();

			log.debug("Session {} closed", this.sessionId);

			if (destroyKurentoClient) {
				kurentoClient.destroy();
			}

			this.closed = true;
			return true;
		} else {
			log.warn("Closing an already closed session '{}'", this.sessionId);
			return false;
		}
	}

	public void sendIceCandidate(String participantId, String endpointName, IceCandidate candidate) {
		this.kurentoSessionHandler.onIceCandidate(sessionId, participantId, endpointName, candidate);
	}

	public void sendMediaError(String participantId, String description) {
		this.kurentoSessionHandler.onMediaElementError(sessionId, participantId, description);
	}

	@Override
	public boolean isClosed() {
		return closed;
	}

	private void checkClosed() {
		if (isClosed()) {
			throw new OpenViduException(Code.ROOM_CLOSED_ERROR_CODE, "The session '" + sessionId + "' is closed");
		}
	}

	private void removeParticipant(Participant participant) {

		checkClosed();

		participants.remove(participant.getParticipantPrivateId());

		log.debug("SESSION {}: Cancel receiving media from participant '{}' for other participant", this.sessionId, participant.getParticipantPublicId());
		for (KurentoParticipant other : participants.values()) {
			other.cancelReceivingMedia(participant.getParticipantPublicId());
		}
	}

	public int getActivePublishers() {
		return activePublishers.get();
	}

	public void registerPublisher() {
		this.activePublishers.incrementAndGet();
	}

	public void deregisterPublisher() {
		this.activePublishers.decrementAndGet();
	}

	public MediaPipeline getPipeline() {
		try {
			pipelineLatch.await(KurentoSession.ASYNC_LATCH_TIMEOUT, TimeUnit.SECONDS);
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		}
		return this.pipeline;
	}

	private void createPipeline() {
		synchronized (pipelineCreateLock) {
			if (pipeline != null) {
				return;
			}
			log.info("SESSION {}: Creating MediaPipeline", sessionId);
			try {
				kurentoClient.createMediaPipeline(new Continuation<MediaPipeline>() {
					@Override
					public void onSuccess(MediaPipeline result) throws Exception {
						pipeline = result;
						pipelineLatch.countDown();
						log.debug("SESSION {}: Created MediaPipeline", sessionId);
					}

					@Override
					public void onError(Throwable cause) throws Exception {
						pipelineLatch.countDown();
						log.error("SESSION {}: Failed to create MediaPipeline", sessionId, cause);
					}
				});
			} catch (Exception e) {
				log.error("Unable to create media pipeline for session '{}'", sessionId, e);
				pipelineLatch.countDown();
			}
			if (getPipeline() == null) {
				throw new OpenViduException(Code.ROOM_CANNOT_BE_CREATED_ERROR_CODE,
						"Unable to create media pipeline for session '" + sessionId + "'");
			}

			pipeline.addErrorListener(new EventListener<ErrorEvent>() {
				@Override
				public void onEvent(ErrorEvent event) {
					String desc = event.getType() + ": " + event.getDescription() + "(errCode=" + event.getErrorCode()
							+ ")";
					log.warn("SESSION {}: Pipeline error encountered: {}", sessionId, desc);
					kurentoSessionHandler.onPipelineError(sessionId, getParticipants(), desc);
				}
			});
		}
	}

	private void closePipeline() {
		synchronized (pipelineReleaseLock) {
			if (pipeline == null || pipelineReleased) {
				return;
			}
			getPipeline().release(new Continuation<Void>() {

				@Override
				public void onSuccess(Void result) throws Exception {
					log.debug("SESSION {}: Released Pipeline", sessionId);
					pipelineReleased = true;
				}

				@Override
				public void onError(Throwable cause) throws Exception {
					log.warn("SESSION {}: Could not successfully release Pipeline", sessionId, cause);
					pipelineReleased = true;
				}
			});
		}
	}

	public synchronized void updateFilter(String filterId) {
		String state = filterStates.get(filterId);
		String newState = kurentoSessionHandler.getNextFilterState(filterId, state);

		filterStates.put(filterId, newState);

		for (Participant participant : participants.values()) {
			kurentoSessionHandler.updateFilter(this.sessionId, participant, filterId, newState);
		}
	}

}
