yangyakun921/MoGoAiCloudSdk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[Update]自定义协议并解决TCP粘包/拆包问题

Browse Source
This commit is contained in:
chenfufeng
2022-02-14 15:58:48 +08:00
parent 4e2c6ffd7a
commit fc89d1a34b
10 changed files with 227 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
app/src/main/java/com/mogo/cloud/netty/client/NettyClientActivity.java
View File

@@ -13,13 +13,14 @@ import androidx.recyclerview.widget.RecyclerView;
import com.mogo.cloud.R;
import com.mogo.cloud.netty.LogBean;
import com.mogo.telematic.MogoProtocolMsg;
import com.mogo.telematic.NSDNettyManager;
import com.mogo.telematic.client.listener.NettyClientListener;
import com.mogo.telematic.client.status.ConnectState;
import java.util.Arrays;
public class NettyClientActivity extends AppCompatActivity implements View.OnClickListener, NettyClientListener<byte[]> {
public class NettyClientActivity extends AppCompatActivity implements View.OnClickListener, NettyClientListener<MogoProtocolMsg> {
private static final String TAG = "NettyClientActivity";
@@ -36,6 +37,8 @@ public class NettyClientActivity extends AppCompatActivity implements View.OnCli
private final byte[] sendByte = new byte[]{0x55, 0x54, 0x72, 0x21};
private final MogoProtocolMsg mogoProtocolMsg = new MogoProtocolMsg(MogoProtocolMsg.NORMAL_DATA, 3, new byte[]{0x11, 0x12, 0x13});
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
@@ -68,7 +71,7 @@ public class NettyClientActivity extends AppCompatActivity implements View.OnCli
mSendBtn.setOnClickListener(this);
mClearLog.setOnClickListener(this);
mSendET.setText(Arrays.toString(sendByte));
mSendET.setText(Arrays.toString(mogoProtocolMsg.getBody()));
}
@Override
@@ -82,15 +85,15 @@ public class NettyClientActivity extends AppCompatActivity implements View.OnCli
if (!NSDNettyManager.getInstance().getConnectStatus()) {
Toast.makeText(getApplicationContext(), "未连接,请先连接", Toast.LENGTH_SHORT).show();
} else {
NSDNettyManager.getInstance().sendByteArrayToServer(sendByte, isSuccess -> {
NSDNettyManager.getInstance().sendMogoProtocolMsgToServer(mogoProtocolMsg, isSuccess -> {
if (isSuccess) {
Log.d(TAG, "Write auth successful");
logSend(Arrays.toString(sendByte));
logSend(Arrays.toString(mogoProtocolMsg.getBody()));
} else {
Log.d(TAG, "Write auth error");
}
});
mSendET.setText(Arrays.toString(sendByte));
mSendET.setText(Arrays.toString(mogoProtocolMsg.getBody()));
}
break;
@@ -104,8 +107,8 @@ public class NettyClientActivity extends AppCompatActivity implements View.OnCli
}
@Override
public void onMessageResponseClient(byte[] msg, int index) {
String result = Arrays.toString(msg);
public void onMessageResponseClient(MogoProtocolMsg msg, int index) {
String result = msg.toString();
Log.e(TAG, "onMessageResponse:" + result);
logRece(result);
}
@@ -125,15 +128,19 @@ public class NettyClientActivity extends AppCompatActivity implements View.OnCli
}
private void logSend(String log) {
LogBean logBean = new LogBean(System.currentTimeMillis(), log);
mSendLogAdapter.getDataList().add(0, logBean);
runOnUiThread(() -> mSendLogAdapter.notifyDataSetChanged());
runOnUiThread(() -> {
LogBean logBean = new LogBean(System.currentTimeMillis(), log);
mSendLogAdapter.getDataList().add(0, logBean);
mSendLogAdapter.notifyDataSetChanged();
});
}
private void logRece(String log) {
LogBean logBean = new LogBean(System.currentTimeMillis(), log);
mReceLogAdapter.getDataList().add(0, logBean);
runOnUiThread(() -> mReceLogAdapter.notifyDataSetChanged());
runOnUiThread(() -> {
LogBean logBean = new LogBean(System.currentTimeMillis(), log);
mReceLogAdapter.getDataList().add(0, logBean);
mReceLogAdapter.notifyDataSetChanged();
});
}
public void disconnect(View view) {

37
app/src/main/java/com/mogo/cloud/netty/server/NettyServerActivity.java
View File

@@ -18,6 +18,7 @@ import androidx.recyclerview.widget.RecyclerView;
import com.mogo.cloud.R;
import com.mogo.cloud.netty.LogBean;
import com.mogo.telematic.MogoProtocolMsg;
import com.mogo.telematic.NSDNettyManager;
import com.mogo.telematic.NetworkUtils;
import com.mogo.telematic.server.bean.ClientChanel;
@@ -30,7 +31,7 @@ import java.util.List;
import io.netty.channel.Channel;
public class NettyServerActivity extends AppCompatActivity implements View.OnClickListener, NettyServerListener<byte[]> {
public class NettyServerActivity extends AppCompatActivity implements View.OnClickListener, NettyServerListener<MogoProtocolMsg> {
private static final String TAG = "NettyServerActivity";
@@ -52,6 +53,8 @@ public class NettyServerActivity extends AppCompatActivity implements View.OnCli
private final byte[] sendByte = new byte[]{0x55, 0x54, 0x72, 0x21};
private final MogoProtocolMsg mogoProtocolMsg = new MogoProtocolMsg(MogoProtocolMsg.NORMAL_DATA, 3, new byte[]{0x11, 0x12, 0x13});
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
@@ -107,7 +110,7 @@ public class NettyServerActivity extends AppCompatActivity implements View.OnCli
startServer.setOnClickListener(this);
mSendBtn.setOnClickListener(this);
mClearLog.setOnClickListener(this);
mSendET.setText(Arrays.toString(sendByte));
mSendET.setText(Arrays.toString(mogoProtocolMsg.getBody()));
}
@Override
@@ -127,15 +130,15 @@ public class NettyServerActivity extends AppCompatActivity implements View.OnCli
if (!NSDNettyManager.getInstance().isServerStart()) {
Toast.makeText(getApplicationContext(), "未连接,请先连接", LENGTH_SHORT).show();
} else {
NSDNettyManager.getInstance().sendByteArrayToClient(sendByte, channelFuture -> {
NSDNettyManager.getInstance().sendMogoProtocolMsgToClient(mogoProtocolMsg, channelFuture -> {
if (channelFuture.isSuccess()) {
Log.d(TAG, "Write auth successful");
logSend(Arrays.toString(sendByte));
logSend(Arrays.toString(mogoProtocolMsg.getBody()));
} else {
Log.d(TAG, "Write auth error");
}
});
mSendET.setText(Arrays.toString(sendByte));
mSendET.setText(Arrays.toString(mogoProtocolMsg.getBody()));
}
break;
@@ -149,10 +152,10 @@ public class NettyServerActivity extends AppCompatActivity implements View.OnCli
}
@Override
public void onMessageResponseServer(byte[] msg, String uniqueId) {
Log.e(TAG,"onMessageResponseServer:ChannelId:"+uniqueId);
public void onMessageResponseServer(MogoProtocolMsg msg, String channelId) {
Log.e(TAG,"onMessageResponseServer:channelId:" + channelId);
String result = Arrays.toString(msg);
String result = msg.toString();
Log.e(TAG, "onMessageResponse:" + result);
logRece(result);
}
@@ -175,7 +178,7 @@ public class NettyServerActivity extends AppCompatActivity implements View.OnCli
@Override
public void onChannelDisConnect(Channel channel) {
Log.e(TAG, "onChannelDisConnect:ChannelId" + channel.id().asShortText());
Log.d(TAG, "onChannelDisConnect:ChannelId is:" + channel.id().asShortText());
for (int i = 0; i < clientChanelArray.size(); i++) {
final ClientChanel clientChanel = clientChanelArray.get(i);
@@ -222,14 +225,18 @@ public class NettyServerActivity extends AppCompatActivity implements View.OnCli
}
private void logSend(String log) {
LogBean logBean = new LogBean(System.currentTimeMillis(), log);
mSendLogAdapter.getDataList().add(0, logBean);
runOnUiThread(() -> mSendLogAdapter.notifyDataSetChanged());
runOnUiThread(() -> {
LogBean logBean = new LogBean(System.currentTimeMillis(), log);
mSendLogAdapter.getDataList().add(0, logBean);
mSendLogAdapter.notifyDataSetChanged();
});
}
private void logRece(String log) {
LogBean logBean = new LogBean(System.currentTimeMillis(), log);
mReceLogAdapter.getDataList().add(0, logBean);
runOnUiThread(() -> mReceLogAdapter.notifyDataSetChanged());
runOnUiThread(() -> {
LogBean logBean = new LogBean(System.currentTimeMillis(), log);
mReceLogAdapter.getDataList().add(0, logBean);
mReceLogAdapter.notifyDataSetChanged();
});
}
}

2
gradle.properties
View File

@@ -59,4 +59,4 @@ MOGO_LOCATION_VERSION=1.3.21
# v2x
MOGO_V2X_VERSION=1.0.2
# 远程通讯模块
MOGO_TELEMATIC_VERSION=1.3.17
MOGO_TELEMATIC_VERSION=1.3.19

34
libraries/mogo-telematic/src/main/java/com/mogo/telematic/MogoLengthFrameDecoder.java Normal file
View File

@@ -0,0 +1,34 @@
package com.mogo.telematic;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
public class MogoLengthFrameDecoder extends LengthFieldBasedFrameDecoder {
public static final int MAX_FRAME_LENGTH = 3 * 1024 * 1024;
// 参考MogoProtocolMsg结构(https://blog.csdn.net/thinking_fioa/article/details/80573483)
public static final int LENGTH_FIELD_OFFSET = 4;
public static final int LENGTH_FIELD_SIZE = 4;
public MogoLengthFrameDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength) {
super(maxFrameLength, lengthFieldOffset, lengthFieldLength);
}
@Override
protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
ByteBuf byteBuf = (ByteBuf) super.decode(ctx, in);
if (byteBuf == null) return null;
MogoProtocolMsg mogoProtocolMsg = new MogoProtocolMsg();
mogoProtocolMsg.setProtocolType(byteBuf.readInt());
int length = byteBuf.readInt();
mogoProtocolMsg.setBodyLength(length);
if (length > 0) {
byte[] body = new byte[length];
byteBuf.readBytes(body);
mogoProtocolMsg.setBody(body);
}
in.release();
return mogoProtocolMsg;
}
}

20
libraries/mogo-telematic/src/main/java/com/mogo/telematic/MogoMessageEncoder.java Normal file
View File

@@ -0,0 +1,20 @@
package com.mogo.telematic;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
public class MogoMessageEncoder extends MessageToByteEncoder<MogoProtocolMsg> {
@Override
protected void encode(ChannelHandlerContext ctx, MogoProtocolMsg msg, ByteBuf out) throws Exception {
int protocolType = msg.getProtocolType();
int length = msg.getBodyLength();
byte[] body = msg.getBody();
out.writeInt(protocolType);
out.writeInt(length);
if (length > 0) {
out.writeBytes(body);
}
}
}

54
libraries/mogo-telematic/src/main/java/com/mogo/telematic/MogoProtocolMsg.java Normal file
View File

@@ -0,0 +1,54 @@
package com.mogo.telematic;
import java.util.Arrays;
public class MogoProtocolMsg {
// 心跳数据包类型
public static final int HEART_DATA = 0;
public static final int NORMAL_DATA = 1;
private int protocolType;
private int bodyLength;
private byte[] body;
public MogoProtocolMsg() {}
public MogoProtocolMsg(int protocolType, int bodyLength, byte[] body) {
this.protocolType = protocolType;
this.bodyLength = bodyLength;
this.body = body;
}
public int getProtocolType() {
return protocolType;
}
public void setProtocolType(int protocolType) {
this.protocolType = protocolType;
}
public int getBodyLength() {
return bodyLength;
}
public void setBodyLength(int bodyLength) {
this.bodyLength = bodyLength;
}
public byte[] getBody() {
return body;
}
public void setBody(byte[] body) {
this.body = body;
}
@Override
public String toString() {
return "MogoProtocolMsg{" +
"protocolType=" + protocolType +
", bodyLength=" + bodyLength +
", body=" + Arrays.toString(body) +
'}';
}
}

20
libraries/mogo-telematic/src/main/java/com/mogo/telematic/NSDNettyManager.java
View File

@@ -49,7 +49,9 @@ public class NSDNettyManager {
private NettyServerListener mDefaultListener = new NettyServerListener() {
@Override
public void onMessageResponseServer(Object msg, String ChannelId) {
if (mListener != null) {
mListener.onMessageResponseServer(msg, ChannelId);
}
}
@Override
@@ -114,6 +116,10 @@ public class NSDNettyManager {
NettyTcpServer.getInstance().sendMsgToClient(byteArray, listener);
}
public void sendMogoProtocolMsgToClient(MogoProtocolMsg mogoProtocolMsg, ChannelFutureListener listener) {
NettyTcpServer.getInstance().sendMsgToClient(mogoProtocolMsg, listener);
}
/**
* 服务器端注册一个可供NSD探测到的网络 Ip 地址，便于给展示叫号机连接此socket
*/
@@ -204,11 +210,9 @@ public class NSDNettyManager {
.setMaxReconnectTimes(5) //设置最大重连次数
.setReconnectIntervalTime(5) //设置重连间隔时间。单位：秒
.setSendHeartBeat(true) //设置是否发送心跳
.setHeartBeatInterval(5) //设置心跳间隔时间。单位：秒
.setHeartBeatData(new byte[]{0x00, 0x00, 0x00, 0x00}) //设置心跳数据，可以是String类型，也可以是byte[]，以后设置的为准
.setHeartBeatInterval(120) //设置心跳间隔时间。单位：秒
.setHeartBeatData(new MogoProtocolMsg(MogoProtocolMsg.HEART_DATA, 2, new byte[]{0x00, 0x00})) //设置心跳数据，可以是String类型，也可以是byte[]，以后设置的为准
.setIndex(0) //设置客户端标识.(因为可能存在多个tcp连接)
// .setPacketSeparator("#")//用特殊字符，作为分隔符，解决粘包问题，默认是用换行符作为分隔符
// .setMaxPacketLong(1024)//设置一次发送数据的最大长度，默认是1024，最大值为Integer.MAX
.build();
if (listener != null) {
mNettyTcpClient.setListener(listener); //设置TCP监听
@@ -226,6 +230,12 @@ public class NSDNettyManager {
return mNettyTcpClient.getConnectStatus();
}
public void sendMogoProtocolMsgToServer(MogoProtocolMsg mogoProtocolMsg, final MessageStateListener listener) {
if (mNettyTcpClient != null) {
mNettyTcpClient.sendMsgToServer(mogoProtocolMsg, listener);
}
}
/**
* 发送byte[]到服务端
* @param byteArray

38
libraries/mogo-telematic/src/main/java/com/mogo/telematic/client/NettyTcpClient.java
View File

@@ -1,9 +1,16 @@
package com.mogo.telematic.client;
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.os.SystemClock;
import android.text.TextUtils;
import android.util.Log;
import com.mogo.telematic.MogoLengthFrameDecoder;
import com.mogo.telematic.MogoMessageEncoder;
import com.mogo.telematic.MogoProtocolMsg;
import com.mogo.telematic.client.handler.NettyClientHandler;
import com.mogo.telematic.client.listener.MessageStateListener;
import com.mogo.telematic.client.listener.NettyClientListener;
@@ -153,6 +160,12 @@ public class NettyTcpClient {
if (isSendheartBeat) {
ch.pipeline().addLast("ping", new IdleStateHandler(0, heartBeatInterval, 0, TimeUnit.SECONDS));//5s未发送数据，回调userEventTriggered
}
// 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 NettyClientHandler(listener, mIndex, isSendheartBeat, heartBeatData, packetSeparator));
}
});
@@ -227,20 +240,11 @@ public class NettyTcpClient {
ChannelFuture channelFuture = channel.writeAndFlush(data + separator).addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture channelFuture) throws Exception {
listener.isSendSuccss(channelFuture.isSuccess());
listener.isSendSuccess(channelFuture.isSuccess());
}
});
}
return flag;
// ByteBuf buffer = Unpooled.copiedBuffer(data, Charset.forName("UTF-8"));
// if (flag) {
// channel.writeAndFlush(buffer).addListener(listener);
// }
// return flag;
// byte[] bytes = strToByteArray(data);
//
// return sendMsgToServer(bytes,listener);
}
/**
@@ -259,13 +263,25 @@ public class NettyTcpClient {
return false;
}
public boolean sendMsgToServer(MogoProtocolMsg protocolMsg, final MessageStateListener listener) {
boolean flag = channel != null && isConnect;
if (flag) {
channel.writeAndFlush(protocolMsg).addListener((ChannelFutureListener) channelFuture -> {
if (listener != null) {
listener.isSendSuccess(channelFuture.isSuccess());
}
});
}
return flag;
}
public boolean sendMsgToServer(byte[] data, final MessageStateListener listener) {
boolean flag = channel != null && isConnect;
if (flag) {
ByteBuf buf = Unpooled.copiedBuffer(data);
channel.writeAndFlush(buf).addListener((ChannelFutureListener) channelFuture -> {
if (listener != null) {
listener.isSendSuccss(channelFuture.isSuccess());
listener.isSendSuccess(channelFuture.isSuccess());
}
});
}

71
libraries/mogo-telematic/src/main/java/com/mogo/telematic/client/handler/NettyClientHandler.java
View File

@@ -3,6 +3,7 @@ package com.mogo.telematic.client.handler;
import android.text.TextUtils;
import android.util.Log;
import com.mogo.telematic.MogoProtocolMsg;
import com.mogo.telematic.client.listener.NettyClientListener;
import com.mogo.telematic.client.status.ConnectState;
@@ -16,7 +17,7 @@ import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
public class NettyClientHandler extends SimpleChannelInboundHandler<Object> {
public class NettyClientHandler extends SimpleChannelInboundHandler<MogoProtocolMsg> {
private static final String TAG = "NettyClientHandler";
private final boolean isSendheartBeat;
@@ -25,11 +26,6 @@ public class NettyClientHandler extends SimpleChannelInboundHandler<Object> {
private Object heartBeatData;
private String packetSeparator;
// private static final ByteBuf HEARTBEAT_SEQUENCE = Unpooled.unreleasableBuffer(Unpooled.copiedBuffer("Heartbeat"+System.getProperty("line.separator"),
// CharsetUtil.UTF_8));
byte[] requestBody = {(byte) 0xFE, (byte) 0xED, (byte) 0xFE, 5, 4, (byte) 0xFF, 0x0a};
public NettyClientHandler(NettyClientListener listener, int index, boolean isSendheartBeat, Object heartBeatData) {
this(listener, index, isSendheartBeat, heartBeatData, null);
}
@@ -43,32 +39,40 @@ public class NettyClientHandler extends SimpleChannelInboundHandler<Object> {
}
/**
* <p>设定IdleStateHandler心跳检测每x秒进行一次读检测，
* 如果x秒内ChannelRead()方法未被调用则触发一次userEventTrigger()方法 </p>
* 设定IdleStateHandler心跳检测每x秒进行一次读检测，
*
* @param ctx ChannelHandlerContext
* @param evt IdleStateEvent
*/
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof IdleStateEvent) {
IdleStateEvent event = (IdleStateEvent) evt;
if (event.state() == IdleState.WRITER_IDLE) { //发送心跳
// ctx.channel().writeAndFlush("Heartbeat" + System.getProperty("line.separator"));
if (isSendheartBeat) {
if (heartBeatData == null) {
ByteBuf buf = Unpooled.copiedBuffer(requestBody);
ctx.channel().writeAndFlush(buf);
} else if (heartBeatData instanceof byte[]) {
ByteBuf buf = Unpooled.copiedBuffer((byte[]) heartBeatData);
ctx.channel().writeAndFlush(buf);
} else {
Log.e(TAG, "userEventTriggered: heartBeatData type error");
}
} else {
Log.e(TAG, "不发送心跳");
}
switch (event.state()) {
case WRITER_IDLE:
case READER_IDLE:
sendHeartbeat(ctx);
break;
default:
break;
}
} else {
super.userEventTriggered(ctx, evt);
}
}
private void sendHeartbeat(ChannelHandlerContext ctx) {
if (isSendheartBeat) {
if (heartBeatData == null) {
MogoProtocolMsg heartData = new MogoProtocolMsg(MogoProtocolMsg.NORMAL_DATA, 2, new byte[]{0x00, 0x00});
ctx.channel().writeAndFlush(heartData);
} else if (heartBeatData instanceof MogoProtocolMsg) {
ctx.channel().writeAndFlush(heartBeatData);
} else {
Log.e(TAG, "userEventTriggered: heartBeatData type error");
}
} else {
Log.d(TAG, "不发送心跳");
}
}
@@ -100,22 +104,13 @@ public class NettyClientHandler extends SimpleChannelInboundHandler<Object> {
/**
* 客户端收到消息
*
* @param channelHandlerContext ChannelHandlerContext
* @param msg 消息
* @param ctx ChannelHandlerContext
* @param msg 消息
*/
@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object msg) {
Log.e(TAG, "channelRead0:" + msg);
try {
ByteBuf buf = (ByteBuf) msg;
byte[] bytes = new byte[buf.readableBytes()];
buf.readBytes(bytes);
String result = new String(bytes, "utf-8");
Log.e(TAG, "Server: " + result);
listener.onMessageResponseClient(bytes, index);
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
protected void channelRead0(ChannelHandlerContext ctx, MogoProtocolMsg msg) throws Exception {
Log.d(TAG, "Client channelRead0:" + msg.toString());
listener.onMessageResponseClient(msg, index);
}
/**

2
libraries/mogo-telematic/src/main/java/com/mogo/telematic/client/listener/MessageStateListener.java
View File

@@ -5,5 +5,5 @@ package com.mogo.telematic.client.listener;
* 发送状态监听
*/
public interface MessageStateListener {
void isSendSuccss(boolean isSuccess);
void isSendSuccess(boolean isSuccess);
}