更改了udp的机制

hxb

2021-12-12 d8bf4f4d66715f002d024cae92862c1d83daa425

更改了udp的机制

 HDLSDK/hdl-connect/src/main/java/com/hdl/sdk/connect/socket/HDLAuthSocket.java

@@ -1,6 +1,5 @@
package com.hdl.sdk.connect.socket;

import android.net.ipsec.ike.TunnelModeChildSessionParams;
import android.text.TextUtils;

import com.google.gson.Gson;
@@ -30,15 +29,13 @@
import com.hdl.sdk.connect.bean.LinkResponse;
import com.hdl.sdk.connect.protocol.LinkMessageDecoder;
import com.hdl.sdk.connect.protocol.LinkMessageEncoder;
import com.hdl.sdk.socket.SocketBoot;
import com.hdl.sdk.socket.SocketOptions;
import com.hdl.sdk.socket.client.UdpClient;
import com.hdl.sdk.socket.codec.MessagePipeLine;
import com.hdl.sdk.socket.listener.SendListener;
import com.hdl.sdk.socket.udp.UdpSocketBoot;
import com.hdl.sdk.socket.udp.UdpSocketOptions;

import java.io.UnsupportedEncodingException;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

@@ -47,7 +44,7 @@
/**
 * Created by jlchen on 11/11/21.
 *
 * @Description : HDLAuthSocket
 * @Description : HDLAuthSocket 由于前期已经命名好，不做更改，可用作Udp服务端使用
 */
public class HDLAuthSocket {
    private static final String TAG = "HDLAuth";
@@ -55,7 +52,10 @@
     * udp默认端口
     */
    private static final int UDP_PORT = 8585;
    private static SocketBoot updBoot;
    /**
     * 因为考虑到使用一个端口，要支持接收多网关的数据，所以只允许使用一个
     */
    private static UdpSocketBoot udpSocketBoot;
    //    private EventListener authEvent;
    //搜索网关
    private EventListener searchGatewayEvent;
@@ -71,6 +71,7 @@

    private HDLAuthSocket() {
        initSearchGatewayEvent();
        initListenerGatewayEvent();
    }

//    public interface CallBack extends BaseCallBack {
@@ -87,30 +88,39 @@
            synchronized (HDLLinkConfig.class) {
                if (instance == null) {
                    instance = new HDLAuthSocket();
                    //1.启动Socket 开启监听
                    instance.getUdpBoot();
                }
            }
        }
        return instance;
    }

    private SocketOptions getUdpOptions() {
        final SocketOptions options = new SocketOptions();
    private UdpSocketOptions getUdpOptions() {
        final UdpSocketOptions options = new UdpSocketOptions();
        final MessagePipeLine pipeLine = new MessagePipeLine();
        pipeLine.add(new LinkMessageDecoder());
        pipeLine.add(new LinkMessageEncoder());
        options.setHandleMessage(pipeLine);
        options.setHeartbeatTimeInterval(10*1000L);
        options.setHeartbeatData("UDP");
        options.setEnabledHeartbeat(true);//是否开启心跳包发送检测
        return options;
    }

    private SocketBoot getUdpBoot(String ip) {
        if (updBoot == null) {
            updBoot = UdpClient.init(ip, UDP_PORT, getUdpOptions());
            updBoot.connect();

    /**
     * 获取当前udp对象，如果不存在就创建
     * @return 返回当前对象
     */
    private UdpSocketBoot getUdpBoot() {
        try {
            if (udpSocketBoot == null) {
                udpSocketBoot = UdpClient.init(UDP_PORT, getUdpOptions());
                udpSocketBoot.bind();
            }
        } catch (Exception e) {
            return null;
        }
        return updBoot;

        return udpSocketBoot;
    }

    /**
@@ -121,8 +131,6 @@
     */
    public void startAuthenticateRequest(AuthenticateRequest request, HDLLinkCallBack callBack) {
        HDLLinkConfig.getInstance().clearConfig();
        //1.启动Socket 开启监听
        getUdpBoot(IpUtils.getBroadcastAddress());
        //2.构建监听Listener
//        authEvent =
        //3.监听网关广播的入网指令
@@ -168,7 +176,7 @@
        String requestStr = gs.toJson(request);
        LinkRequest message = new LinkRequest(topic, requestStr);

        HdlSocketHelper.send(getUdpBoot(ip), message, new HdlSocketHelper.HdlSocketListener() {
        HdlSocketHelper.send(getUdpBoot(), InetSocketAddress.createUnresolved(ip,UDP_PORT), message, new HdlSocketHelper.HdlSocketListener() {
            @Override
            public void onSucceed(Object msg) {
                if (callBack == null) return;
@@ -284,7 +292,7 @@
     * @param callBack 回调
     */
    public void searchGatewayMulticast(SearchGatewayCallBack callBack) {
        searchGateway(HDLLinkConfig.getInstance().getGatewayId(), UDP_GROUP_IP, callBack);
        searchGateway(HDLLinkConfig.getInstance().getGatewayId(),  InetSocketAddress.createUnresolved(UDP_GROUP_IP,UDP_PORT), callBack);
    }

    /**
@@ -294,7 +302,7 @@
     */
    public void searchGatewayBroadcast(SearchGatewayCallBack callBack) {
        String ip = IpUtils.getBroadcastAddress();
        searchGateway(HDLLinkConfig.getInstance().getGatewayId(), ip, callBack);
        searchGateway(HDLLinkConfig.getInstance().getGatewayId(),  InetSocketAddress.createUnresolved(ip,UDP_PORT), callBack);
    }


@@ -312,7 +320,7 @@
                GsonConvert.getGson().toJson(data));

        String ip = IpUtils.getBroadcastAddress();
        HdlSocketHelper.send(getUdpBoot(ip), message, new HdlSocketHelper.HdlSocketListener() {
        HdlSocketHelper.send(getUdpBoot(), InetSocketAddress.createUnresolved(ip,UDP_PORT), message, new HdlSocketHelper.HdlSocketListener() {
            @Override
            public void onSucceed(Object msg) {
                if (callBack == null) return;
@@ -345,7 +353,7 @@
        }
        LinkRequest message = new LinkRequest(topic, bodyStr);
        String ip = IpUtils.getBroadcastAddress();
        HdlSocketHelper.send(getUdpBoot(ip), message, new HdlSocketHelper.HdlSocketListener() {
        HdlSocketHelper.send(getUdpBoot(), InetSocketAddress.createUnresolved(ip,UDP_PORT), message, new HdlSocketHelper.HdlSocketListener() {
                    @Override
                    public void onSucceed(Object msg) {
                        if (callBack == null) return;
@@ -374,7 +382,7 @@
        }
        LinkRequest message = new LinkRequest(topic, bodyStr);
        String ip = IpUtils.getBroadcastAddress();
        getUdpBoot(ip).sendMsg(message.getSendBytes());
        getUdpBoot().sendMsg( InetSocketAddress.createUnresolved(ip,UDP_PORT),message.getSendBytes());
    }

    private GatewaySearchBean getGatewaySearchBean(Object msg) {
@@ -433,7 +441,7 @@
    private SearchGatewayCallBack mSearchGatewayCallBack;

    private void initSearchGatewayEvent() {
        LogUtils.i("搜索网关--", "initSearchGatewayEvent");
        LogUtils.i("搜索网关", "initSearchGatewayEvent");
        searchGatewayEvent = new EventListener() {
            @Override
            public void onMessage(Object msg) {
@@ -441,7 +449,6 @@
                    if (msg instanceof LinkResponse) {
                        LinkResponse linkResponse = (LinkResponse) msg;
                        String data = linkResponse.getData();
                        LogUtils.i("搜索到网关->" + data);
                        if (!TextUtils.isEmpty(data)) {
                            final BaseLocalResponse<GatewaySearchBean> response = GsonConvert.getGson().fromJson(data, new TypeToken<BaseLocalResponse<GatewaySearchBean>>() {
                            }.getType());
@@ -467,14 +474,44 @@
        };
    }

    private void initListenerGatewayEvent() {
        LogUtils.i( "初始化永久监听网关广播事件，不用移除此事件");
        EventListener  gatewayEvent = new EventListener() {
            @Override
            public void onMessage(Object msg) {
                try {
                    if (msg instanceof LinkResponse) {
                        LinkResponse linkResponse = (LinkResponse) msg;
                        String data = linkResponse.getData();
                        LogUtils.i("接收到网关信息：" + data);
                        if (!TextUtils.isEmpty(data)) {
                            final BaseLocalResponse<GatewaySearchBean> response = GsonConvert.getGson().fromJson(data, new TypeToken<BaseLocalResponse<GatewaySearchBean>>() {
                            }.getType());
                            GatewaySearchBean searchBean = response.getObjects();
                            if (searchBean != null && !TextUtils.isEmpty(searchBean.getGatewayId())) {
                                if (searchBean.getGatewayId().equals(HDLLinkConfig.getInstance().getGatewayId())) {
                                    HDLLinkConfig.getInstance().setCurrentGateway(searchBean);//设置当前网关
                                    HDLLinkConfig.getInstance().setLocalEncrypt(searchBean.isLocalEncrypt());//设置是否加密
                                }
                            }
                        }
                    }
                } catch (Exception e) {

                }
            }
        };
        EventDispatcher.getInstance().registerIo(TopicConstant.GATEWAY_SEARCH_REPLY, gatewayEvent);
    }

    /**
     * 搜索指定网关是否在线，搜索到则返回指定的网关对象
     *
     * @param gatewayId 网关id
     * @param ip        接收目标的ip地址
     * @param desInetSocketAddress        目标的地址
     * @param callBack  回调
     */
    public void searchGateway(String gatewayId, String ip, SearchGatewayCallBack callBack) {
    public void searchGateway(String gatewayId, InetSocketAddress desInetSocketAddress, SearchGatewayCallBack callBack) {
        this.searchGatewayId = gatewayId;
        this.mSearchGatewayCallBack = callBack;
        //重置参数
@@ -495,8 +532,8 @@
                    try {
                        //搜索网关
                        searchGatewayCount.set(searchGatewayCount.get() + 1);
                        LogUtils.i("搜索网关--", "搜索网关第" + searchGatewayCount.get() + "次");
                        getUdpBoot(ip).sendMsg(message.getSendBytes());
                        LogUtils.i("搜索网关", "搜索网关第" + searchGatewayCount.get() + "次");
                        getUdpBoot().sendMsg(desInetSocketAddress,message.getSendBytes());
                        Thread.sleep(1000L);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
@@ -517,7 +554,7 @@
     * 注册搜索网关监听
     */
    private void registerSearchGatewayEvent() {
        LogUtils.i("搜索网关--", "注册搜索网关监听");
        LogUtils.i("搜索网关", "注册搜索网关监听");
        EventDispatcher.getInstance().registerIo(TopicConstant.GATEWAY_SEARCH_REPLY, searchGatewayEvent);
    }

@@ -525,7 +562,7 @@
     * 移除搜索网关监听
     */
    private void removeSearchGatewayEvent() {
        LogUtils.i("搜索网关--", "移除搜索网关监听");
        LogUtils.i("搜索网关", "移除搜索网关监听");
        EventDispatcher.getInstance().remove(TopicConstant.GATEWAY_SEARCH_REPLY, searchGatewayEvent);
    }


 HDLSDK/hdl-connect/src/main/java/com/hdl/sdk/connect/socket/HDLSocket.java

@@ -50,40 +50,14 @@
 */

public class HDLSocket {



//    public interface CallBack extends BaseCallBack {

//        void onResponse(String data);

//    }

//

//    /**

//     * udp默认组播ip

//     */

//    private static final String UDP_GROUP_IP = "239.0.168.188";



    /**

     * udp默认端口

     */

    private static final int UDP_PORT = 8585;



    /**

     * tcp默认端口

     */

    private static final int TCP_PORT = 8586;



    //    private String tcpIp;

//    private int tcpPort;



//    private int udpPort;

//    private String udpIp;



//    private static SocketBoot updBoot;

    private SocketBoot tcpBoot;



    private ConnectStatusListener statusListener;



//    private EventListener searchEvent;

//    private CallBack searchCallBack;

//    private ScheduledExecutorService searchGatewayThread;

//    private final AtomicInteger searchCount = new AtomicInteger(0);



    private HDLSocket() {

        statusListener = new ConnectStatusListener() {

@@ -102,43 +76,6 @@


            }

        };

//        searchEvent = new EventListener() {

//            @Override

//            public void onMessage(Object msg) {

//                try {

//                    if (msg instanceof LinkResponse) {

//                        LinkResponse linkResponse = (LinkResponse) msg;

//                        String data = linkResponse.getData();

//                        if (!TextUtils.isEmpty(data)) {

//                            final BaseLocalResponse<GatewaySearchBean> response = GsonConvert.getGson().fromJson(data, new TypeToken<BaseLocalResponse<GatewaySearchBean>>() {

//                            }.getType());

//                            GatewaySearchBean searchBean = response.getObjects();

//                            if (searchBean != null) {

////                                gatewayId = searchBean.getGatewayId();

////                                if (!TextUtils.isEmpty(gatewayId)) {

////                                    SPUtils.put(GATEWAY_KEY, gatewayId);

////                                }

////                                tcpIp = searchBean.getIp_address();

////                                if (!TextUtils.isEmpty(tcpIp)) {

////                                    SPUtils.put(TCP_IP_KEY, tcpIp);

////                                }

//                            }

//

//                            if (searchCallBack != null) {

//                                searchCallBack.onResponse(linkResponse.toString());

//                            }

//                        }

//

//                    }

//                } catch (Exception e) {

//                    if (searchCallBack != null) {

//                        searchCallBack.onError("解析失败");

//                    }

//                }

//

//

//            }

//        };

    }



    private static class SingletonInstance {

@@ -149,16 +86,6 @@
        return SingletonInstance.INSTANCE;

    }





//    private SocketOptions getUdpOptions() {

//        final SocketOptions options = new SocketOptions();

//        final MessagePipeLine pipeLine = new MessagePipeLine();

//        pipeLine.add(new LinkMessageDecoder());

//        pipeLine.add(new LinkMessageEncoder());

//        options.setHandleMessage(pipeLine);

//        options.setEnabledHeartbeat(false);

//        return options;

//    }



    private SocketOptions getTcpOptions() {

        final SocketOptions options = new SocketOptions();

@@ -173,9 +100,6 @@
        return options;

    }



    private int getUdpPort() {

        return UDP_PORT;

    }



    public int getTcpPort() {

        return TCP_PORT;

@@ -188,90 +112,6 @@
    public String getGatewayId() {

        return HDLLinkConfig.getInstance().getGatewayId();

    }





//    private String getUdpIp() {

//        if (TextUtils.isEmpty(udpIp)) {

//            udpIp = UDP_GROUP_IP;

//        }

//        return udpIp;

//    }



//    public void searchGateway() {

//        searchGateway(null);

//    }

//

//    /**

//     * 组播搜索

//     */

//    public void searchGateway(CallBack callBack) {

////        gatewayId = "";//重置网关ID

////        this.searchCallBack = callBack;

////

////        if (searchGatewayThread != null) {

////            searchGatewayThread.shutdownNow();

////        }

////

////

////        new Thread(new Runnable() {

////            @Override

////            public void run() {

////                while (TextUtils.isEmpty(gatewayId)) {

////                    //搜索网关

////                    searchGateway(IdUtils.getUUId(), searchEvent);

////                    try {

////                        Thread.sleep(1000L);

////                    } catch (InterruptedException e) {

////                        e.printStackTrace();

////                    }

////                }

////            }

////        }).start();

//    }

//

//

//    /**

//     * 通过组播搜索网关

//     */

//    public void searchGateway(String msgId, EventListener eventListener) {

//        searchGateway(getUdpIp(), getUdpPort(), msgId, eventListener);

//    }

//

//    /**

//     * 通过广播搜索网关

//     */

//    public void searchGatewayByBroadcast(String msgId, EventListener eventListener) {

//        searchGateway(IpUtils.getBroadcastAddress(), getUdpPort(), msgId, eventListener);

//    }

//

//    /**

//     * 默认是组播搜索网关

//     */

//    public void searchGateway(String ip, int port, String msgId, EventListener eventListener) {

//

//        if (updBoot == null) {

//            updBoot = UdpClient.init(ip, port, getUdpOptions());

//            updBoot.connect();

//        }

//

//

//        String time = String.valueOf(System.currentTimeMillis());

//        JsonObject jsonObject = new JsonObject();

//        jsonObject.addProperty("id", msgId);

//        jsonObject.addProperty("time_stamp", time);

//

//        EventDispatcher.getInstance().registerIo(TopicConstant.GATEWAY_SEARCH_REPLY, eventListener);

//        LinkRequest message = new LinkRequest(TopicConstant.GATEWAY_SEARCH,

//                jsonObject.toString());

//

//        try {

//            updBoot.sendMsg(message.getSendBytes());

//        } catch (UnsupportedEncodingException e) {

//            e.printStackTrace();

//        }

//

//    }





    /**

     * 获取设备列表


 HDLSDK/hdl-connect/src/main/java/com/hdl/sdk/connect/socket/HdlSocketHelper.java

@@ -7,7 +7,9 @@
import com.hdl.sdk.common.utils.ThreadToolUtils;

import com.hdl.sdk.connect.bean.LinkRequest;

import com.hdl.sdk.socket.SocketBoot;

import com.hdl.sdk.socket.udp.UdpSocketBoot;



import java.net.InetSocketAddress;

import java.util.concurrent.ScheduledExecutorService;

import java.util.concurrent.TimeUnit;

import java.util.concurrent.atomic.AtomicBoolean;

@@ -25,15 +27,17 @@
    private final Long sendAwaitTime;

    private final int maxRetry;



    private final SocketBoot boot;

    private SocketBoot boot;

    private UdpSocketBoot udpSocketBoot;

    private InetSocketAddress inetSocketAddress;

    private final LinkRequest linkRequest;

    private final EventListener eventListener;



    private final AtomicInteger sendNumber;



    private final AtomicBoolean isSend = new AtomicBoolean();

    private final AtomicBoolean isSend = new AtomicBoolean(false);



    private final HdlSocketListener listener;

    private HdlSocketListener listener;



    private ScheduledExecutorService sendThread;



@@ -66,12 +70,37 @@
        EventDispatcher.getInstance().register(observeTopic, eventListener);

    }



    private HdlSocketHelper(Long sendAwaitTime, int maxRetry, UdpSocketBoot udpSocketBoot,

                            InetSocketAddress inetSocketAddress, LinkRequest linkRequest, String observeTopic, HdlSocketListener listener) {

        this.sendAwaitTime = sendAwaitTime;

        this.maxRetry = maxRetry;

        this.udpSocketBoot = udpSocketBoot;

        this.inetSocketAddress = inetSocketAddress;

        this.linkRequest = linkRequest;

        this.listener = listener;

        this.sendNumber = new AtomicInteger(0);

        eventListener = new EventListener() {

            @Override

            public void onMessage(Object msg) {

                isSend.set(true);

                if (listener != null) {

                    listener.onSucceed(msg);

                }

                if (sendThread != null) {

                    sendThread.shutdownNow();

                }

                EventDispatcher.getInstance().remove(eventListener);

            }

        };

        EventDispatcher.getInstance().register(observeTopic, eventListener);

    }



    public static void send(SocketBoot boot, LinkRequest linkRequest, String observeTopic, HdlSocketListener listener, Long sendAwaitTime, int maxRetry) {

        if (TextUtils.isEmpty(observeTopic)) {

            observeTopic = linkRequest.getTopic() + "_reply";

        }

        HdlSocketHelper socketHelper = new HdlSocketHelper(sendAwaitTime, maxRetry, boot, linkRequest, observeTopic, listener);

        socketHelper.resend();

        socketHelper.send();

    }



    public static void send(SocketBoot boot, LinkRequest linkRequest, String observeTopic, HdlSocketListener listener) {

@@ -87,34 +116,78 @@
        send(boot, linkRequest, "", listener, DEF_SEND_TIMEOUT, DEF_SEND_ONE);

    }



    private void resend() {

    /**

     * Udp的发送方法

     *

     * @param udpSocketBoot     Udp当前对接

     * @param inetSocketAddress 发送的目标地址

     * @param linkRequest       发送的数据

     * @param observeTopic      发送的主题

     * @param listener          回调

     */

    public static void send(UdpSocketBoot udpSocketBoot, InetSocketAddress inetSocketAddress, LinkRequest linkRequest, String observeTopic, HdlSocketListener listener) {

        if (TextUtils.isEmpty(observeTopic)) {

            observeTopic = linkRequest.getTopic() + "_reply";

        }

        HdlSocketHelper socketHelper = new HdlSocketHelper(DEF_SEND_TIMEOUT, DEF_SEND_ONE, udpSocketBoot, inetSocketAddress, linkRequest, observeTopic, listener);

        socketHelper.send();

    }



    /**

     * Udp发送的方法

     *

     * @param udpSocketBoot     当前Udp对象

     * @param inetSocketAddress 发送的目标地址

     * @param linkRequest       发送的对象

     * @param listener          回调

     */

    public static void send(UdpSocketBoot udpSocketBoot, InetSocketAddress inetSocketAddress, LinkRequest linkRequest, HdlSocketListener listener) {

        String observeTopic = null;

        if (TextUtils.isEmpty(observeTopic)) {

            observeTopic = linkRequest.getTopic() + "_reply";

        }

        HdlSocketHelper socketHelper = new HdlSocketHelper(DEF_SEND_TIMEOUT, DEF_SEND_ONE, udpSocketBoot, inetSocketAddress, linkRequest, observeTopic, listener);

        socketHelper.send();

    }



    private void send() {

        getSendThread().scheduleWithFixedDelay(new Runnable() {

            @Override

            public void run() {

                if ((sendNumber.get() < maxRetry + 2) || !isSend.get()) {

                //发送次数小于重发次数

                if ((sendNumber.get() < maxRetry)) {

                    try {

                        if (sendNumber.get() < maxRetry + 1) {

                        //还没有收到回复，再发送

                        if (!isSend.get()) {

                            sendNumber.set(sendNumber.get() + 1);

                            //如是tcp

                            if (boot != null) {

                                boot.sendMsg(linkRequest.getSendBytes());

                            }

                            sendNumber.set(sendNumber.get() + 1);

                        } else {

                            notifyFailure();

                            //如果是udp

                            if (null != udpSocketBoot) {

                                udpSocketBoot.sendMsg(inetSocketAddress, linkRequest.getSendBytes());

                            }

                        }

                    } catch (Exception e) {

                        e.printStackTrace();

                        sendNumber.set(sendNumber.get() + 1);

                    } finally {

                        if (sendNumber.get() > maxRetry + 1 && !isSend.get()) {

                            notifyFailure();

                        }

                    }

                } else {

                    //超出重发次数并没有收到回复

                    if (!isSend.get()) {

                        notifyFailure();

                    }

                }

            }

        }, sendAwaitTime, sendAwaitTime, TimeUnit.MILLISECONDS);

    }



    public ScheduledExecutorService getSendThread() {

    /**

     * 获取发送线程

     *

     * @return 返回获取到的线程

     */

    private ScheduledExecutorService getSendThread() {

        if (sendThread == null) {

            sendThread = ThreadToolUtils.getInstance().newScheduledThreadPool(1);

        }

@@ -122,14 +195,18 @@
    }





    /**

     * 发送失败

     */

    private void notifyFailure() {

        EventDispatcher.getInstance().remove(eventListener);

        if (sendThread != null) {

            sendThread.shutdownNow();

            sendThread = null;

        }

        if (listener != null) {

            listener.onFailure();

            listener = null;

        }



        EventDispatcher.getInstance().remove(eventListener);

    }

}


 HDLSDK/hdl-socket/src/main/java/com/hdl/sdk/socket/SocketBoot.java

@@ -127,12 +127,10 @@
            public void run() {

                while (isRun.get()) {

                    if (client.isConnect()) {

                        LogUtils.d("=====", "==发送数据==");



                        try {

                            SocketRequest socketRequest = mMessageQueue.take();

                            final String sendStr = new String(socketRequest.getData(), 0, socketRequest.getData().length);

                            LogUtils.d("=====", "==发送数据==："+sendStr);

                            LogUtils.i("发送数据："+sendStr);

                            final String action = socketRequest.getAction();

                            try {

                                client.sendMsg(socketRequest.getData());

@@ -172,28 +170,30 @@


    }



    /**

     * 初始化接收线程，只需要初始化一次

     */

    public void initReceiveThread() {

        if (receiveThread == null) {

            receiveThread = ThreadToolUtils.getInstance().newFixedThreadPool(1);

        }

        receiveThread.execute(new Runnable() {

            @Override

            public void run() {

                while (isRun.get()) {

                    if (client.isConnect()) {

                        try {

                            //读取数据

                            client.onHandleResponse();

                        } catch (Exception e) {

                            e.printStackTrace();

                            LogUtils.d("====", "断开连接" + e.getMessage());

                            disconnectError();

            receiveThread.execute(new Runnable() {

                @Override

                public void run() {

                    while (isRun.get()) {

                        if (client.isConnect()) {

                            try {

                                //读取数据

                                client.onHandleResponse();

                            } catch (Exception e) {

                                e.printStackTrace();

                                disconnectError();

                            }

                        }

                    }

                    LogUtils.i("Socket 线程退出接收数据");

                }

            }

        });



            });

        }

    }





@@ -258,7 +258,6 @@
        if (!client.isConnect()) {

            resetConnect(false);

        }



    }



    /**

@@ -278,7 +277,6 @@
                    if (!client.isConnect() && isOpenRetry.get()) {

                        resetConnect(false);

                    }



                }

            }, 3000, TimeUnit.MILLISECONDS);

        }


 HDLSDK/hdl-socket/src/main/java/com/hdl/sdk/socket/SocketPool.java

@@ -35,7 +35,7 @@
        mUdpClientPool.clear();

    }



    public synchronized DatagramSocket getUdpSocket(InetSocketAddress address) throws SocketException {

    public synchronized DatagramSocket getUdpSocket1(InetSocketAddress address) throws SocketException {

        DatagramSocket socket;

        final String key = address.toString();

        if (mUdpClientPool.containsKey(key)) {

@@ -51,5 +51,4 @@
        mUdpClientPool.put(address.toString(), socket);

        return socket;

    }



}


 HDLSDK/hdl-socket/src/main/java/com/hdl/sdk/socket/client/IUdpClient.java

New file
@@ -0,0 +1,40 @@
package com.hdl.sdk.socket.client;

import com.hdl.sdk.socket.SocketOptions;
import com.hdl.sdk.socket.udp.UdpSocketOptions;

import java.net.InetSocketAddress;

/**
 * Created by hxb on 2021/12/12.
 */
public interface IUdpClient {

    /**
     * 绑定端口
     */
    void bind() throws Exception;

    /**
     * 关闭当前连接
     * @return
     */
    boolean close();


    UdpSocketOptions getOptions();

    /**
     * 监听数据
     */
    void onHandleResponse() throws Exception;

    /**
     * 发送数据
     * @param inetSocketAddress 目标地址
     * @param msg 发送数据
     * @throws Exception
     */
    void sendMsg(InetSocketAddress inetSocketAddress, byte[] msg) throws Exception;

}

 HDLSDK/hdl-socket/src/main/java/com/hdl/sdk/socket/client/UdpClient.java

@@ -1,135 +1,138 @@
package com.hdl.sdk.socket.client;





import android.util.Log;

import android.text.TextUtils;



import com.hdl.sdk.common.HDLSdk;

import com.hdl.sdk.common.utils.IpUtils;

import com.hdl.sdk.common.utils.LogUtils;

import com.hdl.sdk.common.utils.ThreadToolUtils;

import com.hdl.sdk.socket.SocketBoot;

import com.hdl.sdk.socket.SocketOptions;

import com.hdl.sdk.socket.SocketPool;

import com.hdl.sdk.socket.annotation.ConnectStatus;

import com.hdl.sdk.socket.udp.UdpSocketBoot;

import com.hdl.sdk.socket.udp.UdpSocketOptions;

import com.hdl.sdk.socket.codec.IHandleMessage;

import com.hdl.sdk.socket.listener.ConnectStatusListener;



import java.io.IOException;

import java.net.DatagramPacket;

import java.net.DatagramSocket;

import java.net.InetAddress;

import java.net.InetSocketAddress;

import java.net.SocketAddress;

import java.util.List;

import java.util.concurrent.atomic.AtomicBoolean;





/**

 * Created by Tong on 2021/9/15.

 * 组播需要android.permission.CHANGE_WIFI_MULTICAST_STATE权限

 * MulticastSocket

 * Created by hxb on 2021/12/12.

 */

public class UdpClient implements IClient {



    private static DatagramSocket mSocket;



    private DatagramPacket receivePacket;



    private final int BUFFER = 4 * 1024;



    private final byte[] receiveByte;



    private final String ip;



    private final int port;



    private int monitorPort;

    private int sendPort;



    private SocketOptions socketOptions;



    private final AtomicBoolean isConnect = new AtomicBoolean(false);

public class UdpClient implements IUdpClient {



    /**

     * @param sendPort -1 表示随机端口

     * 当前socket

     */

    private UdpClient(String ip, int port, int monitorPort, int sendPort, SocketOptions socketOptions) {

    private  DatagramSocket mSocket;



    /**

     * 接收数据包

     */

    private DatagramPacket receivePacket;



    /**

     * 缓冲区大小

     */

    private final int BUFFER = 2 * 1024;



    /**

     * 本地监听IP地址

     */

    private String ipAddress;

    /**

     * 本地监听端口

     */

    private int port;



    /**

     * socket配置信息

     */

    private UdpSocketOptions socketOptions;



    /**

     * 初始化参数

     * @param ipAddress 本地监听端口

     * @param port 本地监听端口

     * @param socketOptions

     */

    private UdpClient(String ipAddress,int port, UdpSocketOptions socketOptions) {

        this.ipAddress = ipAddress;

        this.port = port;

        this.socketOptions = socketOptions;

        this.ip = ip;

        this.port = port;

        this.sendPort = sendPort;

        this.monitorPort = monitorPort;

        this.receiveByte = new byte[BUFFER];

        byte[] receiveByte = new byte[BUFFER];

        receivePacket = new DatagramPacket(receiveByte, receiveByte.length);

    }



    public UdpClient(String ip, int port) {

        this.ip = ip;

        this.port = port;

        this.receiveByte = new byte[BUFFER];

    /**

     * 初始化参数

     * @param ipAddress 本地监听IP地址

     * @param port 本地监听端口

     * @param options

     * @return

     */

    public static UdpSocketBoot init(String ipAddress, int port, UdpSocketOptions options) {

        return new UdpSocketBoot(new UdpClient(ipAddress, port, options));

    }



    public static SocketBoot init(String ip, int port, int monitorPort, int sendPort, SocketOptions options) {

        return new SocketBoot(new UdpClient(ip, port, monitorPort, sendPort, options));

    /**

     * 初始化参数

     * @param port 本地监听端口

     * @param options

     * @return

     */

    public static UdpSocketBoot init(int port, UdpSocketOptions options) {

        return init(null, port, options);

    }



    public static SocketBoot init(String ip, int port, int monitorPort, SocketOptions options) {

        return init(ip, port, monitorPort, -1, options);

    }



    public static SocketBoot init(String ip, int port, SocketOptions options) {

        return init(ip, port, port, -1, options);

    }



    @Override

    public void connect() throws Exception {

    public void bind() throws Exception {

        try {

            mSocket = SocketPool.getInstance().getUdpSocket(new InetSocketAddress(monitorPort));



            //已经绑定过就不用再绑定

            if (null != mSocket) {

                return;

            }

            if (TextUtils.isEmpty(ipAddress)) {

                mSocket = SocketPool.getInstance().getUdpSocket1(new InetSocketAddress(port));

            }

            mSocket.setBroadcast(true);

            mSocket.setReuseAddress(true);

            isConnect.set(true);

            if (receivePacket == null) {

                receivePacket = new DatagramPacket(receiveByte, BUFFER);

            }



        } catch (Exception e) {

            isConnect.set(false);

            LogUtils.e("初始化Socket 失败：" + e.getMessage());

            throw e;

        }





    }



    @Override

    public void disconnect() {

        if (mSocket != null) {

    public boolean close() {

        try {

            mSocket.close();

        } catch (Exception e) {



        }

        isConnect.set(false);

        mSocket = null;

        return true;

    }



    @Override

    public boolean isConnect() {

        return isConnect.get();

    public UdpSocketOptions getOptions() {

        return null;

    }



    @Override

    public synchronized SocketOptions getOptions() {

        if (socketOptions == null) {

            socketOptions = new SocketOptions();

        }

        return socketOptions;

    }



    @Override

    public void onHandleResponse() throws Exception {

        if (receivePacket == null || mSocket == null) {

        if (mSocket == null) {

            return;

        }

        try {

           mSocket.receive(receivePacket);

        } catch (IOException e) {

            e.printStackTrace();

            isConnect.set(false);

        }

        if (receivePacket.getLength() == 0) {

            return;

@@ -142,68 +145,27 @@


        }



        IHandleMessage handleMessage = getOptions().getHandleMessage();

        if (handleMessage != null) {

            byte[] data = new byte[receivePacket.getLength()];

            System.arraycopy(receivePacket.getData(), 0, data, 0, data.length);

            handleMessage.read(data);

        }

        final String receive = new String(receivePacket.getData(), 0, receivePacket.getLength());

        try {

            LogUtils.i( "接收到Udp数据包，网络地址：" + receivePacket.getAddress().getHostAddress() + ":" + receivePacket.getPort());



        LogUtils.d("---->", receive + " from " + receivePacket.getAddress().getHostAddress() + ":" + receivePacket.getPort());



        //重置长度

        if (receivePacket != null) {

            receivePacket.setLength(BUFFER);

        }

    }



    @Override

    public void sendMsg(byte[] msg) throws Exception {

        if (msg == null) {

            msg = new byte[1];

        }

        InetAddress serverAddress = InetAddress.getByName(ip);

        final DatagramPacket sendPacket = new DatagramPacket(msg, msg.length, serverAddress, port);

        if (sendPort < 0) {

            final DatagramSocket sendSocket = new DatagramSocket();

            sendSocket.send(sendPacket);

            sendSocket.close();

        } else if (sendPort == monitorPort) {

            mSocket.send(sendPacket);

        } else {

            final DatagramSocket sendSocket = new DatagramSocket(sendPort);

            sendSocket.send(sendPacket);

            sendSocket.close();

        }



    }



    @Override

    public void onConnectStatus(int status) {

        ThreadToolUtils.getInstance().runOnUiThread(new Runnable() {

            @Override

            public void run() {

                final List<ConnectStatusListener> list = getOptions().getConnectStatusListener();

                if (list != null && !list.isEmpty()) {

                    for (ConnectStatusListener listener : list) {

                        switch (status) {

                            case ConnectStatus

                                    .CONNECTING:

                                listener.onConnecting();

                                break;

                            case ConnectStatus

                                    .CONNECTED:

                                listener.onConnected();

                                break;

                            case ConnectStatus

                                    .DISCONNECT:

                                listener.onConnectFailed();

                                break;

                        }

                    }

                }

            IHandleMessage handleMessage = getOptions().getHandleMessage();

            if (handleMessage != null) {

                byte[] data = new byte[receivePacket.getLength()];

                System.arraycopy(receivePacket.getData(), 0, data, 0, data.length);

                handleMessage.read(data);

            }

        });



        }catch (Exception e){



        }

    }



    @Override

    public void sendMsg(InetSocketAddress inetSocketAddress, byte[] msg) throws Exception {

        if (msg == null) {

            return;

        }

        final DatagramPacket sendPacket = new DatagramPacket(msg, msg.length, inetSocketAddress);

        mSocket.send(sendPacket);

    }

}


 HDLSDK/hdl-socket/src/main/java/com/hdl/sdk/socket/udp/UdpSocketBoot.java

New file
@@ -0,0 +1,103 @@
package com.hdl.sdk.socket.udp;



import android.text.TextUtils;



import androidx.collection.ArrayMap;



import com.hdl.sdk.common.utils.LogUtils;

import com.hdl.sdk.common.utils.ThreadToolUtils;

import com.hdl.sdk.socket.SocketRequest;

import com.hdl.sdk.socket.annotation.ConnectStatus;

import com.hdl.sdk.socket.client.IClient;

import com.hdl.sdk.socket.client.IUdpClient;

import com.hdl.sdk.socket.listener.SendListener;



import java.net.ConnectException;

import java.net.InetSocketAddress;

import java.util.concurrent.BlockingQueue;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.LinkedBlockingDeque;

import java.util.concurrent.ScheduledExecutorService;

import java.util.concurrent.TimeUnit;

import java.util.concurrent.atomic.AtomicBoolean;

import java.util.concurrent.atomic.AtomicInteger;



/**

 * Created by hxb on 2021/12/12.

 */

public class UdpSocketBoot {



    private final IUdpClient client;



    private final AtomicBoolean isOpenRetry = new AtomicBoolean(false);



    private final AtomicInteger resendCount = new AtomicInteger(0);



    private final ArrayMap<String, SendListener> sendMap = new ArrayMap<>();



    public UdpSocketBoot(IUdpClient client) {

        this.client = client;

    }



    /**

     * 绑定 socket

     * @throws Exception 可能端口冲突

     */

    public void bind() throws Exception {

        if (null != client) {

            client.bind();

        }

    }



    /**

     * 发送数据

     * @param inetSocketAddress 目的的IP地址

     * @param msg 发送数据

     * @param listener 发送回调

     */

    public void sendMsg(InetSocketAddress inetSocketAddress,byte[] msg, SendListener listener) {

        sendMsg(inetSocketAddress, msg, true, listener);

    }



    /**

     * 发送数据

     * @param inetSocketAddress 目的的IP地址

     * @param msg 发送数据

     */

    public void sendMsg(InetSocketAddress inetSocketAddress,byte[] msg) {

        sendMsg(inetSocketAddress, msg, true, null);

    }



    /**

     * 发送数据

     * @param inetSocketAddress 目的IP地址

     * @param msg 发送的数据

     * @param isRefreshRetry 是否要重发

     * @param listener 发送回调

     */

    public void sendMsg(InetSocketAddress inetSocketAddress, byte[] msg, boolean isRefreshRetry, SendListener listener) {

        if (isRefreshRetry) {

            //重置连接次数

            resendCount.set(0);

        }

        try {

            SocketRequest request = new SocketRequest(msg);

            if (listener != null && !TextUtils.isEmpty(request.getAction())) {

                sendMap.put(request.getAction(), listener);

            }

            client.sendMsg(inetSocketAddress, msg);

        } catch (Exception e) {

            LogUtils.i("发送失败:" + e.getMessage());

        }

    }



    /**

     * 关闭当前socket

     */

    public synchronized void close() {

        isOpenRetry.set(false);



        sendMap.clear();

        client.close();

    }

}


 HDLSDK/hdl-socket/src/main/java/com/hdl/sdk/socket/udp/UdpSocketOptions.java

New file
@@ -0,0 +1,26 @@
package com.hdl.sdk.socket.udp;





import com.hdl.sdk.socket.codec.IHandleMessage;

import com.hdl.sdk.socket.listener.ConnectStatusListener;



import java.io.UnsupportedEncodingException;

import java.util.ArrayList;

import java.util.List;



/**

 * Created by hxb on 2021/12/12.

 */

public class UdpSocketOptions {



    //处理数据

    private IHandleMessage handleMessage;



    public IHandleMessage getHandleMessage() {

        return handleMessage;

    }



    public void setHandleMessage(IHandleMessage handleMessage) {

        this.handleMessage = handleMessage;

    }

}