package allbegray.slack.rtm;

import allbegray.slack.exception.SlackException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.asynchttpclient.AsyncHttpClient;
import org.asynchttpclient.BoundRequestBuilder;
import org.asynchttpclient.proxy.ProxyServer;
import org.asynchttpclient.ws.DefaultWebSocketListener;
import org.asynchttpclient.ws.WebSocket;
import org.asynchttpclient.ws.WebSocketUpgradeHandler;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static org.asynchttpclient.Dsl.*;

public class SlackRealTimeMessagingClient {

	private static Log logger = LogFactory.getLog(SlackRealTimeMessagingClient.class);

	private String webSocketUrl;
	private ProxyServerInfo proxyServerInfo;
	private AsyncHttpClient asyncHttpClient;
	private WebSocket webSocket;
	private Map<String, List<EventListener>> listeners = new HashMap<String, List<EventListener>>();
	private List<CloseListener> closeListeners = new ArrayList<CloseListener>();
	private List<FailureListener> failureListeners = new ArrayList<FailureListener>();
	private boolean stop;
	private ObjectMapper mapper;
	private Integer pingMillis;

	public SlackRealTimeMessagingClient(String webSocketUrl) {
		this(webSocketUrl, null, null, null);
	}

	public SlackRealTimeMessagingClient(String webSocketUrl, ProxyServerInfo proxyServerInfo) {
		this(webSocketUrl, null, null, proxyServerInfo);
	}

	public SlackRealTimeMessagingClient(String webSocketUrl, ObjectMapper mapper) {
		this(webSocketUrl, mapper, null, null);
	}

	public SlackRealTimeMessagingClient(String webSocketUrl, ObjectMapper mapper, ProxyServerInfo proxyServerInfo) {
		this(webSocketUrl, mapper, null, proxyServerInfo);
	}

	public SlackRealTimeMessagingClient(String webSocketUrl, Integer pingMillis) {
		this(webSocketUrl, null, pingMillis, null);
	}

	public SlackRealTimeMessagingClient(String webSocketUrl, Integer pingMillis, ProxyServerInfo proxyServerInfo) {
		this(webSocketUrl, null, pingMillis, proxyServerInfo);
	}

	public SlackRealTimeMessagingClient(String webSocketUrl, ObjectMapper mapper, Integer pingMillis, ProxyServerInfo proxyServerInfo) {
		if (mapper == null) {
			mapper = new ObjectMapper();
		}
		if (pingMillis == null) {
			pingMillis = 3 * 1000;
		}
		this.webSocketUrl = webSocketUrl;
		this.proxyServerInfo = proxyServerInfo;
		this.mapper = mapper;
		this.pingMillis = pingMillis;
	}

	public void addListener(Event event, EventListener listener) {
		addListener(event.name().toLowerCase(), listener);
	}

	public void addListener(String event, EventListener listener) {
		List<EventListener> eventListeners = listeners.get(event);
		if (eventListeners == null) {
			eventListeners = new ArrayList<EventListener>();
			listeners.put(event, eventListeners);
		}
		eventListeners.add(listener);
	}

	public void addCloseListener(CloseListener listener) {
		closeListeners.add(listener);
	}

	public void addFailureListener(FailureListener listener) {
		failureListeners.add(listener);
	}

	public void close() {
		logger.info("Slack RTM closing...");

		stop = true;
		if (webSocket != null && webSocket.isOpen()) {
			try {
				webSocket.close();
			} catch (IOException e) {
				// ignore
			}
		}
		if (asyncHttpClient != null && !asyncHttpClient.isClosed()) {
			try {
				asyncHttpClient.close();
			} catch (IOException e) {
				// ignore
			}
		}

		logger.info("Slack RTM closed.");
	}

	public boolean connect() {
		try {
			if (proxyServerInfo != null) {
				ProxyServer.Builder proxyServer = proxyServer(proxyServerInfo.getHost(), proxyServerInfo.getPort());
				if (proxyServerInfo.getPrincipal() != null && proxyServerInfo.getPassword() != null) {
					proxyServer = proxyServer.setRealm(basicAuthRealm(proxyServerInfo.getPrincipal(), proxyServerInfo.getPassword()));
				}
				asyncHttpClient = asyncHttpClient(config().setProxyServer(proxyServer));
			} else {
				asyncHttpClient = asyncHttpClient();
			}
			BoundRequestBuilder requestBuilder = asyncHttpClient.prepareGet(webSocketUrl);
			webSocket = requestBuilder.execute(new WebSocketUpgradeHandler.Builder().addWebSocketListener(new DefaultWebSocketListener() {

				@Override
				public void onMessage(String message) {
					String type = null;
					JsonNode node = null;
					try {
						node = mapper.readTree(message);
						type = node.findPath("type").asText();
					} catch (Exception e) {
						logger.error(e);
					}

					if (!"pong".equals(type)) {
						logger.info("Slack RTM message : " + message);
					}

					if (type != null) {
						List<EventListener> eventListeners = listeners.get(type);
						if (eventListeners != null && !eventListeners.isEmpty()) {
							for (EventListener listener : eventListeners) {
								listener.onMessage(node);
							}
						}
					}
				}

				@Override
				public void onClose(WebSocket websocket) {
					super.onClose(websocket);
					stop = true;
					if (closeListeners != null && !closeListeners.isEmpty()) {
						for (CloseListener listener : closeListeners) {
							listener.onClose();
						}
					}
				}

				@Override
				public void onError(Throwable t) {
					stop = true;
					t.printStackTrace();
					if (failureListeners != null && !failureListeners.isEmpty()) {
						for (FailureListener listener : failureListeners) {
							listener.onFailure(new SlackException(t));
						}
					}
				}

			}).build()).get();

			logger.info("connected Slack RTM(Real Time Messaging) server : " + webSocketUrl);

			await();

		} catch (Exception e) {
			close();
			throw new SlackException(e);
		}
		return true;
	}

	private long socketId = 0;

	private void ping() {
		ObjectNode pingMessage = mapper.createObjectNode();
		pingMessage.put("id", ++socketId);
		pingMessage.put("type", "ping");
		String pingJson = pingMessage.toString();
		webSocket.sendMessage(pingJson);

		logger.debug("ping : " + pingJson);
	}

	private void await() {
		Thread thread = new Thread(new Runnable() {

			@Override
			public void run() {
				try {
					Thread.sleep(pingMillis);
					while (!stop) {
						ping();
						Thread.sleep(pingMillis);
					}
				} catch (Exception e) {
					throw new SlackException(e);
				}
			}
		});
		thread.start();
	}

	/**
	 * This method sends message to slack using webSocket client
	 * @param message
	 */
	public void sendMessage(String message){
		webSocket.sendMessage(message);
	}

	/**
	 * This method returns the webSocket connected with Slack RTM
	 * @return the webSocket client object
	 */
	public WebSocket getWebsocket(){
		return webSocket;
	}

}

@@ -5359,24 +5359,47 @@
 top = true;%0A
+%09%09%09%09%09if (t != null) %7B%0A%09
 %09%09%09%09%09t.print
@@ -5408,24 +5408,31 @@
 ackTrace();%0A
+%09%09%09%09%09%7D%0A
 %09%09%09%09%09if (fai

fix NPE