zjyr
/
PlatformServer
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

1. 增加控制器协议发送服务

This commit is contained in:
HuangXin 2023-08-13 11:48:05 +08:00
parent da50d2545b
commit c39634aa4a
11 changed files with 205 additions and 153 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/main/java/com/zjyr/beidouservice/adapter/impl/netty/ISocketServer.java
View File

@ -1,5 +1,6 @@
package com.zjyr.beidouservice.adapter.impl.netty;
import io.netty.channel.Channel;
import jakarta.annotation.PreDestroy;
/**
@ -19,9 +20,8 @@ public interface ISocketServer {
* @param <T> the type parameter
* @param data the data
* @param devId the dev id
* @return the int
*/
<T> int sendData(T data, Long devId);
void sendData(Channel ch, Object data);
/**
* Destory.

86
src/main/java/com/zjyr/beidouservice/adapter/impl/netty/MessageHandler.java
View File

@ -2,10 +2,7 @@ package com.zjyr.beidouservice.adapter.impl.netty;
import com.zjyr.beidouservice.adapter.impl.netty.notify.SocketNotifyEvent;
import com.zjyr.beidouservice.common.impl.SocketEventName;
import com.zjyr.beidouservice.misc.HelperUtils;
import com.zjyr.beidouservice.pojo.vo.ControlAdapterSocketCtx;
import com.zjyr.beidouservice.pojo.vo.binary.BaseBinaryProtocol;
import com.zjyr.beidouservice.pojo.vo.binary.ControllerStatus;
import com.zjyr.beidouservice.pojo.vo.binary.HeartProtocol;
import com.zjyr.beidouservice.pojo.vo.binary.MessageContent;
import io.netty.channel.Channel;
@ -21,7 +18,6 @@ import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Component;
import java.net.InetSocketAddress;
import java.util.concurrent.ConcurrentHashMap;
/**
* The type Message handler.
@ -33,33 +29,17 @@ import java.util.concurrent.ConcurrentHashMap;
@ChannelHandler.Sharable
@RequiredArgsConstructor
public class MessageHandler<T> extends SimpleChannelInboundHandler<BaseBinaryProtocol<T>> {
/**
* The constant ctxMap.
*/
public static ConcurrentHashMap<Long, ControlAdapterSocketCtx> ctxMap = new ConcurrentHashMap<>();
/**
* The Application event publisher.
*/
@Resource
ApplicationEventPublisher applicationEventPublisher;
//@Resource
//private ControlDeviceServiceImpl controlDeviceService;
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof IdleStateEvent idleStateEvent) {
if (idleStateEvent.state() == IdleState.ALL_IDLE) {
log.info("{}:: Trigger Heart Signe", ctx.channel().id());
// MessageContent<HeartProtocol> msgCont = MessageContent.<HeartProtocol>builder()
// .msgBody(new HeartProtocol())
// .build();
// BaseBinaryProtocol<HeartProtocol> h = BaseBinaryProtocol.<HeartProtocol>builder()
// .msgContent(msgCont)
// .build();
//
// ctx.writeAndFlush(h);
SocketNotifyEvent notify = new SocketNotifyEvent(this,
ctx.channel(),
SocketEventName.SOCKET_EVT_IDLE_TIMEOUT,
@ -73,40 +53,17 @@ public class MessageHandler<T> extends SimpleChannelInboundHandler<BaseBinaryPro
@Override
public void channelRead0(ChannelHandlerContext ctx, BaseBinaryProtocol<T> message) throws Exception {
if (message.getMsgContent().getMsgBody() instanceof ControllerStatus status) {
log.info("{}:: Receive Message: {}",
ctx.channel().id(),
HelperUtils.bytesToHexString(status.getBeidouSignalStrength()));
} else {
log.info("{}:: Receive Message: {}", ctx.channel().id(), "");
}
SocketNotifyEvent notify = new SocketNotifyEvent(this,
ctx.channel(),
SocketEventName.SOCKET_EVT_RECV,
message.getMsgContent().getMsgBody());
applicationEventPublisher.publishEvent(notify);
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
InetSocketAddress sa = (InetSocketAddress) ctx.channel().remoteAddress();
//List<ControlDevice> list = controlDeviceMapper.selectAll();
log.info("{}:: Connected <-- {}:{}", ctx.channel().id(), sa.getAddress().getHostAddress(), sa.getPort());
// ControlDevice dev = ControlDevice.builder()
// .deviceType(BeidouAdapterTypeName.ADAPTER_SOCKET_TCP)
// .deviceAddr(sa.getAddress().getHostAddress())
// .build();
//controlDeviceService.addControlDevice(dev);
// if (ctxMap.get(dev.getId()) != null) {
// ctxMap.remove(dev.getId());
// } else {
// ctxMap.put(dev.getId(),
// ControlAdapterSocketCtx.builder().controlAdapterId(dev.getId()).channel(ctx.channel()).build
// ());
// }
// MessageContent<HeartProtocol> msgCont = MessageContent.<HeartProtocol>builder()
// .msgBody(new HeartProtocol())
// .build();
// ctx.writeAndFlush(BaseBinaryProtocol.<HeartProtocol>builder().msgContent(msgCont).build());
super.channelActive(ctx);
SocketNotifyEvent notify = new SocketNotifyEvent(this, ctx.channel(), SocketEventName.SOCKET_EVT_CONNECT, null);
applicationEventPublisher.publishEvent(notify);
@ -117,28 +74,14 @@ public class MessageHandler<T> extends SimpleChannelInboundHandler<BaseBinaryPro
InetSocketAddress sa = (InetSocketAddress) ctx.channel().remoteAddress();
log.info("{}:: Disonnected <-- {}", ctx.channel().id(), sa.getAddress().getHostAddress());
for (ConcurrentHashMap.Entry<Long, ControlAdapterSocketCtx> entry : ctxMap.entrySet()) {
if (entry.getValue().getChannel().id() == ctx.channel().id()) {
ctxMap.remove(entry.getKey());
}
}
super.channelActive(ctx);
super.channelInactive(ctx);
ctx.close();
}
/**
* Channel send data.
*
* @param <E> the type parameter
* @param proMsg the pro msg
* @param devId the dev id
*/
public <E> void channelSendData(E proMsg, Long devId) {
ControlAdapterSocketCtx ctx = ctxMap.get(devId);
if (ctx != null && ctx.getChannel() != null && ctx.getChannel().isActive()) {
ctx.getChannel().writeAndFlush(proMsg);
}
SocketNotifyEvent notify = new SocketNotifyEvent(this,
ctx.channel(),
SocketEventName.SOCKET_EVT_DISCONNECT,
null);
applicationEventPublisher.publishEvent(notify);
}
/**
@ -147,12 +90,9 @@ public class MessageHandler<T> extends SimpleChannelInboundHandler<BaseBinaryPro
* @param ctxChannel the ctx channel
* @param proMsg the pro msg
*/
public void channelSendData(Channel ctxChannel, T proMsg) {
public void channelSendData(Channel ctxChannel, Object proMsg) {
if (ctxChannel != null && ctxChannel.isActive()) {
MessageContent<HeartProtocol> msgCont = MessageContent.<HeartProtocol>builder()
.msgBody(new HeartProtocol())
.build();
ctxChannel.writeAndFlush(BaseBinaryProtocol.<HeartProtocol>builder().msgContent(msgCont).build());
ctxChannel.writeAndFlush(proMsg);
}
}
}

6
src/main/java/com/zjyr/beidouservice/adapter/impl/netty/impl/TcpServer.java
View File

@ -4,6 +4,7 @@ import com.zjyr.beidouservice.adapter.impl.netty.ChannelInit;
import com.zjyr.beidouservice.adapter.impl.netty.ISocketServer;
import com.zjyr.beidouservice.config.NettySocketConfigure;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
@ -83,8 +84,7 @@ public class TcpServer implements ISocketServer {
}
@Override
public <T> int sendData(T data, Long devId) {
channelInit.getMessageHandler().channelSendData(data, devId);
return 0;
public void sendData(Channel ch, Object data) {
channelInit.getMessageHandler().channelSendData(ch, data);
}
}

2
src/main/java/com/zjyr/beidouservice/adapter/impl/netty/notify/SocketNotifyEvent.java
View File

@ -35,7 +35,7 @@ public class SocketNotifyEvent extends ApplicationEvent {
* @param evtType the evt type
* @param evtMessage the evt message
*/
public SocketNotifyEvent(Object source, Channel ctx, SocketEventName evtType, List<Object> evtMessage) {
public SocketNotifyEvent(Object source, Channel ctx, SocketEventName evtType, Object evtMessage) {
super(source);
this.socketEvent = evtType;
this.evtMessage = evtMessage;

13
src/main/java/com/zjyr/beidouservice/misc/HelperUtils.java
View File

@ -1,5 +1,7 @@
package com.zjyr.beidouservice.misc;
import com.zjyr.beidouservice.pojo.vo.binary.BaseBinaryProtocol;
/**
* The type Helper utils.
*/
@ -18,8 +20,17 @@ public class HelperUtils {
if (sTemp.length() < 2) {
sb.append(0);
}
sb.append(sTemp.toUpperCase() + " ");
sb.append(sTemp.toUpperCase()).append(" ");
}
return sb.toString();
}
//
// @SuppressWarnings("unchecked")
// public static <T> BaseBinaryProtocol<T> CastBaseBinaryProtocol(Object obj) {
// return (BaseBinaryProtocol<T>)obj;
// }
//
// private HelperUtils() {
// throw new UnsupportedOperationException();
// }
}

22
src/main/java/com/zjyr/beidouservice/pojo/po/BeidouAdapterControlContent.java Normal file
View File

@ -0,0 +1,22 @@
package com.zjyr.beidouservice.pojo.po;
import com.zjyr.beidouservice.common.impl.SensorControlActionName;
import lombok.Builder;
import lombok.Data;
@Data
@Builder
public class BeidouAdapterControlContent {
/**
* The Districts code.
*/
private Integer districtsCode;
/**
* The Control action.
*/
private SensorControlActionName controlAction;
/**
* The Sensor id.
*/
private Integer sensorId;
}

7
src/main/java/com/zjyr/beidouservice/pojo/vo/ControlAdapterSocketCtx.java
View File

@ -3,12 +3,14 @@ package com.zjyr.beidouservice.pojo.vo;
import io.netty.channel.Channel;
import lombok.Builder;
import lombok.Data;
import lombok.RequiredArgsConstructor;
/**
* The type Control adapter socket ctx.
*/
@Data
@Builder
@RequiredArgsConstructor
public class ControlAdapterSocketCtx {
/**
* The Control adapter id.
@ -18,4 +20,9 @@ public class ControlAdapterSocketCtx {
* The Channel.
*/
private Channel channel;
public ControlAdapterSocketCtx(Long id, Channel ch) {
this.controlAdapterId = id;
this.channel = ch;
}
}

36
src/main/java/com/zjyr/beidouservice/service/AdapterProtocolService.java
View File

@ -1,15 +1,35 @@
package com.zjyr.beidouservice.service;
import com.zjyr.beidouservice.common.impl.SensorControlModeName;
import com.zjyr.beidouservice.common.impl.SensorControlTunnelName;
import com.zjyr.beidouservice.pojo.po.BeidouAdapterControlContent;
import com.zjyr.beidouservice.pojo.vo.binary.HeartProtocol;
import com.zjyr.beidouservice.pojo.vo.binary.QuerySensorProtocol;
import com.zjyr.beidouservice.pojo.vo.binary.SensorControlProtocol;
import com.zjyr.beidouservice.pojo.vo.binary.SensorControllContent;
import java.util.List;
/**
* The interface Adapter protocol service.
*/
public interface AdapterProtocolService {
/**
* Create heart protocol heart protocol.
*
* @return the heart protocol
*/
HeartProtocol createHeartProtocol();
List<SensorControllContent> createSensorCtrlInfo(List<BeidouAdapterControlContent> ctxList);
SensorControlProtocol createSensorControlProtocol(SensorControlTunnelName tunnelName,
byte[] tunnelAddr,
Integer month,
Integer days,
Integer hour,
Integer minute,
Integer province,
Integer city,
SensorControlModeName ctrlMode,
Integer ctrlCmd,
List<SensorControllContent> ctrlItems);
QuerySensorProtocol createSensorQueryProtocol(Integer province,
Integer city,
List<SensorControllContent> ctrlItems);
<T> Object createTransmissionProtocol(T proContent);
}

37
src/main/java/com/zjyr/beidouservice/service/BaidouAdapterService.java
View File

@ -1,47 +1,20 @@
package com.zjyr.beidouservice.service;
import com.zjyr.beidouservice.pojo.po.BeidouAdapterDevice;
import io.netty.channel.Channel;
import java.util.List;
/**
* The interface Baidou adapter service.
*/
public interface BaidouAdapterService {
/**
* Start beidou adapter.
*/
void startBeidouAdapter();
/**
* Send commond.
*
* @param <T> the type parameter
* @param adapterId the adapter id
* @param command the command
*/
<T> void sendCommond(Long adapterId, T command);
void sendCommond(Long adapterId, Object command);
void sendCommond(Channel ch, Object command);
/**
* Gets all adapter.
*
* @return the all adapter
*/
List<BeidouAdapterDevice> getAllAdapter();
/**
* Add new beidou adapter int.
*
* @param adapter the adapter
* @return the int
*/
int addNewBeidouAdapter(BeidouAdapterDevice adapter);
void addNewBeidouAdapter(BeidouAdapterDevice adapter);
/**
* Gets adapter by addr.
*
* @param addr the addr
* @return the adapter by addr
*/
BeidouAdapterDevice getAdapterByAddr(String addr);
}

71
src/main/java/com/zjyr/beidouservice/service/impl/BaidouSocketAdapterServiceImpl.java
View File

@ -4,14 +4,15 @@ import com.zjyr.beidouservice.adapter.impl.netty.impl.TcpServer;
import com.zjyr.beidouservice.adapter.impl.netty.notify.SocketNotifyEvent;
import com.zjyr.beidouservice.common.impl.BeidouAdapterTypeName;
import com.zjyr.beidouservice.mapper.ControlDeviceMapper;
import com.zjyr.beidouservice.misc.HelperUtils;
import com.zjyr.beidouservice.pojo.entry.ControlDevice;
import com.zjyr.beidouservice.pojo.po.BeidouAdapterDevice;
import com.zjyr.beidouservice.pojo.vo.ControlAdapterSocketCtx;
import com.zjyr.beidouservice.pojo.vo.binary.BaseBinaryProtocol;
import com.zjyr.beidouservice.pojo.vo.binary.ControllerStatus;
import com.zjyr.beidouservice.pojo.vo.binary.HeartProtocol;
import com.zjyr.beidouservice.pojo.vo.binary.MessageContent;
import com.zjyr.beidouservice.service.AdapterProtocolService;
import com.zjyr.beidouservice.service.BaidouAdapterService;
import io.netty.channel.Channel;
import jakarta.annotation.Nonnull;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
@ -26,36 +27,19 @@ import java.util.concurrent.ConcurrentHashMap;
import static com.zjyr.beidouservice.common.impl.SocketEventName.SOCKET_EVT_CONNECT;
/**
* The type Baidou socket adapter service.
*
* @param <T> the type parameter
*/
@Service
@Slf4j
public class BaidouSocketAdapterServiceImpl<T extends ApplicationEvent> implements BaidouAdapterService,
ApplicationListener<T> {
/**
* The Ctx map.
*/
private final ConcurrentHashMap<Long, ControlAdapterSocketCtx> ctxMap = new ConcurrentHashMap<>();
/**
* The Control device mapper.
*/
@Resource
ControlDeviceMapper controlDeviceMapper;
/**
* The Adapter protocol service.
*/
@Resource
AdapterProtocolService adapterProtocolService;
/**
* The Tcp server.
*/
@Resource
private TcpServer tcpServer;
@ -65,29 +49,56 @@ public class BaidouSocketAdapterServiceImpl<T extends ApplicationEvent> implemen
}
@Override
public <E> void sendCommond(Long adapterId, E command) {
public void sendCommond(Long adapterId, Object command) {
ControlAdapterSocketCtx ctx = ctxMap.get(adapterId);
if (ctx != null) {
sendCommond(ctx.getChannel(), command);
}
}
@Override
public void sendCommond(Channel ch, Object command) {
if (command instanceof HeartProtocol heart) {
tcpServer.sendData(ch, adapterProtocolService.createTransmissionProtocol(heart));
}
}
@Override
public void onApplicationEvent(@Nonnull T evtContent) {
if (evtContent instanceof SocketNotifyEvent event) {
log.info("Recvie {} event data {}", event.getSocketEvent(), event.getEvtMessage());
InetSocketAddress sa = (InetSocketAddress) event.getCtxChannel().remoteAddress();
switch (event.getSocketEvent()) {
case SOCKET_EVT_CONNECT, SOCKET_EVT_IDLE_TIMEOUT -> {
MessageContent<HeartProtocol> msgCont = MessageContent.<HeartProtocol>builder().msgBody(
adapterProtocolService.createHeartProtocol()).build();
event.getCtxChannel().writeAndFlush(BaseBinaryProtocol.<HeartProtocol>builder()
.msgContent(msgCont)
.build());
sendCommond(event.getCtxChannel(), adapterProtocolService.createHeartProtocol());
if (event.getSocketEvent() == SOCKET_EVT_CONNECT) {
InetSocketAddress sa = (InetSocketAddress) event.getCtxChannel().remoteAddress();
BeidouAdapterDevice a = BeidouAdapterDevice.builder()
.adapterType(BeidouAdapterTypeName.ADAPTER_SOCKET_TCP)
.adapterAddr(sa.getAddress().getHostAddress())
.build();
addNewBeidouAdapter(a);
ControlAdapterSocketCtx sockCtx = new ControlAdapterSocketCtx(a.getId(), event.getCtxChannel());
ctxMap.put(a.getId(), sockCtx);
}
}
case SOCKET_EVT_DISCONNECT -> {
for (ConcurrentHashMap.Entry<Long, ControlAdapterSocketCtx> entry : ctxMap.entrySet()) {
if (entry.getValue().getChannel().id() == event.getCtxChannel().id()) {
ctxMap.remove(entry.getKey());
}
}
}
case SOCKET_EVT_RECV -> {
if (event.getEvtMessage() instanceof ControllerStatus status) {
log.info("+++{}, {}, {}",
status.getWirelessStatus(),
status.getTelphoneStatus(),
HelperUtils.bytesToHexString(status.getBeidouSignalStrength()));
}
}
}
@ -119,7 +130,7 @@ public class BaidouSocketAdapterServiceImpl<T extends ApplicationEvent> implemen
}
@Override
public int addNewBeidouAdapter(BeidouAdapterDevice adapter) {
public void addNewBeidouAdapter(BeidouAdapterDevice adapter) {
ControlDevice device = ControlDevice.builder()
.id(adapter.getId())
.deviceType(adapter.getAdapterType())
@ -130,10 +141,10 @@ public class BaidouSocketAdapterServiceImpl<T extends ApplicationEvent> implemen
.build();
BeidouAdapterDevice dev = getAdapterByAddr(adapter.getAdapterAddr());
if (dev == null) {
return controlDeviceMapper.addControlDevice(device);
controlDeviceMapper.addControlDevice(device);
} else {
adapter.setId(dev.getId());
return controlDeviceMapper.upgradeControlDevice(device);
controlDeviceMapper.upgradeControlDevice(device);
}
}

74
src/main/java/com/zjyr/beidouservice/service/impl/BeidouSocketProtocolServiceImpl.java
View File

@ -1,13 +1,21 @@
package com.zjyr.beidouservice.service.impl;
import com.zjyr.beidouservice.common.impl.SensorControlModeName;
import com.zjyr.beidouservice.common.impl.SensorControlTunnelName;
import com.zjyr.beidouservice.pojo.po.BeidouAdapterControlContent;
import com.zjyr.beidouservice.pojo.vo.binary.BaseBinaryProtocol;
import com.zjyr.beidouservice.pojo.vo.binary.HeartProtocol;
import com.zjyr.beidouservice.pojo.vo.binary.MessageContent;
import com.zjyr.beidouservice.pojo.vo.binary.QuerySensorProtocol;
import com.zjyr.beidouservice.pojo.vo.binary.SensorControlProtocol;
import com.zjyr.beidouservice.pojo.vo.binary.SensorControllContent;
import com.zjyr.beidouservice.service.AdapterProtocolService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
/**
* The type Beidou socket protocol service.
*/
import java.util.ArrayList;
import java.util.List;
@Service
@Slf4j
public class BeidouSocketProtocolServiceImpl implements AdapterProtocolService {
@ -16,4 +24,64 @@ public class BeidouSocketProtocolServiceImpl implements AdapterProtocolService {
return new HeartProtocol();
}
@Override
public List<SensorControllContent> createSensorCtrlInfo(List<BeidouAdapterControlContent> ctxList) {
List<SensorControllContent> list = new ArrayList<>();
ctxList.forEach(v -> {
list.add(SensorControllContent.builder()
.districtsCode(v.getDistrictsCode().byteValue())
.controlAction(v.getControlAction())
.sensorId(v.getSensorId().shortValue())
.build());
});
return list;
}
@Override
public SensorControlProtocol createSensorControlProtocol(SensorControlTunnelName tunnelName,
byte[] tunnelAddr,
Integer month,
Integer days,
Integer hour,
Integer minute,
Integer province,
Integer city,
SensorControlModeName ctrlMode,
Integer ctrlCmd,
List<SensorControllContent> ctrlItems) {
return SensorControlProtocol.builder()
.controlTunnel(tunnelName)
.tunnelAddr(tunnelAddr)
.month(month.byteValue())
.days(days.byteValue())
.hour(hour.byteValue())
.minute(minute.byteValue())
.province(province.byteValue())
.city(city.byteValue())
.controlMode(ctrlMode)
.controlCommand(ctrlCmd.byteValue())
.nControlInfo(((Integer) ctrlItems.size()).byteValue())
.controllContents(new ArrayList<>(ctrlItems))
.build();
}
@Override
public QuerySensorProtocol createSensorQueryProtocol(Integer province,
Integer city,
List<SensorControllContent> ctrlItems) {
return QuerySensorProtocol.builder()
.province(province.byteValue())
.city(city.byteValue())
.nControlInfo(((Integer) ctrlItems.size()).byteValue())
.controllContents(new ArrayList<>(ctrlItems))
.build();
}
@Override
public <T> Object createTransmissionProtocol(T proContent) {
MessageContent<T> msgCont = MessageContent.<T>builder().msgBody(proContent).build();
return BaseBinaryProtocol.<T>builder().msgContent(msgCont).build();
}
}