| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212 |
- 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();
- }
- }
|