package com.goafanti.easemob;

import java.io.IOException;
import java.util.Optional;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.client.methods.HttpRequestBase;
import org.apache.http.client.methods.HttpUriRequest;
import org.apache.http.concurrent.FutureCallback;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.HttpMethod;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.goafanti.common.utils.LoggerUtils;
import com.goafanti.easemob.bo.EasemobInfo;
import com.goafanti.easemob.queue.EasemobRedisQueue;

public class EasemobUtils implements InitializingBean, DisposableBean {
	private static final String				UTF_8		= "UTF-8";
	private static final Logger				logger		= LoggerFactory.getLogger(EasemobUtils.class);
	private static final Optional<Integer>	UN_AUTH		= Optional.of(401);
	private static final Optional<Integer>	REACH_LIMIT	= Optional.of(429);

	@Value(value = "${easemob.client.url}")
	private String							clientUrl;
	@Value(value = "${easemob.client.id}")
	private String							clientId;
	@Value(value = "${easemob.client.secret}")
	private String							clientSecret;

	private String							token;

	@Autowired
	private EasemobRedisQueue				jedisQueue;

	private void auth() {
		JSONObject jo = new JSONObject();
		jo.put("grant_type", "client_credentials");
		jo.put("client_id", clientId);
		jo.put("client_secret", clientSecret);
		JSONObject res = send(
				new EasemobInfo().uri("/token").data(jo.toJSONString()).method(HttpMethod.POST).withAuth(false));
		if (res != null) {
			token = res.getString("access_token");
		}
	}

	private HttpUriRequest buildRequest(EasemobInfo info) {
		HttpUriRequest req = null;
		switch (info.getMethod()) {
		case POST:
			req = new HttpPost(clientUrl + info.getUri());
			((HttpPost) req).setEntity(new StringEntity(info.getData(), UTF_8));
			break;
		default:
			req = new HttpGet(clientUrl + info.getUri());
			break;
		}
		if (info.isWithAuth()) {
			req.addHeader("Authorization", "Bearer " + token);
		}
		((HttpRequestBase) req).setConfig(RequestConfig.custom().setConnectTimeout(10000)
				.setConnectionRequestTimeout(10000).setSocketTimeout(10000).build());
		req.addHeader("Content-Type", "application/json");
		return req;
	}

	public JSONObject send(EasemobInfo info) {
		HttpUriRequest req = buildRequest(info);
		CloseableHttpAsyncClient httpclient = HttpAsyncClients.createDefault();
		httpclient.start();
		LoggerUtils.debug(logger, "SEND: [%s] - [%s]", info.getMethod(), info.getUri());
		Future<HttpResponse> future = httpclient.execute(req, null);
		try {
			HttpResponse response = future.get();
			Optional<Integer> rescode = getResCode(response);
			if (info.isWithAuth() && isUnauth(rescode)) {
				auth();
			} else if (isReachLimit(rescode)) {
				LoggerUtils.debug(logger, "Reach Easemob API limitation!");
			} else {
				HttpEntity entity = response.getEntity();
				if (entity != null) {
					try {
						String resStr = EntityUtils.toString(entity, UTF_8);
						return (JSONObject) JSON.parse(resStr);
					} catch (Exception e) {
						LoggerUtils.debug(logger, e.getMessage(), e);
					}
				}
			}
			LoggerUtils.debug(logger, response);
		} catch (InterruptedException | ExecutionException e) {
			LoggerUtils.debug(logger, e.getMessage(), e);
		} finally {
			try {
				httpclient.close();
			} catch (IOException e) {
			}
		}
		return info.getTryCount() > 1 ? null : send(info.tryCount(info.getTryCount() + 1));
	}

	public void sendLater(EasemobInfo info) {
		jedisQueue.pushFromTail(info);
	}

	public void sendAsync(EasemobInfo info) {
		HttpUriRequest req = buildRequest(info);
		CloseableHttpAsyncClient httpclient = HttpAsyncClients.createDefault();
		httpclient.start();
		LoggerUtils.debug(logger, "SEND: [%s] - [%s]", info.getMethod(), info.getUri());
		httpclient.execute(req, new FutureCallback<HttpResponse>() {
			@Override
			public void failed(Exception ex) {
				sendLater(info.tryCount(info.getTryCount() + 1));
				close();
			}

			@Override
			public void completed(HttpResponse result) {
				Optional<Integer> rescode = getResCode(result);
				if (info.isWithAuth() && isUnauth(rescode)) {
					auth();
				} else if (isReachLimit(rescode)) {
					LoggerUtils.debug(logger, "Reach Easemob API limitation!");
					sendLater(info.tryCount(info.getTryCount() + 1));
				}
				LoggerUtils.debug(logger, result);
				close();
			}

			@Override
			public void cancelled() {
				close();
				LoggerUtils.debug(logger, "CANCELLED: [%s] - [%s]", req.getMethod(), req.getURI());
			}

			private void close() {
				try {
					httpclient.close();
				} catch (IOException e) {
				}
			}
		});
	}

	public void sendMessage(String from, String to, String msg, Object... value) {
		sendAsync(new EasemobInfo().uri("/messages").data(buildMessage(from, to, null, String.format(msg, value)))
				.method(HttpMethod.POST));
	}

	public void sendMessage(String from, String to, Object ext, String msg, Object... value) {
		sendAsync(new EasemobInfo().uri("/messages").data(buildMessage(from, to, ext, String.format(msg, value)))
				.method(HttpMethod.POST));
	}

	private String buildMessage(String from, String to, Object ext, String msg) {
		JSONObject message = new JSONObject();
		message.put("type", "txt");
		message.put("msg", msg);
		JSONObject jo = new JSONObject();
		jo.put("target_type", "users");
		jo.put("target", new String[] { to });
		jo.put("msg", message);
		jo.put("from", from);
		if (ext != null) {
			jo.put("ext", ext);
		}
		return jo.toJSONString();
	}

	private Optional<Integer> getResCode(HttpResponse response) {
		return Optional.ofNullable(response).map(res -> res.getStatusLine()).map(sl -> sl.getStatusCode());
	}

	private boolean isUnauth(Optional<Integer> responseStatus) {
		return UN_AUTH.equals(responseStatus);
	}

	private boolean isReachLimit(Optional<Integer> responseStatus) {
		return REACH_LIMIT.equals(responseStatus);
	}

	@Override
	public void destroy() throws Exception {
		LoggerUtils.debug(logger, "消息系统关闭");
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		auth();
	}

}
