1.添加websocket的基础service

websocket/src/main/java/com/xhly/websocket/service/LongConnService.java

@@ -0,0 +1,379 @@
+package com.xhly.websocket.service;
+
+import android.app.Service;
+import android.content.Intent;
+import android.os.Handler;
+import android.os.IBinder;
+import android.text.TextUtils;
+import android.util.Log;
+import android.widget.Toast;
+
+import androidx.annotation.Nullable;
+
+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.xhly.corelib.utils.NetworkUtils;
+import com.xhly.websocket.utils.WebSocketUtils;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import io.reactivex.Observable;
+import io.reactivex.Observer;
+import io.reactivex.android.schedulers.AndroidSchedulers;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.schedulers.Schedulers;
+
+public abstract class LongConnService extends Service {
+    /**
+     * 聊天
+     */
+    protected static final int SOCKET_TYPE = 1;
+    /**
+     * WebSocket config
+     */
+    private static final int FRAME_QUEUE_SIZE = 5;
+    /**
+     * 心跳时间间隔单位是s
+     */
+    private static final long HEART_INTERVAL = 10;
+    /**
+     * 未收到心跳响应的次数
+     */
+    public static int HEART_NOT_RECEIVE_TIME = 0;
+    /**
+     * 是否收到心跳响应
+     */
+    public static boolean HAS_HEART_RECEIVE = false;
+    /**
+     * 连接超时时间，单位毫秒
+     */
+    public static int CONNECT_TIMEOUT = 20000;
+    public static boolean closeWebsocket = false;
+    public String TAG = LongConnService.class.getSimpleName();
+    protected int type = 1;
+    protected String connectUrl;
+    /**
+     * 重连最小时间间隔,单位毫秒
+     */
+    private long minInterval = 10000;
+    /**
+     * 重连最大时间间隔,单位毫秒
+     */
+    private long maxInterval = 60000;
+    private Handler mHandler = new Handler();
+    /**
+     * 重连次数
+     */
+    private int reconnectCount = 0;
+    /**
+     * 心跳subscription
+     */
+    private Disposable heartSubscription;
+    /**
+     * 是否连接状态
+     */
+    private boolean connected = false;
+    /**
+     * 是否允许重连
+     */
+    private boolean isReConnect = true;
+    private WsStatus mStatus;
+    private WebSocket webSocket;
+    private WsListener mListener;
+
+    /**
+     * 设置已收到心跳
+     */
+    public static void hasReceiveHeart() {
+        HEART_NOT_RECEIVE_TIME = 0;
+        HAS_HEART_RECEIVE = true;
+    }
+
+    @Override
+    public void onCreate() {
+        super.onCreate();
+        Log.i(TAG, "onCreate()............");
+        mListener = new WsListener();
+    }
+
+    @Override
+    public int onStartCommand(Intent intent, int flags, int startId) {
+        Log.i(TAG, "onStartCommand()............");
+        closeWebsocket = false;
+        if (mListener == null) {
+            mListener = new WsListener();
+        }
+        startConnect();
+        return START_STICKY;
+    }
+
+    @Override
+    public void onDestroy() {
+        Log.i(TAG, "onDestroy()............");
+        if (heartSubscription != null) {
+            heartSubscription.dispose();
+        }
+        disConnect();
+        closeWebsocket = true;
+        super.onDestroy();
+    }
+
+    @Nullable
+    @Override
+    public IBinder onBind(Intent intent) {
+        throw new UnsupportedOperationException("Not yet implemented");
+    }
+
+    /**
+     * 断开连接
+     */
+    private void disConnect() {
+        if (mHandler != null) {
+            mHandler.removeCallbacksAndMessages(null);
+        }
+        isReConnect = false;
+        connected = false;
+        if (webSocket != null) {
+            setStatus(WsStatus.DISCONNECT);
+            cancelReconnect();
+            webSocket.disconnect();
+            webSocket.clearListeners();  // 清除回调，避免多次链接导致上个链接的消息接收问题
+            webSocket = null;
+        }
+    }
+
+    /**
+     * 取消重连
+     */
+    private void cancelReconnect() {
+        reconnectCount = 0;
+    }
+
+
+    /**
+     * 链接成功,子类可以拓展
+     */
+    public void onConnectedSuccess() {
+
+    }
+
+    /**
+     * 连接长链接
+     */
+    private synchronized void startConnect() {
+        if (TextUtils.isEmpty("用户ID")) {
+            setStatus(WsStatus.DISCONNECT);
+            return;
+        }
+        if (!NetworkUtils.isAvailable(this)) {
+            Toast.makeText(this, "网络不可用", Toast.LENGTH_SHORT);
+            return;
+        }
+
+        isReConnect = true;
+      /*
+      //存在！=null 并且open时断联的情况，所以注释
+      if (webSocket != null && webSocket.isOpen()) {
+            setStatus(WsStatus.CONNECT_SUCCESS);
+            return;
+        }*/
+        if (getStatus() == WsStatus.CONNECTING) {
+            return;
+        }
+
+        if (!connected) {
+            if (webSocket != null) {
+                webSocket.disconnect();
+                webSocket = null;
+            }
+            try {
+                Log.i(TAG, "第一次连接: " + connectUrl);
+                // 异步连接
+                webSocket = createWebSocket();
+                setStatus(WsStatus.CONNECTING);
+//                checkConnectState();
+            } catch (IOException e) {
+                e.printStackTrace();
+            }
+        }
+    }
+
+    public WsStatus getStatus() {
+        return mStatus;
+    }
+
+    private void setStatus(WsStatus status) {
+        this.mStatus = status;
+    }
+
+    private WebSocket createWebSocket() throws IOException {
+        Log.i(TAG, "创建连接：createWebSocket()");
+        return new WebSocketFactory().createSocket(connectUrl, CONNECT_TIMEOUT)
+                //设置帧队列最大值为5
+                .setFrameQueueSize(FRAME_QUEUE_SIZE)
+                //设置不允许服务端关闭连接却未发送关闭帧
+                .setMissingCloseFrameAllowed(false)
+                //添加回调监听
+                .addListener(mListener).connectAsynchronously();
+    }
+
+    /**
+     * 重新连接
+     */
+    private void reConnect() {
+        if (!NetworkUtils.isAvailable(this)) {
+            reconnectCount = 0;
+            isReConnect = false;
+            closeWebsocket = true;
+            Log.i(TAG, "重连失败网络不可用，当前closeWebsocket==" + closeWebsocket);
+            return;
+        }
+        if (HAS_HEART_RECEIVE) {
+            return;
+        }
+        if (webSocket != null && !webSocket.isOpen() &&//当前连接断开了
+                getStatus() != WsStatus.CONNECTING && !connected) { //不是正在重连状态
+            setStatus(WsStatus.CONNECTING);
+            long reconnectTime = minInterval;
+            if (reconnectCount > 3) {
+                long temp = minInterval * (reconnectCount - 2);
+                reconnectTime = temp > maxInterval ? maxInterval : temp;
+            }
+            mHandler.postDelayed(new Runnable() {
+                @Override
+                public void run() {
+                    try {
+                        if (webSocket.isOpen() && isConnect()) {
+                            return;
+                        }
+                        if (webSocket != null) {
+                            webSocket.disconnect();
+                            webSocket = null;
+                        }
+                        webSocket = createWebSocket();
+                        reconnectCount++;
+                    } catch (IOException e) {
+                        e.printStackTrace();
+                    }
+                }
+            }, reconnectTime);
+        }
+    }
+
+    /**
+     * 是否连接
+     */
+    private boolean isConnect() {
+        return connected;
+    }
+
+    /**
+     * 开始socket心跳
+     */
+    private void startHeartInterval(final WebSocket websocket) {
+        if (HEART_NOT_RECEIVE_TIME > 3) {
+            // 有3次心跳没有收到服务器响应，则认为是已经断开连接，则重新连接
+            connected = false;
+            startConnect();
+            return;
+        }
+        // 在客户端上增加重试的逻辑。先发送心跳包，在N秒内若没有收到回复，则再发送一次。重试M次。
+        WebSocketUtils.hearBeat(webSocket);
+        HEART_NOT_RECEIVE_TIME++;
+        HAS_HEART_RECEIVE = false;
+    }
+
+    /**
+     * 处理socket响应消息
+     *
+     * @param text 要处理的文本
+     */
+    public abstract void processTextMessage(String text);
+
+    /**
+     * 继承默认的监听空实现WebSocketAdapter,重写我们需要的方法
+     * onTextMessage 收到文字信息
+     * onConnected 连接成功
+     * onConnectError 连接失败
+     * onDisconnected 连接关闭
+     */
+    private class WsListener extends WebSocketAdapter {
+        @Override
+        public void onConnected(final WebSocket websocket, Map<String, List<String>> headers) throws Exception {
+            super.onConnected(websocket, headers);
+            Log.i(TAG, "连接成功：" + headers.toString());
+            connected = true;
+            setStatus(WsStatus.CONNECT_SUCCESS);
+            cancelReconnect();// 连接成功的时候取消重连,初始化连接次数
+            if (type == SOCKET_TYPE) {
+                WebSocketHelper.getInstance().build(websocket);
+                //调用成功方法
+                onConnectedSuccess();
+                Observable.interval(HEART_INTERVAL, TimeUnit.SECONDS).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<Long>() {
+                    @Override
+                    public void onSubscribe(Disposable d) {
+                        Log.d(TAG, "订阅开始");
+                        heartSubscription = d;
+                    }
+
+                    @Override
+                    public void onNext(Long aLong) {
+                        Log.v(TAG, "发送心跳");
+                        startHeartInterval(websocket);
+                    }
+
+                    @Override
+                    public void onError(Throwable e) {
+
+                    }
+
+                    @Override
+                    public void onComplete() {
+
+                    }
+                });
+            }
+        }
+
+        @Override
+        public void onConnectError(WebSocket websocket, WebSocketException exception) throws Exception {
+            super.onConnectError(websocket, exception);
+            exception.printStackTrace();
+            Log.i(TAG, "连接错误: " + exception.getMessage());
+            HAS_HEART_RECEIVE = false;
+            setStatus(WsStatus.CONNECT_FAIL);
+            connected = false;
+            if (isReConnect) {
+                reConnect();//连接断开的时候调用重连方法
+            }
+        }
+
+        @Override
+        public void onDisconnected(WebSocket websocket, WebSocketFrame serverCloseFrame, WebSocketFrame clientCloseFrame, boolean closedByServer) throws Exception {
+            super.onDisconnected(websocket, serverCloseFrame, clientCloseFrame, closedByServer);
+            Log.i(TAG, "断开连接,是否重连" + isReConnect + ",当前closeWebsocket==" + closeWebsocket);
+            HAS_HEART_RECEIVE = false;
+            connected = false;
+            if (heartSubscription != null) {
+                heartSubscription.dispose();
+            }
+            setStatus(WsStatus.CONNECT_FAIL);
+            if (isReConnect) {
+                reConnect();//连接断开的时候调用重连方法
+            }
+        }
+
+        @Override
+        public void onTextMessage(WebSocket websocket, String text) throws Exception {
+            super.onTextMessage(websocket, text);
+            hasReceiveHeart();
+            processTextMessage(text);
+        }
+    }
+}