QueueBean.java

package fi.codecrew.moya.beans;

import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicLong;

import javax.annotation.security.DeclareRoles;
import javax.annotation.security.RolesAllowed;
import javax.ejb.Asynchronous;
import javax.ejb.EJB;
import javax.ejb.LocalBean;
import javax.ejb.Lock;
import javax.ejb.LockType;
import javax.ejb.Singleton;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import fi.codecrew.moya.enums.apps.MapPermission;
import fi.codecrew.moya.enums.apps.UserPermission;
import fi.codecrew.moya.facade.PlaceSlotFacade;
import fi.codecrew.moya.model.EventMap;
import fi.codecrew.moya.model.EventUser;
import fi.codecrew.moya.model.LanEventProperty;
import fi.codecrew.moya.model.LanEventPropertyKey;
import fi.codecrew.moya.model.PlaceSlot;
import fi.codecrew.moya.model.map.MapReservationQueueEntry;

/**
 * Session Bean implementation class QueueBean
 */
@Singleton
@LocalBean
@Lock(LockType.READ)
@DeclareRoles({ MapPermission.S_MANAGE_MAPS })
public class QueueBean implements QueueBeanLocal {
	private static final Logger logger = LoggerFactory.getLogger(QueueBean.class);

	/**
	 * Default constructor.
	 */
	public QueueBean() {
		logger.info("Initialized2 QueueBean, {}", mapqueues);
	}

	@EJB
	private PlaceBeanLocal placebean;

	private final ConcurrentHashMap<Integer, MapQueue> mapqueues = new ConcurrentHashMap<Integer, QueueBean.MapQueue>();
	private int defaultTimeoutMin = 10;
	private int minimumSlotsInQueue = 9999;
	private int reservingSize = 1;

	private class MapQueue implements MapQueueI {
		// private final Set<MapReservationQueueEntry> reserving = new HashSet<>();
		private final Set<EventUser> reserving = Collections.newSetFromMap(new ConcurrentHashMap<EventUser, Boolean>());
		private final ConcurrentSkipListSet<MapReservationQueueEntry> queue = new ConcurrentSkipListSet<>();
		private final ConcurrentMap<EventUser, MapReservationQueueEntry> queEntries = new ConcurrentHashMap<>();

		private void timeoutEntries() {
			logger.info("Timeouting entries");
			// give 10 seconds mercy ( and give us some time to go through all entries)
			Date now = new Date(System.currentTimeMillis() + 1000 * 15);
			for (EventUser r : reserving) {
				MapReservationQueueEntry entry = queEntries.get(r);
				logger.info("Checking if should remove user from queue {}, timeout {}", r, entry.getReservationTimeout());
				if (entry.getReservationTimeout() == null || now.after(entry.getReservationTimeout())) {
					logger.info("Removing Eventuser {} from reserving queue due to reservationTimeout: {}", r, entry.getReservationTimeout());
					this.remove(r);
				}
			}

			// Set idle time to the past.
			// Idle timeout after 60 seconds
			Date idleTimeout = new Date(System.currentTimeMillis() - 1000 * 60);
			for (MapReservationQueueEntry entry : queue) {

				if (entry.getSeenTime() == null) {
					entry.setSeenTime(new Date());
					continue;
				}
				if (idleTimeout.after(entry.getSeenTime())) {
					remove(entry.getUser());
				}

			}
		}

		public boolean isReserving(EventUser e) {

			// Check queue size and add entry to queue
			checkReservingEntry();
			MapReservationQueueEntry que = queEntries.get(e);
			if (que != null) {
				que.setSeenTime(new Date());
			}
			return reserving.contains(e);

		}

		private void checkReservingEntry() {
			if (reserving.size() < reservingSize) {
				synchronized (queue) {
					if (reserving.size() < reservingSize) {
						MapReservationQueueEntry queEntry = queue.pollLast();
						if (queEntry != null) {
							reserving.add(queEntry.getUser());
							queEntry.setReservationTimeout(new Date(System.currentTimeMillis() + defaultTimeoutMin * 60 * 1000));
						}
					}
				}
			}
		}

		public MapReservationQueueEntry remove(EventUser user)
		{
			if (user == null) {
				return null;
			}
			MapReservationQueueEntry ret = null;

			synchronized (queue) {
				if (reserving.remove(user)) {
					logger.info("Removed user {} from reserving queue", user);
				}
				MapReservationQueueEntry entry = queEntries.get(user);
				// There should neve be more than one instance, but make sure
				while (queue.remove(entry)) {
					logger.info("Removed user {} from queue");
				}
				ret = queEntries.remove(user);
				checkReservingEntry();
			}
			return ret;
		}

		public MapReservationQueueEntry enter(EventUser user) {
			MapReservationQueueEntry ret = queEntries.get(user);
			synchronized (queue) {
				if (!reserving.contains(user) && (ret == null || !queue.contains(ret))) {

					if (ret == null) {
						ret = new MapReservationQueueEntry();
						ret.setUser(user);
						ret.setSeenTime(new Date());
						ret.setPlaceslotcount(placebean.getPlaceslots(user).size());
					}
					queEntries.put(user, ret);
					boolean queStat = queue.add(ret);
					logger.info("User {} not in queue, offer state {}", user, queStat);

					// Check if the user can be put to reservation queue immediately
					checkReservingEntry();

				} else {
					ret = queEntries.get(user);
					logger.info("User {} already in queue. Not entering again {}", user, ret);
				}
			}
			return ret;

		}

		public void forceAdd(EventUser u, Date time) {
			MapReservationQueueEntry entry = queEntries.get(u);
			synchronized (queue)
			{
				if (entry == null) {
					entry = new MapReservationQueueEntry();
					entry.setCreated(new Date());
					entry.setUser(u);
					entry.setPlaceslotcount(placebean.getPlaceslots(u).size());
					queEntries.put(u, entry);
				}

				entry.setReservationTimeout(time);
				entry.setSeenTime(new Date());
				reserving.remove(u);

				queue.remove(entry);
				reserving.add(u);

			}

		}

		public Integer getPosition(EventUser user) {
			Integer ret = null;
			MapReservationQueueEntry entry = queEntries.get(user);
			if (reserving.contains(user)) {
				ret = 0;
				logger.info("User in reserving queue {}", user);
			} else if (entry != null && queue.contains(entry)) {
				ret = 1;
				for (Iterator<MapReservationQueueEntry> iterator = queue.descendingIterator(); iterator.hasNext();) {
					MapReservationQueueEntry eu = iterator.next();
					if (eu.getUser().equals(user)) {
						break;
					}
					++ret;
				}
				logger.info("User is in queue {}, position {}", user, ret);
			} else {
				logger.info("Not in queue, while checking position");
			}
			logger.info("Got position {} for user {}", ret, user);
			return ret;
		}

		public boolean isInQueue(EventUser user) {
			MapReservationQueueEntry entry = queEntries.get(user);
			return reserving.contains(user) || (entry != null && queue.contains(entry));
		}

		public MapReservationQueueEntry getEntry(EventUser user) {
			return queEntries.get(user);
		}

		public Set<EventUser> getReserving() {
			return reserving;
		}

		@Override
		public ConcurrentSkipListSet<MapReservationQueueEntry> getQueue() {
			return queue;
		}

	}

	@EJB
	private EventBeanLocal eventbean;
	private AtomicLong nextReservingTimeoutCheck = new AtomicLong();
	@EJB
	private PlaceSlotFacade slotfacade;
	@EJB
	private PermissionBeanLocal permbean;

	@Lock(LockType.READ)
	@Override
	public boolean isReserving(EventMap map, EventUser user) {
		// If queue is not enabled, user can always reserve

		if (!isQueueEnabled())
			return true;
		if (!permbean.isCurrentUser(user) && !permbean.hasPermission(MapPermission.MANAGE_OTHERS))
			throw new RuntimeException("Use " + permbean.getCurrentUser() + "tried to enter to queue in behalf of another user: " + user);

		if (map == null || user == null) {
			logger.warn("Can not check map {}, user {}", map, user);
			return false;
		}
		boolean ret = getMapque(map).isReserving(user);

		// Do some housekeeping, but only on each 120

		long now = System.currentTimeMillis();
		long nextTime = nextReservingTimeoutCheck.get();
		// Update next checktime to 120 seconds in to the future, so we should have plenty of time
		// to do the checks we need..
		if (now > nextTime && nextReservingTimeoutCheck.compareAndSet(nextTime, now + 1000 * 120)) {
			logger.info("Launcing reservingTimeout check ");
			checkReservingTimeouts();
			logger.info("Done launching reservingTimeoutCheck");
		}

		return ret;

	}

	@Lock(LockType.READ)
	@Asynchronous
	private void checkReservingTimeouts() {
		try {
			final long oldTime = nextReservingTimeoutCheck.get();
			for (MapQueue m : mapqueues.values()) {
				m.timeoutEntries();
			}

			// DO housekeeping every 10 seconds.
			nextReservingTimeoutCheck.compareAndSet(oldTime, System.currentTimeMillis() + 1000 * 10);
		} catch (Throwable t) {
			logger.warn("Exception while checking reservingTimeouts", t);
		}
	}

	@Lock(LockType.READ)
	private MapQueue getMapque(EventMap map) {
		if (map == null) {
			return null;
		}
		MapQueue ret = mapqueues.get(map.getId());

		if (ret == null) {
			ret = new MapQueue();
			MapQueue nret = mapqueues.putIfAbsent(map.getId(), ret);
			if (nret != null) {
				ret = nret;
			}
		}
		logger.info("returning queue {} for map {}", ret, map);
		return ret;
	}

	@Lock(LockType.READ)
	@Override
	public Date getReservationTimeout(EventMap map)
	{
		Date d = null;
		MapReservationQueueEntry ret = getMapque(map).getEntry(permbean.getCurrentUser());
		if (ret != null)
			d = ret.getReservationTimeout();
		return d;
	}

	public Date getReservingTimeout(EventMap map)
	{
		MapReservationQueueEntry ret = getMapque(map).getEntry(permbean.getCurrentUser());
		Date time = null;
		if (ret != null)
			time = ret.getReservationTimeout();
		return time;
	}

	@Lock(LockType.READ)
	@Override
	public Integer getQueuePosition(EventMap map, EventUser user)
	{
		if (!permbean.isCurrentUser(user) && !permbean.hasPermission(MapPermission.MANAGE_OTHERS))
			throw new RuntimeException("Use " + permbean.getCurrentUser() + "tried to get queuepos in behalf of another user: " + user);

		return getMapque(map).getPosition(permbean.getCurrentUser());
	}

	@Lock(LockType.READ)
	@Override
	public boolean isQueueEnabled() {
		return eventbean.getPropertyBoolean(LanEventPropertyKey.MAP_QUEUE);
	}

	@Lock(LockType.READ)
	@Override
	public MapReservationQueueEntry remove(EventMap map, EventUser user) {
		if (!permbean.isCurrentUser(user) && !permbean.hasPermission(MapPermission.MANAGE_OTHERS))
			throw new RuntimeException("Use " + permbean.getCurrentUser() + "tried to enter to queue in behalf of another user: " + user);

		MapQueue queue = getMapque(map);
		return queue.remove(user);
	}

	@Override
	@Lock(LockType.READ)
	@RolesAllowed(MapPermission.S_MANAGE_MAPS)
	public void forceAddToReserving(EventMap map, EventUser u, Date time) {
		MapQueue q = getMapque(map);
		q.forceAdd(u, time);

	}

	@Override
	@Lock(LockType.READ)
	public MapReservationQueueEntry enterQueue(EventMap map, EventUser user) {
		if (!isQueueEnabled()) {
			return null;
		}

		if (!permbean.isCurrentUser(user) && !permbean.hasPermission(MapPermission.MANAGE_OTHERS))
			throw new RuntimeException("Use " + permbean.getCurrentUser() + "tried to enter to queue in behalf of another user: " + user);

		if (user == null || user.isAnonymous()) {

			return null;
		}
		MapQueue queue = getMapque(map);

		MapReservationQueueEntry ret = null;
		if (queue.isInQueue(user)) {
			logger.info("User {} already in queue");
			ret = queue.getEntry(user);
		} else {
			List<PlaceSlot> slots = slotfacade.findFreePlaceSlotsForProduct(user, null);
			logger.info("User {} not yet in queue. User has {} slots", user, slots.size());
			if (!slots.isEmpty() && slots.size() >= getMinimumSlotsInQueue()) {
				ret = queue.enter(user);
				logger.info("Entered queue: {}", ret);
			}
		}
		return ret;

	}

	@Override
	@Lock(LockType.READ)
	public int getMinimumSlotsInQueue() {
		return minimumSlotsInQueue;
	}

	@Override
	@Lock(LockType.READ)
	@RolesAllowed(MapPermission.S_MANAGE_MAPS)
	public void setMinimumSlotsInQueue(int minimumSlotsInQueue) {
		this.minimumSlotsInQueue = minimumSlotsInQueue;
	}

	@Lock(LockType.READ)
	@RolesAllowed(MapPermission.S_MANAGE_MAPS)
	@Override
	public MapQueue getMapQueue(EventMap map) {
		return getMapque(map);

	}

	@Override
	@Lock(LockType.READ)
	public int getDefaultTimeoutMin() {
		return defaultTimeoutMin;
	}

	@Override
	@Lock(LockType.READ)
	public void setDefaultTimeoutMin(int defaultTimeoutMin) {
		this.defaultTimeoutMin = defaultTimeoutMin;
	}

	@Override
	@Lock(LockType.READ)
	public int getReservingSize() {
		return reservingSize;
	}

	@Override
	@Lock(LockType.READ)
	@RolesAllowed(MapPermission.S_MANAGE_MAPS)
	public void setReservingSize(int reservingSize) {
		this.reservingSize = reservingSize;
	}

	@Override
	@Lock(LockType.READ)
	@RolesAllowed(MapPermission.S_MANAGE_MAPS)
	public void forceRemove(EventMap e, EventUser u) {
		MapQueue q = getMapque(e);
		q.remove(u);
	}

}