WebSocket安卓客户端实现详解(三)–服务端主动通知

package com.example.msi.websocketdemo.websocket; import android.content.Context; import android.net.ConnectivityManager; import android.net.NetworkInfo; import android.os.Handler; import android.os.Looper; import android.os.Message; import android.os.SystemClock; import android.text.TextUtils; import com.example.msi.websocketdemo.BuildConfig; import com.example.msi.websocketdemo.WsApplication; import com.example.msi.websocketdemo.websocket.common.CallbackDataWrapper; import com.example.msi.websocketdemo.websocket.common.CallbackWrapper; import com.example.msi.websocketdemo.websocket.common.Codec; import com.example.msi.websocketdemo.websocket.common.ErrorCode; import com.example.msi.websocketdemo.websocket.common.ICallback; import com.example.msi.websocketdemo.websocket.common.IWsCallback; import com.example.msi.websocketdemo.websocket.common.WsStatus; import com.example.msi.websocketdemo.websocket.notify.NotifyListenerManager; import com.example.msi.websocketdemo.websocket.request.Action; import com.example.msi.websocketdemo.websocket.request.Request; import com.example.msi.websocketdemo.websocket.response.ChildResponse; import com.example.msi.websocketdemo.websocket.response.Response; import com.google.gson.Gson; import com.google.gson.JsonSyntaxException; import com.neovisionaries.ws.client.WebSocket; import com.neovisionaries.ws.client.WebSocketAdapter; import com.neovisionaries.ws.client.WebSocketException; import com.neovisionaries.ws.client.WebSocketFactory; import com.neovisionaries.ws.client.WebSocketFrame; import com.orhanobut.logger.Logger; import java.io.IOException; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; /** * Created by zly on 2017/6/8. */ public class WsManager { private final String TAG = this.getClass().getSimpleName(); /** * WebSocket config */ private static final long HEARTBEAT_INTERVAL = 30000;//心跳间隔 private static final int FRAME_QUEUE_SIZE = 5; private static final int CONNECT_TIMEOUT = 5000; private static final int REQUEST_TIMEOUT = 10000;//请求超时时间 private static final String DEF_TEST_URL = "测试服地址";//测试服默认地址 private static final String DEF_RELEASE_URL = "正式服地址";//正式服默认地址 private static final String DEF_URL = BuildConfig.DEBUG ? DEF_TEST_URL : DEF_RELEASE_URL; private String url; private WsStatus mStatus; private WebSocket ws; private WsListener mListener; private AtomicLong seqId = new AtomicLong(SystemClock.uptimeMillis());//每个请求的唯一标识 private ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); private Map<Long, CallbackWrapper> callbacks = new HashMap<>(); private final int SUCCESS_HANDLE = 0x01; private final int ERROR_HANDLE = 0x02; private Handler mHandler = new Handler(Looper.getMainLooper()) { @Override public void handleMessage(Message msg) { switch (msg.what) { case SUCCESS_HANDLE: CallbackDataWrapper successWrapper = (CallbackDataWrapper) msg.obj; successWrapper.getCallback().onSuccess(successWrapper.getData()); break; case ERROR_HANDLE: CallbackDataWrapper errorWrapper = (CallbackDataWrapper) msg.obj; errorWrapper.getCallback().onFail((String) errorWrapper.getData()); break; } } }; private WsManager() { } private static class WsManagerHolder { private static WsManager mInstance = new WsManager(); } public static WsManager getInstance() { return WsManagerHolder.mInstance; } public void init() { try { /** * configUrl其实是缓存在本地的连接地址 * 这个缓存本地连接地址是app启动的时候通过http请求去服务端获取的, * 每次app启动的时候会拿当前时间与缓存时间比较,超过6小时就再次去服务端获取新的连接地址更新本地缓存 */ String configUrl = ""; url = TextUtils.isEmpty(configUrl) ? DEF_URL : configUrl; ws = new WebSocketFactory().createSocket(url, CONNECT_TIMEOUT) .setFrameQueueSize(FRAME_QUEUE_SIZE)//设置帧队列最大值为5 .setMissingCloseFrameAllowed(false)//设置不允许服务端关闭连接却未发送关闭帧 .addListener(mListener = new WsListener())//添加回调监听 .connectAsynchronously();//异步连接 setStatus(WsStatus.CONNECTING); Logger.t(TAG).d("第一次连接"); } catch (IOException e) { e.printStackTrace(); } } private void doAuth() { sendReq(Action.LOGIN, null, new ICallback() { @Override public void onSuccess(Object o) { Logger.t(TAG).d("授权成功"); setStatus(WsStatus.AUTH_SUCCESS); startHeartbeat(); delaySyncData(); } @Override public void onFail(String msg) { } }); } //同步数据 private void delaySyncData() { mHandler.postDelayed(new Runnable() { @Override public void run() { sendReq(Action.SYNC, null, new ICallback() { @Override public void onSuccess(Object o) { } @Override public void onFail(String msg) { } }); } }, 300); } private void startHeartbeat() { mHandler.postDelayed(heartbeatTask, HEARTBEAT_INTERVAL); } private void cancelHeartbeat() { heartbeatFailCount = 0; mHandler.removeCallbacks(heartbeatTask); } private int heartbeatFailCount = 0; private Runnable heartbeatTask = new Runnable() { @Override public void run() { sendReq(Action.HEARTBEAT, null, new ICallback() { @Override public void onSuccess(Object o) { heartbeatFailCount = 0; } @Override public void onFail(String msg) { heartbeatFailCount++; if (heartbeatFailCount >= 3) { reconnect(); } } }); mHandler.postDelayed(this, HEARTBEAT_INTERVAL); } }; public void sendReq(Action action, Object req, ICallback callback) { sendReq(action, req, callback, REQUEST_TIMEOUT); } public void sendReq(Action action, Object req, ICallback callback, long timeout) { sendReq(action, req, callback, timeout, 1); } /** * @param action Action * @param req 请求参数 * @param callback 回调 * @param timeout 超时时间 * @param reqCount 请求次数 */ @SuppressWarnings("unchecked") private <T> void sendReq(Action action, T req, final ICallback callback, final long timeout, int reqCount) { if (!isNetConnect()) { callback.onFail("网络不可用"); return; } if (WsStatus.AUTH_SUCCESS.equals(getStatus()) || Action.LOGIN.equals(action)) { Request request = new Request.Builder<T>() .action(action.getAction()) .reqEvent(action.getReqEvent()) .seqId(seqId.getAndIncrement()) .reqCount(reqCount)