EasemobUtils.java 6.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212
  1. package com.goafanti.easemob;
  2. import java.io.IOException;
  3. import java.util.Optional;
  4. import java.util.concurrent.ExecutionException;
  5. import java.util.concurrent.Future;
  6. import org.apache.http.HttpEntity;
  7. import org.apache.http.HttpResponse;
  8. import org.apache.http.client.config.RequestConfig;
  9. import org.apache.http.client.methods.HttpGet;
  10. import org.apache.http.client.methods.HttpPost;
  11. import org.apache.http.client.methods.HttpRequestBase;
  12. import org.apache.http.client.methods.HttpUriRequest;
  13. import org.apache.http.concurrent.FutureCallback;
  14. import org.apache.http.entity.StringEntity;
  15. import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
  16. import org.apache.http.impl.nio.client.HttpAsyncClients;
  17. import org.apache.http.util.EntityUtils;
  18. import org.slf4j.Logger;
  19. import org.slf4j.LoggerFactory;
  20. import org.springframework.beans.factory.DisposableBean;
  21. import org.springframework.beans.factory.InitializingBean;
  22. import org.springframework.beans.factory.annotation.Autowired;
  23. import org.springframework.beans.factory.annotation.Value;
  24. import org.springframework.http.HttpMethod;
  25. import com.alibaba.fastjson.JSON;
  26. import com.alibaba.fastjson.JSONObject;
  27. import com.goafanti.common.utils.LoggerUtils;
  28. import com.goafanti.easemob.bo.EasemobInfo;
  29. import com.goafanti.easemob.queue.EasemobRedisQueue;
  30. public class EasemobUtils implements InitializingBean, DisposableBean {
  31. private static final String UTF_8 = "UTF-8";
  32. private static final Logger logger = LoggerFactory.getLogger(EasemobUtils.class);
  33. private static final Optional<Integer> UN_AUTH = Optional.of(401);
  34. private static final Optional<Integer> REACH_LIMIT = Optional.of(429);
  35. @Value(value = "${easemob.client.url}")
  36. private String clientUrl;
  37. @Value(value = "${easemob.client.id}")
  38. private String clientId;
  39. @Value(value = "${easemob.client.secret}")
  40. private String clientSecret;
  41. private String token;
  42. @Autowired
  43. private EasemobRedisQueue jedisQueue;
  44. private void auth() {
  45. JSONObject jo = new JSONObject();
  46. jo.put("grant_type", "client_credentials");
  47. jo.put("client_id", clientId);
  48. jo.put("client_secret", clientSecret);
  49. JSONObject res = send(
  50. new EasemobInfo().uri("/token").data(jo.toJSONString()).method(HttpMethod.POST).withAuth(false));
  51. if (res != null) {
  52. token = res.getString("access_token");
  53. }
  54. }
  55. private HttpUriRequest buildRequest(EasemobInfo info) {
  56. HttpUriRequest req = null;
  57. switch (info.getMethod()) {
  58. case POST:
  59. req = new HttpPost(clientUrl + info.getUri());
  60. ((HttpPost) req).setEntity(new StringEntity(info.getData(), UTF_8));
  61. break;
  62. default:
  63. req = new HttpGet(clientUrl + info.getUri());
  64. break;
  65. }
  66. if (info.isWithAuth()) {
  67. req.addHeader("Authorization", "Bearer " + token);
  68. }
  69. ((HttpRequestBase) req).setConfig(RequestConfig.custom().setConnectTimeout(10000)
  70. .setConnectionRequestTimeout(10000).setSocketTimeout(10000).build());
  71. req.addHeader("Content-Type", "application/json");
  72. return req;
  73. }
  74. public JSONObject send(EasemobInfo info) {
  75. HttpUriRequest req = buildRequest(info);
  76. CloseableHttpAsyncClient httpclient = HttpAsyncClients.createDefault();
  77. httpclient.start();
  78. LoggerUtils.debug(logger, "SEND: [%s] - [%s]", info.getMethod(), info.getUri());
  79. Future<HttpResponse> future = httpclient.execute(req, null);
  80. try {
  81. HttpResponse response = future.get();
  82. Optional<Integer> rescode = getResCode(response);
  83. if (info.isWithAuth() && isUnauth(rescode)) {
  84. auth();
  85. } else if (isReachLimit(rescode)) {
  86. LoggerUtils.debug(logger, "Reach Easemob API limitation!");
  87. } else {
  88. HttpEntity entity = response.getEntity();
  89. if (entity != null) {
  90. try {
  91. String resStr = EntityUtils.toString(entity, UTF_8);
  92. return (JSONObject) JSON.parse(resStr);
  93. } catch (Exception e) {
  94. LoggerUtils.debug(logger, e.getMessage(), e);
  95. }
  96. }
  97. }
  98. LoggerUtils.debug(logger, response);
  99. } catch (InterruptedException | ExecutionException e) {
  100. LoggerUtils.debug(logger, e.getMessage(), e);
  101. } finally {
  102. try {
  103. httpclient.close();
  104. } catch (IOException e) {
  105. }
  106. }
  107. return info.getTryCount() > 1 ? null : send(info.tryCount(info.getTryCount() + 1));
  108. }
  109. public void sendLater(EasemobInfo info) {
  110. jedisQueue.pushFromTail(info);
  111. }
  112. public void sendAsync(EasemobInfo info) {
  113. HttpUriRequest req = buildRequest(info);
  114. CloseableHttpAsyncClient httpclient = HttpAsyncClients.createDefault();
  115. httpclient.start();
  116. LoggerUtils.debug(logger, "SEND: [%s] - [%s]", info.getMethod(), info.getUri());
  117. httpclient.execute(req, new FutureCallback<HttpResponse>() {
  118. @Override
  119. public void failed(Exception ex) {
  120. sendLater(info.tryCount(info.getTryCount() + 1));
  121. close();
  122. }
  123. @Override
  124. public void completed(HttpResponse result) {
  125. Optional<Integer> rescode = getResCode(result);
  126. if (info.isWithAuth() && isUnauth(rescode)) {
  127. auth();
  128. } else if (isReachLimit(rescode)) {
  129. LoggerUtils.debug(logger, "Reach Easemob API limitation!");
  130. sendLater(info.tryCount(info.getTryCount() + 1));
  131. }
  132. LoggerUtils.debug(logger, result);
  133. close();
  134. }
  135. @Override
  136. public void cancelled() {
  137. close();
  138. LoggerUtils.debug(logger, "CANCELLED: [%s] - [%s]", req.getMethod(), req.getURI());
  139. }
  140. private void close() {
  141. try {
  142. httpclient.close();
  143. } catch (IOException e) {
  144. }
  145. }
  146. });
  147. }
  148. public void sendMessage(String from, String to, String msg, Object... value) {
  149. sendAsync(new EasemobInfo().uri("/messages").data(buildMessage(from, to, null, String.format(msg, value)))
  150. .method(HttpMethod.POST));
  151. }
  152. public void sendMessage(String from, String to, Object ext, String msg, Object... value) {
  153. sendAsync(new EasemobInfo().uri("/messages").data(buildMessage(from, to, ext, String.format(msg, value)))
  154. .method(HttpMethod.POST));
  155. }
  156. private String buildMessage(String from, String to, Object ext, String msg) {
  157. JSONObject message = new JSONObject();
  158. message.put("type", "txt");
  159. message.put("msg", msg);
  160. JSONObject jo = new JSONObject();
  161. jo.put("target_type", "users");
  162. jo.put("target", new String[] { to });
  163. jo.put("msg", message);
  164. jo.put("from", from);
  165. if (ext != null) {
  166. jo.put("ext", ext);
  167. }
  168. return jo.toJSONString();
  169. }
  170. private Optional<Integer> getResCode(HttpResponse response) {
  171. return Optional.ofNullable(response).map(res -> res.getStatusLine()).map(sl -> sl.getStatusCode());
  172. }
  173. private boolean isUnauth(Optional<Integer> responseStatus) {
  174. return UN_AUTH.equals(responseStatus);
  175. }
  176. private boolean isReachLimit(Optional<Integer> responseStatus) {
  177. return REACH_LIMIT.equals(responseStatus);
  178. }
  179. @Override
  180. public void destroy() throws Exception {
  181. LoggerUtils.debug(logger, "消息系统关闭");
  182. }
  183. @Override
  184. public void afterPropertiesSet() throws Exception {
  185. auth();
  186. }
  187. }