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 UN_AUTH = Optional.of(401); private static final Optional 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 future = httpclient.execute(req, null); try { HttpResponse response = future.get(); Optional 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() { @Override public void failed(Exception ex) { sendLater(info.tryCount(info.getTryCount() + 1)); close(); } @Override public void completed(HttpResponse result) { Optional 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 getResCode(HttpResponse response) { return Optional.ofNullable(response).map(res -> res.getStatusLine()).map(sl -> sl.getStatusCode()); } private boolean isUnauth(Optional responseStatus) { return UN_AUTH.equals(responseStatus); } private boolean isReachLimit(Optional responseStatus) { return REACH_LIMIT.equals(responseStatus); } @Override public void destroy() throws Exception { LoggerUtils.debug(logger, "消息系统关闭"); } @Override public void afterPropertiesSet() throws Exception { auth(); } }