package org.flowersinthesand.chat;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

import javax.servlet.ServletException;
import javax.servlet.annotation.WebServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import org.apache.catalina.websocket.MessageInbound;
import org.apache.catalina.websocket.StreamInbound;
import org.apache.catalina.websocket.WebSocketServlet;
import org.apache.catalina.websocket.WsOutbound;

import com.google.gson.Gson;

@WebServlet(urlPatterns = "/chat", asyncSupported = true)
public class ChatServlet extends WebSocketServlet {

	private static final long serialVersionUID = -8348163963937728320L;

	private Map<String, ChatMessageInbound> connections = new ConcurrentHashMap<String, ChatMessageInbound>();
	private BlockingQueue<Event> queue = new LinkedBlockingQueue<Event>();
	private Thread broadcaster = new Thread(new Runnable() {
		@Override
		public void run() {
			while (true) {
				try {
					Event event = queue.take();
					for (Entry<String, ChatMessageInbound> entry : connections.entrySet()) {
						try {
							send(entry.getValue(), event);
						} catch (IOException ex) {
							Event e = new Event();
							e.socket = entry.getKey();
							e.type = "close";

							fire(e);
						}
					}
				} catch (InterruptedException e) {
					break;
				}
			}
		}
	});
	private ThreadLocal<HttpServletRequest> requestHolder = new ThreadLocal<HttpServletRequest>();

	@Override
	public void init() throws ServletException {
		super.init();
		broadcaster.setDaemon(true);
		broadcaster.start();
	}

	@Override
	protected void service(HttpServletRequest request, HttpServletResponse response)
			throws ServletException, IOException {
		requestHolder.set(request);
		super.service(request, response);
	}

	@Override
	protected StreamInbound createWebSocketInbound(String protocol) {
		HttpServletRequest request = requestHolder.get();
		return new ChatMessageInbound(request.getParameter("id"));
	}

	private void send(ChatMessageInbound messageInbound, Event event) throws IOException {
		String data = new Gson().toJson(event);
		messageInbound.getWsOutbound().writeTextMessage(CharBuffer.wrap(data));
	}

	private void fire(Event event) {
		if (event.type.equals("close")) {
			connections.remove(event.socket);
		}

		handle(event);
	}

	private void handle(Event event) {
		if (event.type.equals("message")) {
			queue.offer(new Event("message", event.data));
		}
	}

	private class ChatMessageInbound extends MessageInbound {
		String id;

		public ChatMessageInbound(String id) {
			this.id = id;
		}

		@Override
		protected void onOpen(WsOutbound outbound) {
			connections.put(id, this);

			Event e = new Event();
			e.socket = id;
			e.type = "open";

			fire(e);
		}

		@Override
		protected void onClose(int status) {
			Event e = new Event();
			e.socket = id;
			e.type = "close";

			fire(e);
		}

		@Override
		protected void onTextMessage(CharBuffer message) throws IOException {
			fire(new Gson().fromJson(message.toString(), Event.class));
		}

		@Override
		protected void onBinaryMessage(ByteBuffer message) throws IOException {
			throw new UnsupportedOperationException("Binary message not supported.");
		}
	}

	private static class Event {
		private String socket;
		private String type;
		private Object data;

		public Event() {

		}

		public Event(String type, Object data) {
			this.type = type;
			this.data = data;
		}
	}

}
@@ -926,16 +926,126 @@
 28320L;%0D
+%0A%09private static final ThreadLocal%3CHttpServletRequest%3E requestHolder = new ThreadLocal%3CHttpServletRequest%3E();%0D
 %0A%0D%0A%09priv
@@ -1753,105 +1753,8 @@
 %7D);%0D
-%0A%09private ThreadLocal%3CHttpServletRequest%3E requestHolder = new ThreadLocal%3CHttpServletRequest%3E();%0D
 %0A%0D%0A%09

Declare requestHolder as static final