package com.beecavegames.messaging;

import java.util.ArrayList;
import java.util.Collection;
import java.util.PriorityQueue;

import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.RequiredArgsConstructor;

import org.joda.time.DateTime;
import org.joda.time.Duration;
import org.joda.time.ReadableDuration;

import com.beecavegames.util.FastRNG;

@RequiredArgsConstructor
public class LocalMQ<T> implements MessageQueue<T> {
	
	private static final ReadableDuration IN_FLIGHT_TIMEOUT = Duration.standardSeconds(30);

	@RequiredArgsConstructor
	@EqualsAndHashCode(of={"id"})
	public class LocalMessage implements Comparable<LocalMessage>, Message<T> {
		long id=FastRNG.nextLong();
		final DateTime until;
		final T message;
		
		
		@Override
		public int compareTo(LocalMessage o) {
			if (until != null) {
				if (o.until == null) {
					return 1;
				} else {
					int d = Long.compare(until.getMillis(), o.until.getMillis());
					if (d != 0) {
						return d;
					}
				}
			} else if (o.until != null) {
				return -1;
			} 
			return Long.compare(id, o.id);
		}


		@Override
		public T get() {
			return message;
		}


		@Override
		public boolean consume() {
			// TODO Auto-generated method stub
			return false;
		}
	}

	

	@Getter
	final String name;
	PriorityQueue<LocalMessage> 
		inFlight = new PriorityQueue<>(), 
		main=new PriorityQueue<>();
	
	@Override
	public boolean sendMessage(T message) {
		synchronized(main) {
			main.add(new LocalMessage(new DateTime(), message));
			main.notifyAll();
			return true;
		}
	}

	@Override
	public Collection<Message<T>> receiveMessages(Duration waitTime, int max) {
		ArrayList<Message<T>> result = new ArrayList<>(max);
		
		if (empty()) {
			synchronized(main) {
				if (empty()) {
					try {
						main.wait(waitTime.getMillis());
					} catch (InterruptedException e) {
					}
				}
			}
		}
		
		while (!empty()) {
			synchronized(main) {
				try {
					DateTime now = new DateTime();
					T elem = null;
					LocalMessage w = main.peek();
					if (w != null && w.until.isBefore(now)) {
						w = main.poll();
						elem = w.message;
					} else {
						w = inFlight.peek();
						if (w != null && w.until.isBefore(now)) {
							w = inFlight.poll();
							elem = w.message;
						}
					}
					
					if (elem != null)
						result.add(new LocalMessage(now.plus(IN_FLIGHT_TIMEOUT), elem));
				} catch (IndexOutOfBoundsException iob) {}
			}
		}
		return result;
	}

	private boolean empty() {
		DateTime now = new DateTime();
		LocalMessage elem = inFlight.peek();
		if (elem != null && elem.until.isBefore(now)) {
			return false;
		}
		
		elem = main.peek();
		if (elem != null && elem.until.isBefore(now)) {
			return false;
		}
		return true;
	}

	@Override
	public boolean consumeAll(Collection<Message<T>> messages) {
		return inFlight.removeAll(messages);
	}

	@Override
	public boolean sendMessage(final T message, Duration delay) {
		main.add(new LocalMessage(new DateTime().plus(delay), message));
		synchronized(main) {
			main.notifyAll();
		}
		return true;
	}

	@Override
	public long size() {
		return main.size();
	}

	@Override
	public long totalSize() {
		return main.size() + inFlight.size();
	}

}

