From 8d68f7865f44e21876488ef21b058e7f05f53af1 Mon Sep 17 00:00:00 2001
From: chenfufeng <chenfufeng@zhidaoauto.com>
Date: Fri, 25 Feb 2022 15:10:38 +0800
Subject: [PATCH] =?UTF-8?q?[Fix]=E8=A7=A3=E5=86=B3merge=E4=BA=A7=E7=94=9F?=
 =?UTF-8?q?=E7=9A=84issue?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/src/main/res/layout/activity_main.xml     |   6 +
 gradle.properties                             |   5 +-
 .../netty/MogoProtocolServerHandler.java      |  69 ++++++
 .../server/netty/NettyServerListener.java     |  43 ++++
 .../server/netty/NettyTcpServer.java          | 207 ++++++++++++++++++
 5 files changed, 328 insertions(+), 2 deletions(-)
 create mode 100644 libraries/mogo-telematic/src/main/java/com/mogo/telematic/server/netty/MogoProtocolServerHandler.java
 create mode 100644 libraries/mogo-telematic/src/main/java/com/mogo/telematic/server/netty/NettyServerListener.java
 create mode 100644 libraries/mogo-telematic/src/main/java/com/mogo/telematic/server/netty/NettyTcpServer.java

diff --git a/app/src/main/res/layout/activity_main.xml b/app/src/main/res/layout/activity_main.xml
index 8b9a3d6..40452d1 100644
--- a/app/src/main/res/layout/activity_main.xml
+++ b/app/src/main/res/layout/activity_main.xml
@@ -112,6 +112,12 @@
             android:text="查看周边可直播车"
             android:visibility="visible" />
 
+        <Button
+            android:id="@+id/btnV2XFunctionTest"
+            android:layout_width="match_parent"
+            android:layout_height="match_parent"
+            android:visibility="gone"/>
+
         <Button
             android:id="@+id/btnNSDNetty"
             android:layout_width="match_parent"
diff --git a/gradle.properties b/gradle.properties
index 44a411e..9bf96bd 100644
--- a/gradle.properties
+++ b/gradle.properties
@@ -33,7 +33,7 @@ SNAPSHOT_REPOSITORY_URL=http://nexus.zhidaoauto.com/repository/maven-snapshots/
 USERNAME=xintai
 PASSWORD=xintai2018
 # 编译模式: false - 依赖本地版本， true - 依赖 maven 版本
-RELEASE=true
+RELEASE=false
 # AI CLOUD 云平台
 # 工具类
 MOGO_UTILS_VERSION=1.3.21
@@ -55,7 +55,8 @@ MOGO_LIVE_VERSION=1.3.21
 MOGO_TRAFFICLIVE_VERSION=1.3.21
 # 定位服务
 MOGO_LOCATION_VERSION=1.3.21
-
+# 远程通讯模块
+MOGO_TELEMATIC_VERSION=1.3.19
 # v2x
 MOGO_V2X_VERSION=1.0.2
 # 远程通讯模块
diff --git a/libraries/mogo-telematic/src/main/java/com/mogo/telematic/server/netty/MogoProtocolServerHandler.java b/libraries/mogo-telematic/src/main/java/com/mogo/telematic/server/netty/MogoProtocolServerHandler.java
new file mode 100644
index 0000000..cb2a333
--- /dev/null
+++ b/libraries/mogo-telematic/src/main/java/com/mogo/telematic/server/netty/MogoProtocolServerHandler.java
@@ -0,0 +1,69 @@
+package com.mogo.telematic.server.netty;
+
+import android.util.Log;
+
+import com.mogo.telematic.MogoProtocolMsg;
+
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+
+@ChannelHandler.Sharable
+public class MogoProtocolServerHandler extends SimpleChannelInboundHandler<MogoProtocolMsg> {
+    private static final String TAG = "MogoServerHandler";
+    private NettyServerListener mListener;
+
+    public MogoProtocolServerHandler(NettyServerListener listener) {
+        this.mListener = listener;
+    }
+
+    @Override
+    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
+//    	System.out.println("channelReadComplete");
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx,
+                                Throwable cause) {
+        cause.printStackTrace();                //5
+        ctx.close();
+        //6
+    }
+
+    @Override
+    protected void channelRead0(ChannelHandlerContext ctx, MogoProtocolMsg msg) throws Exception {
+        Log.d(TAG, "server$channelRead0():" + msg.toString());
+        if (msg.getProtocolType() == MogoProtocolMsg.IDENTITY_REGIST) {
+            String sn = new String(msg.getBody());
+            NettyTcpServer.getInstance().putChannelSN(ctx.channel().id(), sn);
+        } else {
+            if (mListener != null) {
+                mListener.onMessageResponseServer(msg, ctx.channel());
+            }
+        }
+    }
+
+    /**
+     * 连接成功
+     *
+     * @param ctx
+     * @throws Exception
+     */
+    @Override
+    public void channelActive(ChannelHandlerContext ctx) throws Exception {
+        Log.e(TAG, "channelActive");
+        if (mListener != null) {
+            mListener.onChannelConnect(ctx.channel());
+        }
+//        NettyTcpServer.getInstance().setConnectStatus(true);
+    }
+
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        Log.e(TAG, "channelInactive");
+//        NettyTcpServer.getInstance().setConnectStatus(false);
+        if (mListener != null) {
+            mListener.onChannelDisConnect(ctx.channel());
+        }
+    }
+}
diff --git a/libraries/mogo-telematic/src/main/java/com/mogo/telematic/server/netty/NettyServerListener.java b/libraries/mogo-telematic/src/main/java/com/mogo/telematic/server/netty/NettyServerListener.java
new file mode 100644
index 0000000..8d9f4c7
--- /dev/null
+++ b/libraries/mogo-telematic/src/main/java/com/mogo/telematic/server/netty/NettyServerListener.java
@@ -0,0 +1,43 @@
+package com.mogo.telematic.server.netty;
+
+
+import io.netty.channel.Channel;
+
+
+public interface NettyServerListener<T> {
+
+    public final static byte STATUS_CONNECT_SUCCESS = 1;
+
+    public final static byte STATUS_CONNECT_CLOSED = 0;
+
+    public final static byte STATUS_CONNECT_ERROR = 0;
+
+    /**
+     * @param msg: 收到Client发送过来的消息
+     */
+    void onMessageResponseServer(T msg, Channel channel);
+
+    /**
+     * server开启成功
+     */
+    void onStartServer();
+
+    /**
+     * server关闭
+     */
+    void onStopServer();
+
+    /**
+     * 与客户端建立连接
+     *
+     * @param channel
+     */
+    void onChannelConnect(Channel channel);
+
+    /**
+     * 与客户端断开连接
+     * @param
+     */
+    void onChannelDisConnect(Channel channel);
+
+}
diff --git a/libraries/mogo-telematic/src/main/java/com/mogo/telematic/server/netty/NettyTcpServer.java b/libraries/mogo-telematic/src/main/java/com/mogo/telematic/server/netty/NettyTcpServer.java
new file mode 100644
index 0000000..4653635
--- /dev/null
+++ b/libraries/mogo-telematic/src/main/java/com/mogo/telematic/server/netty/NettyTcpServer.java
@@ -0,0 +1,207 @@
+package com.mogo.telematic.server.netty;
+
+import static com.mogo.telematic.MogoLengthFrameDecoder.LENGTH_FIELD_OFFSET;
+import static com.mogo.telematic.MogoLengthFrameDecoder.LENGTH_FIELD_SIZE;
+import static com.mogo.telematic.MogoLengthFrameDecoder.MAX_FRAME_LENGTH;
+
+import android.util.Log;
+
+import com.mogo.telematic.MogoLengthFrameDecoder;
+import com.mogo.telematic.MogoMessageEncoder;
+import com.mogo.telematic.MogoProtocolMsg;
+
+import java.net.InetSocketAddress;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelId;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+
+
+/**
+ * TCP 服务端
+ * 目前服务端支持连接多个客户端
+ */
+public class NettyTcpServer {
+
+    private static final String TAG = "NettyTcpServer";
+    // client端用来过滤的
+    public static final String SERVER_NAME = "NSD_SERVER";
+    public static final int SERVER_PORT = 1088;
+
+    private Channel selectChannel;
+    private CopyOnWriteArrayList<Channel> channelList = new CopyOnWriteArrayList<>();
+    /**
+     * channelId和业务标识(sn)绑定(场景一是断开连接时知道是哪个设备断开)
+     */
+    private ConcurrentMap<ChannelId, String> chlIdSnMap = new ConcurrentHashMap<>();
+
+    private static NettyTcpServer instance = null;
+    private NettyServerListener mListener;
+    //    private boolean connectStatus;
+    private EventLoopGroup bossGroup;
+    private EventLoopGroup workerGroup;
+    private boolean isServerStart;
+
+    public static NettyTcpServer getInstance() {
+        if (instance == null) {
+            synchronized (NettyTcpServer.class) {
+                if (instance == null) {
+                    instance = new NettyTcpServer();
+                }
+            }
+        }
+        return instance;
+    }
+
+    private NettyTcpServer() {
+    }
+
+    public void start() {
+        new Thread() {
+            @Override
+            public void run() {
+                super.run();
+                startNettyServer();
+            }
+        }.start();
+    }
+
+    public void startNettyServer() {
+        bossGroup = new NioEventLoopGroup(1);
+        workerGroup = new NioEventLoopGroup();
+        try {
+            ServerBootstrap b = new ServerBootstrap();
+            b.group(bossGroup, workerGroup)
+                    .channel(NioServerSocketChannel.class)
+                    .localAddress(new InetSocketAddress(SERVER_PORT))
+                    .childOption(ChannelOption.SO_KEEPALIVE, true)
+                    .childOption(ChannelOption.SO_REUSEADDR, true)
+                    .childHandler(new ChannelInitializer<SocketChannel>() {
+
+                        @Override
+                        public void initChannel(SocketChannel ch) throws Exception {
+                            // decoder
+                            ch.pipeline().addLast(new MogoLengthFrameDecoder(MAX_FRAME_LENGTH,
+                                    LENGTH_FIELD_OFFSET,
+                                    LENGTH_FIELD_SIZE));
+                            // encoder
+                            ch.pipeline().addLast(new MogoMessageEncoder());
+                            ch.pipeline().addLast(new MogoProtocolServerHandler(mListener));
+                        }
+                    });
+
+            // Bind and start to accept incoming connections.
+            ChannelFuture f = b.bind().sync();
+            Log.e(TAG, NettyTcpServer.class.getName() + " started and listen on " + f.channel().localAddress());
+            isServerStart = true;
+            if (mListener != null) {
+                mListener.onStartServer();
+            }
+            // Wait until the server socket is closed.
+            // In this example, this does not happen, but you can do that to gracefully
+            // shut down your server.
+            f.channel().closeFuture().sync();
+        } catch (Exception e) {
+            Log.e(TAG, e.getLocalizedMessage());
+            e.printStackTrace();
+        } finally {
+            isServerStart = false;
+            if (mListener != null) {
+                mListener.onStopServer();
+            }
+            workerGroup.shutdownGracefully();
+            bossGroup.shutdownGracefully();
+        }
+    }
+
+    public void disconnect() {
+        workerGroup.shutdownGracefully();
+        bossGroup.shutdownGracefully();
+    }
+
+    public void setListener(NettyServerListener listener) {
+        this.mListener = listener;
+    }
+
+    public boolean isServerStart() {
+        return isServerStart;
+    }
+
+    public boolean sendMsgToClient(MogoProtocolMsg mogoProtocolMsg, ChannelFutureListener listener) {
+        boolean flag = selectChannel != null && selectChannel.isActive();
+        if (flag) {
+            selectChannel.writeAndFlush(mogoProtocolMsg).addListener(listener);
+        }
+        return flag;
+    }
+
+    public void sendMsgToAllClients(MogoProtocolMsg mogoProtocolMsg) {
+        for (Channel chl : channelList) {
+            if (chl != null && chl.isActive()) {
+                chl.writeAndFlush(mogoProtocolMsg).addListener(mProxyListener);
+            }
+        }
+    }
+
+    private ChannelFutureListener mProxyListener = channelFuture -> {
+        if (channelFuture.isSuccess()) {
+            Log.d(TAG, "SendMsgToAllClients result was successful");
+        } else {
+            Log.d(TAG, "SendMsgToAllClients result was unsuccessful");
+        }
+    };
+
+    /**
+     * 切换通道(单通道通信)
+     * 设置服务端，与哪个客户端通信
+     *
+     * @param channel
+     */
+    public void selectorChannel(Channel channel) {
+        this.selectChannel = channel;
+    }
+
+    /**
+     *
+     * @param channel: Socket连接成功时保存下
+     */
+    public void addChannel(Channel channel) {
+        channelList.add(channel);
+    }
+
+    public void removeChannel(Channel channel) {
+        channelList.remove(channel);
+    }
+
+    public void putChannelSN(ChannelId channelId, String sn) {
+        chlIdSnMap.put(channelId, sn);
+    }
+
+    public void removeChannelSN(ChannelId channelId) {
+        chlIdSnMap.remove(channelId);
+    }
+
+    /**
+     *
+     * @param channelId 每次socket连接成功时分配的channelId，断开重连后会改变
+     * @return
+     */
+    public String getSNByChlId(String channelId) {
+        String sn = "";
+        if (chlIdSnMap != null) {
+            sn = chlIdSnMap.get(channelId);
+        }
+        return sn;
+    }
+}