提交通讯模块

gateway/forecast-iec104/pom.xml

@@ -0,0 +1,31 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>forecast-iec104</artifactId>
+    <description>iec104模块</description>
+
+    <parent>
+        <artifactId>forecast-gateway</artifactId>
+        <groupId>cn.oleauto</groupId>
+        <version>${revision}</version>
+    </parent>
+
+    <dependencies>
+        <dependency>
+            <groupId>cn.oleauto</groupId>
+            <artifactId>forecast-iec60870</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>cn.oleauto</groupId>
+            <artifactId>forecast-protocol-common</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>cn.hutool</groupId>
+            <artifactId>hutool-all</artifactId>
+        </dependency>
+    </dependencies>
+
+</project>

gateway/forecast-iec104/src/main/java/cn/oleauto/iec104/ProtocolImpl.java

@@ -0,0 +1,578 @@
+package cn.oleauto.iec104;
+
+import cn.hutool.core.convert.Convert;
+import cn.hutool.core.thread.ThreadUtil;
+import cn.oleauto.biz.domain.BizYkLog;
+import cn.oleauto.biz.domain.BizYtLog;
+import cn.oleauto.biz.service.IBizYkLogService;
+import cn.oleauto.biz.service.IBizYtLogService;
+import cn.oleauto.common.enums.ControlStepType;
+import cn.oleauto.common.enums.ControlType;
+import cn.oleauto.common.enums.MessageOwnerType;
+import cn.oleauto.common.enums.OperateStateType;
+import cn.oleauto.common.exception.ServiceException;
+import cn.oleauto.common.utils.StringUtils;
+import cn.oleauto.common.utils.spring.SpringUtils;
+import cn.oleauto.iec104.util.DataConverter;
+import cn.oleauto.iec60870.*;
+import cn.oleauto.iec60870.ie.*;
+import cn.oleauto.protocol.common.IChannel;
+import cn.oleauto.protocol.common.cache.DefaultMessageCache;
+import cn.oleauto.protocol.common.dto.ReportMessageDto;
+import cn.oleauto.protocol.common.dto.ReportValueDtoList;
+import cn.oleauto.protocol.common.event.DefaultEventArgs;
+import cn.oleauto.protocol.common.event.EventType;
+import cn.oleauto.protocol.common.event.TransmitMessageEventArgs;
+import cn.oleauto.protocol.common.mq.MQCenter;
+import cn.oleauto.protocol.common.protocol.AbstractProtocol;
+import cn.oleauto.protocol.common.transmit.TransmitState;
+import io.netty.buffer.ByteBufUtil;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.util.ConcurrentReferenceHashMap;
+
+import java.io.IOException;
+import java.io.StringReader;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.concurrent.BlockingDeque;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.LinkedBlockingDeque;
+
+@Slf4j
+public class ProtocolImpl extends AbstractProtocol{
+    private ConnectionSettings settings = new ConnectionSettings();
+    private DefaultMessageCache messageCache = new DefaultMessageCache(1024 * 20);
+    private Thread thParse;
+    private Thread thTask;
+    private volatile boolean isStop = false;
+    private volatile short sendSq = 0;
+    private volatile short recSq = 0;
+    private byte[] packageBuffer = new byte[255];
+    private static final int LAZY_AMOUNT = 256;
+    private BlockingDeque<LazyRequest> waitSendApdus = new LinkedBlockingDeque<>(LAZY_AMOUNT);
+    private ConcurrentMap<String, WaitResponseRequest> waitResponseMap = new ConcurrentReferenceHashMap<>();
+    private long last100 = 0;
+    private long last103 = 0;
+    private int tm100 = 120_000;
+    private int tm103 = 86400_000/2;
+
+    public ProtocolImpl(IChannel channel){
+        super(channel);
+    }
+
+    @Override
+    protected void onTransmitConnected(DefaultEventArgs e) {
+        sendSq = recSq = 0;
+        messageCache.clear();
+        notifyTransmitState(1);
+    }
+
+    @Override
+    protected void onTransmitDisconnected(DefaultEventArgs e) {
+        sendSq = recSq = 0;
+        messageCache.clear();
+        notifyTransmitState(0);
+    }
+
+    @Override
+    protected void onTransmitSent(TransmitMessageEventArgs e) {
+        notifyMessage(EventType.PROTOCOL_SENT, e.getData());
+    }
+
+    @Override
+    protected void onTransmitReceived(TransmitMessageEventArgs e) {
+        if(!messageCache.append(e.getData())){
+            log.warn("message cache is full, lost={}", ByteBufUtil.hexDump(e.getData()));
+        }
+        synchronized (this) {
+            notify();
+        }
+    }
+
+    @Override
+    public int start(String config) {
+        log.info("protocol start channel={}", channel.getName());
+        if(thParse != null || thTask != null){
+            return 0;
+        }
+        try {
+            Properties properties = new Properties();
+            properties.load(new StringReader(config));
+            // 总召唤间隔
+            tm100 = Convert.toInt(properties.get("ti1000Interval"), tm100);
+            settings.setCotFieldLength(Convert.toInt(properties.get("cotFieldLength"), 2));
+            settings.setCommonAddressFieldLength(Convert.toInt(properties.get("commonAddressFieldLength"), 2));
+            settings.setIoaFieldLength(Convert.toInt(properties.get("ioaFieldLength"), 3));
+            settings.setCommonAddress(Convert.toInt(properties.get("commonAddress"), 1));
+            settings.setOriginAddress(Convert.toInt(properties.get("originAddress"), 0));
+
+            buildPointRelation();
+
+            thParse = new Thread(()->{
+              this.parseMessageTask();
+            });
+            thParse.setName("ch-" + channel.getName() + "-protocol-parse");
+            thParse.start();
+
+            thTask = new Thread(()->{
+                this.timerTask();
+            });
+            thTask.setName("ch-" + channel.getName() + "-protocol-task");
+            thTask.start();
+
+            messageCache.clear();
+        } catch (Exception ex){
+            log.error("start protocol failed, config={} ex={}",config, ex);
+            return -1;
+        }
+        return 0;
+    }
+
+    @Override
+    public void stop() {
+        isStop = true;
+        thTask = null;
+        thParse = null;
+        messageCache.clear();
+        synchronized (this) {
+            notifyAll();
+        }
+    }
+
+    @Override
+    public List<Object> decode(byte[] data) {
+        APdu apdu = APdu.decode(data, settings);
+        if(apdu.getApciType() == APdu.ApciType.I_FORMAT){
+            int theirSn = apdu.getSendSeqNumber();
+            if(recSq == theirSn){
+                decodeAsdu(apdu);
+                // 发回一个s帧
+                recSq = (short) (theirSn + 1);
+                sendLazyRequest(new LazyRequest(APdu.ApciType.S_FORMAT, null, null, null, 0, 0));
+            }
+            if(theirSn < recSq || (theirSn > recSq && Math.abs(theirSn - recSq) > 32700)){
+                // 重复的情况
+                return List.of();
+            }
+            else {
+                // 丢了报文
+            }
+        } else if(apdu.getApciType() == APdu.ApciType.S_FORMAT){
+            // 监视帧
+        } else {
+            // U帧
+        }
+        return List.of();
+    }
+    public void notifyMessage(EventType eventType, byte[] data){
+        String frame = switch (data[2] & 0x3){
+            case 3->"U";
+            case 1->"S";
+            default -> "I";
+        };
+        log.info("channel={} frame={} ti={} action={} bytes ={}", channel.getName(), frame, (data.length > 6? data[6]: 0), eventType, ByteBufUtil.hexDump(data));
+        String fun = data.length <= 6? frame : Byte.toString(data[6]);
+        MessageOwnerType owner = (eventType == EventType.TRANSMIT_RECEIVED || eventType == EventType.PROTOCOL_RECEIVED)? MessageOwnerType.DEVICE : MessageOwnerType.SYSTEM;
+        ReportMessageDto channelMessageDto = new ReportMessageDto(channel.getId(), null, owner, fun, data, System.currentTimeMillis());
+        MQCenter.getInstance().send(ReportMessageDto.REPORT_MESSAGE, channelMessageDto);
+    }
+    private void decodeAsdu(APdu apdu){
+        try {
+            ASdu asdu = ASdu.decode(apdu.getASduBuffer(), settings);
+            ASduType typeIdentification = asdu.getTypeIdentification();
+            switch (typeIdentification){
+                case C_SE_NC_1, C_SE_NA_1-> receiveYtResponse(asdu);
+                case C_SC_NA_1, C_DC_NA_1-> receiveYkResponse(asdu);
+            }
+            ReportValueDtoList realData = DataConverter.toRealData(this, asdu);
+            if(realData.size() > 0) {
+                notifyRealValue(realData);
+            }
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private void receiveYtResponse(ASdu asdu){
+        String wrrId = getWaitResponseRequestId(asdu.getTypeIdentification().getId(), asdu.getInformationObjects()[0].getInformationObjectAddress(), "");
+        WaitResponseRequest waitResponseRequest = waitResponseMap.remove(wrrId);
+        if(waitResponseRequest != null && waitResponseRequest.extraObject instanceof BizYtLog bizYtLog){
+            CauseOfTransmission causeOfTransmission = asdu.getCauseOfTransmission();
+            bizYtLog.setState((CauseOfTransmission.ACTIVATION_CON == causeOfTransmission && !asdu.isNegativeConfirm())
+                    ? OperateStateType.ES.getType()
+                    : OperateStateType.EF.getType());
+            IBizYtLogService bean = SpringUtils.getBean(IBizYtLogService.class);
+            bean.updateBizYtLog(bizYtLog);
+        }
+    }
+
+    private void receiveYkResponse(ASdu asdu) {
+        String wrrId = getWaitResponseRequestId(asdu.getTypeIdentification().getId(), asdu.getInformationObjects()[0].getInformationObjectAddress(), "");
+        WaitResponseRequest waitResponseRequest = waitResponseMap.remove(wrrId);
+        if(waitResponseRequest != null && waitResponseRequest.extraObject instanceof BizYkLog bizYkLog){
+            CauseOfTransmission causeOfTransmission = asdu.getCauseOfTransmission();
+            if(causeOfTransmission == CauseOfTransmission.ACTIVATION_CON){
+                if(asdu.isNegativeConfirm()){
+                    if(StringUtils.equals(bizYkLog.getState(), OperateStateType.SE.getType())){
+                        bizYkLog.setState(OperateStateType.SF.getType());
+                    } else {
+                        bizYkLog.setState(OperateStateType.EF.getType());
+                    }
+                }
+                else {
+                    if(StringUtils.equals(bizYkLog.getState(), OperateStateType.SE.getType())){
+                        bizYkLog.setState(OperateStateType.SS.getType());
+                    } else {
+                        bizYkLog.setState(OperateStateType.ES.getType());
+                    }
+                }
+            } else if(causeOfTransmission == CauseOfTransmission.DEACTIVATION_CON){
+                if(asdu.isNegativeConfirm()){
+                    bizYkLog.setState(OperateStateType.CF.getType());
+                } else {
+                    bizYkLog.setState(OperateStateType.CS.getType());
+                }
+            }
+
+            IBizYkLogService bean = SpringUtils.getBean(IBizYkLogService.class);
+            bean.updateBizYkLog(bizYkLog);
+        }
+    }
+
+    @Override
+    public void write(Object obj) {
+        if(null == transmit || transmit.state() != TransmitState.CONNECTED){
+            throw new RuntimeException("没有连接上设备");
+        }
+        if(obj instanceof BizYtLog bizYtLog){
+            addYtToLazy(bizYtLog);
+        } else if(obj instanceof BizYkLog bizYkLog){
+            addYkToLazy(bizYkLog);
+        }
+    }
+
+    private void parseMessageTask(){
+        log.info("start channel protocol paserTask, channel={}", channel.getName());
+        byte[] head = new byte[APdu.APCI_LENGTH];
+        while(!isStop){
+            boolean noData = true;
+            if(messageCache.pick(head)){
+                int i = 0;
+                for(; head[i] != APdu.START_FLAG && i < head.length; ++i){
+                }
+                if(i > 0){
+                    // 首字节不是0x68,
+                    messageCache.remove(i);
+                    continue;
+                }
+                int len = head[1] & 0xff;
+                if(len < APdu.MIN_APDU_LENGTH || len > APdu.MAX_APDU_LENGTH){
+                    // 报文长度字节有错误了
+                    messageCache.remove(2);
+                    continue;
+                }
+
+                if(len == APdu.MIN_APDU_LENGTH){
+                    // 如果是U或者S帧
+                    messageCache.remove(APdu.APCI_LENGTH);
+                    decode(head);
+                }
+                else if(len + 2 <= messageCache.size()){
+                    // I帧，并且数据够，则取出解析
+                    byte[] message = new byte[len + 2];
+                    messageCache.take(message);
+                    notifyMessage(EventType.PROTOCOL_RECEIVED, message);
+                    decode(message);
+                    noData = false;
+                }
+            }
+
+            if(noData){
+                synchronized (this) {
+                    try {
+                        this.wait(1000);
+                    } catch (InterruptedException ex) {
+                        log.warn("wait exception={}", ex);
+                    }
+                }
+            }
+        }
+
+        log.info("end channel protocol paserTask, channel={}", channel.getName());
+    }
+
+    private void timerTask(){
+        log.info("start channel protocol timerTask, channel={}", channel.getName());
+        while(!isStop){
+            try{
+                TransmitState transmitState = transmit.state();
+                if(transmitState == TransmitState.DISCONNECTED || transmitState == TransmitState.STARTED){
+                    if(transmit.connect() == 0){
+                        ThreadUtil.sleep(1000);
+                        waitSendApdus.push(new LazyRequest(APdu.ApciType.STARTDT_ACT, null, null, null, System.currentTimeMillis(), 0));
+                    }else{
+                        ThreadUtil.sleep(10000);
+                    }
+                    ThreadUtil.sleep(10000);
+                    continue;
+                }
+
+                LazyRequest lazyRequest = waitSendApdus.poll();
+                if(lazyRequest != null){
+                    sendLazyRequest(lazyRequest);
+                    ThreadUtil.sleep(100);
+                    continue;
+                }
+
+                Long curTm = System.currentTimeMillis();
+                clearWaitResponseRequest();
+                if(curTm - last100 > tm100){
+                    send100();
+                    last100 = curTm;
+                }
+                else if(curTm - last103 > tm103){
+                    send103();
+                    last103 = curTm;
+                }
+
+                ThreadUtil.sleep(1000);
+            } catch (Exception ex)
+            {
+                log.error("protocol timerTask occur exception, channel={}, ex={}",channel.getName(), ex.getMessage());
+            }
+        }
+        log.info("end channel protocol timerTask, channel={}", channel.getName());
+    }
+
+    private void sendLazyRequest(LazyRequest lazyRequest) {
+        boolean exist = false;
+        long nowTm = System.currentTimeMillis();
+        if(StringUtils.isBlank(lazyRequest.wrpId) ) {
+            sendApdu(lazyRequest.apciType, lazyRequest.getAsdu());
+            return;
+        } else if(!waitResponseMap.containsKey(lazyRequest.wrpId)){
+            sendApdu(lazyRequest.apciType, lazyRequest.getAsdu());
+        }else {
+            if(nowTm - lazyRequest.beginTm > lazyRequest.tmOutMs){
+                exist = true;
+            }
+            else {
+                // 重新等发送
+                waitSendApdus.push(lazyRequest);
+                return;
+            }
+        }
+
+        if(lazyRequest.extraObject instanceof BizYtLog bizYtLog){
+            if(exist){
+                bizYtLog.setState(OperateStateType.AL.getType());
+            }
+            else if(bizYtLog.getTmOut() <= 0) {
+                bizYtLog.setState(OperateStateType.ES.getType());
+            } else {
+                bizYtLog.setState(OperateStateType.EX.getType());
+                waitResponseMap.put(lazyRequest.wrpId, new WaitResponseRequest(lazyRequest.wrpId, bizYtLog, nowTm));
+            }
+            IBizYtLogService bean = SpringUtils.getBean(IBizYtLogService.class);
+            bean.updateBizYtLog(bizYtLog);
+        } else if(lazyRequest.extraObject instanceof BizYkLog bizYkLog){
+            if(exist){
+                bizYkLog.setState(OperateStateType.AL.getType());
+            }
+            else if(Objects.equals(OperateStateType.SE.getType(), bizYkLog.getState())
+                    && Objects.equals(bizYkLog.getControlStep(), ControlStepType.TWO.getType())) {
+                waitResponseMap.put(lazyRequest.wrpId, new WaitResponseRequest(lazyRequest.wrpId, bizYkLog, nowTm));
+            } else {
+                if(bizYkLog.getTmOut() <= 0){
+                    bizYkLog.setState(OperateStateType.ES.getType());
+                } else {
+                    waitResponseMap.put(lazyRequest.wrpId, new WaitResponseRequest(lazyRequest.wrpId, bizYkLog, nowTm));
+                }
+            }
+            IBizYkLogService bean = SpringUtils.getBean(IBizYkLogService.class);
+            bean.updateBizYkLog(bizYkLog);
+        }
+    }
+
+    private void clearWaitResponseRequest() {
+        for(var val : waitResponseMap.values()){
+            if(val.extraObject instanceof BizYtLog bizYtLog){
+                if(System.currentTimeMillis() - val.tmMs > bizYtLog.getTmOut() * 1000){
+                    if(waitResponseMap.remove(val.id) != null) {
+                        bizYtLog.setState(OperateStateType.ET.getType());
+                        IBizYtLogService bean = SpringUtils.getBean(IBizYtLogService.class);
+                        bean.updateBizYtLog(bizYtLog);
+                    }
+                }
+            } else if(val.extraObject instanceof BizYkLog bizYkLog){
+                if(System.currentTimeMillis() - val.tmMs > bizYkLog.getTmOut() * 1000) {
+                    if(waitResponseMap.remove(val.id)!=null) {
+                        if(StringUtils.equals(bizYkLog.getState(), OperateStateType.SE.getType())){
+                            bizYkLog.setState(OperateStateType.ST.getType());
+                        } else if(StringUtils.equals(bizYkLog.getState(), OperateStateType.CA.getType())){
+                            bizYkLog.setState(OperateStateType.CT.getType());
+                        } else {
+                            bizYkLog.setState(OperateStateType.ET.getType());
+                        }
+                        IBizYkLogService bean = SpringUtils.getBean(IBizYkLogService.class);
+                        bean.updateBizYkLog(bizYkLog);
+                    }
+                }
+            }
+        }
+    }
+
+    private void sendApdu(APdu.ApciType type,  ASdu aSdu){
+        byte[] sendBuf = null;
+        synchronized (packageBuffer) {
+            APdu apdu = null;
+            if(type == APdu.ApciType.I_FORMAT){
+                int asduLength = aSdu.encode(packageBuffer, 0, settings);
+                byte[] asduBufferCut = Arrays.copyOf(packageBuffer, asduLength);
+                apdu = new APdu(sendSq, recSq, type, asduBufferCut);
+                sendSq = (short)(sendSq+1);
+            } else {
+                apdu = new APdu(sendSq, recSq, type);
+            }
+            int len = apdu.encode(packageBuffer, settings);
+            sendBuf = Arrays.copyOf(packageBuffer, len);
+        }
+        if(sendBuf != null){
+            transmit.send(sendBuf);
+        }
+    }
+
+    private void send100(){
+        ASdu aSdu = new ASdu(ASduType.C_IC_NA_1,
+                false,
+                CauseOfTransmission.ACTIVATION,
+                false,
+                false,
+                0,
+                1,
+                new InformationObject(0, new IeQualifierOfInterrogation(20)));
+        sendApdu(APdu.ApciType.I_FORMAT, aSdu);
+    }
+
+    private void send103() {
+        IeTime56 time = new IeTime56(System.currentTimeMillis());
+        InformationObject io = new InformationObject(0, time);
+        ASdu asdu = new ASdu(ASduType.C_CS_NA_1,
+                false,
+                CauseOfTransmission.ACTIVATION,
+                false,
+                false,
+                0,
+                settings.getCommonAddress(),
+                 io);
+        sendApdu(APdu.ApciType.I_FORMAT, asdu);
+    }
+
+    private void addYtToLazy(BizYtLog bizYtLog){
+        Integer iFunCode = Convert.toInt(bizYtLog.getFun());
+        if(iFunCode == null){
+            throw new ServiceException("fun value is invalid, val=" +  bizYtLog.getFun());
+        }
+        Integer addr = Convert.toInt(bizYtLog.getAddrCode());
+        if(addr == null){
+            throw new ServiceException("addr code value is invalid, addrCode=" + bizYtLog.getAddrCode());
+        }
+        Float ytVal = Convert.toFloat(bizYtLog.getControlVal());
+        if(ytVal == null){
+            throw new ServiceException("yt value is invalid, val=" +  bizYtLog.getControlVal());
+        }
+
+        IeQualifierOfSetPointCommand qualifier = new IeQualifierOfSetPointCommand(0, false);
+        InformationObject io;
+        ASduType aSduType = ASduType.typeFor(iFunCode);
+        if(aSduType == ASduType.C_SE_NC_1) {
+            IeShortFloat val = new IeShortFloat(ytVal);
+            io = new InformationObject(addr.intValue(), val, qualifier);
+        } else if(aSduType == ASduType.C_SE_NA_1) {
+            IeNormalizedValue val = new IeNormalizedValue(ytVal.intValue());
+            io = new InformationObject(addr.intValue(), val, qualifier);
+        } else {
+            throw new ServiceException("fun value is invalid, val=" +  bizYtLog.getFun());
+        }
+        ASdu aSdu = new ASdu(aSduType,
+                false,
+                CauseOfTransmission.ACTIVATION,
+                false,
+                false,
+                settings.getOriginAddress(),
+                settings.getCommonAddress(),
+                io);
+        String wrqId = getWaitResponseRequestId(aSduType.getId(), addr.intValue(), "");
+        waitSendApdus.push(new LazyRequest(APdu.ApciType.I_FORMAT, aSdu, bizYtLog, wrqId, System.currentTimeMillis(), bizYtLog.getTmOut() * 1000));
+    }
+
+    private void addYkToLazy(BizYkLog bizYkLog) {
+        Integer ykVal = null;
+        if(Objects.equals(bizYkLog.getControlVal(), ControlType.OPEN.getType())){
+            ykVal = 0;
+        } else if(Objects.equals(bizYkLog.getControlVal(), ControlType.CLOSE.getType())){
+            ykVal = 1;
+        }
+        if(ykVal == null){
+            log.error("yk value is invalid, val={}", bizYkLog.getControlVal());
+            throw new ServiceException("yk value is invalid, val=" + bizYkLog.getControlVal());
+        }
+
+        Integer addr = Convert.toInt(bizYkLog.getAddrCode());
+        if(addr == null){
+            log.error("addr code value is invalid, addrCode={}", bizYkLog.getAddrCode());
+            throw new ServiceException("addr code value is invalid, addrCode=" + bizYkLog.getAddrCode());
+        }
+        boolean isSelect = OperateStateType.SE.getType().equals(bizYkLog.getState());
+
+        int fun = Convert.toInt(bizYkLog.getFun(), 45);
+        InformationElement val;
+        ASduType ti = ASduType.C_SC_NA_1;
+        if(fun == ASduType.C_DC_NA_1.getId()){
+            val = new IeDoubleCommand((ykVal.intValue()!=0? IeDoubleCommand.DoubleCommandState.ON: IeDoubleCommand.DoubleCommandState.OFF), 0, isSelect);
+            ti = ASduType.C_DC_NA_1;
+        } else {
+            val = new IeSingleCommand(ykVal.intValue()!=0, 0, isSelect);
+        }
+        CauseOfTransmission cot = StringUtils.equals(OperateStateType.CA.getType(), bizYkLog.getState())? CauseOfTransmission.DEACTIVATION: CauseOfTransmission.ACTIVATION;
+        InformationObject io = new InformationObject(addr, val);
+        ASdu aSdu = new ASdu(ti,
+                false,
+                cot,
+                false,
+                false,
+                settings.getOriginAddress(),
+                settings.getCommonAddress(),
+                io);
+        String wrqId = getWaitResponseRequestId(ti.getId(), addr.intValue(), "");
+        waitSendApdus.push(new LazyRequest(APdu.ApciType.I_FORMAT, aSdu, bizYkLog, wrqId, System.currentTimeMillis(), bizYkLog.getTmOut() * 1000));
+    }
+
+    static String getWaitResponseRequestId(int cmd, int ioa, String val){
+        return "" + cmd + "-" + ioa;
+    }
+
+    @Data
+    @AllArgsConstructor
+    static class LazyRequest{
+        private APdu.ApciType apciType;
+        private ASdu asdu;
+        private Object extraObject;
+        private String wrpId;
+        private long beginTm;
+        private long tmOutMs;
+    }
+
+    @Data
+    @AllArgsConstructor
+    static class WaitResponseRequest{
+        private String id;
+        private Object extraObject;
+        // 开始时间
+        private long tmMs;
+    }
+}

gateway/forecast-iec104/src/main/java/cn/oleauto/iec104/util/DataConverter.java

@@ -0,0 +1,110 @@
+package cn.oleauto.iec104.util;
+import cn.oleauto.iec104.ProtocolImpl;
+import cn.oleauto.iec60870.ie.*;
+import cn.oleauto.protocol.common.dto.ReportValueDto;
+import cn.oleauto.common.enums.PointType;
+import cn.oleauto.iec60870.ASdu;
+import cn.oleauto.iec60870.ASduType;
+import cn.oleauto.protocol.common.dto.ReportValueDtoList;
+import cn.oleauto.protocol.common.protocol.AbstractProtocol;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.function.BiFunction;
+
+
+@Slf4j
+public class DataConverter {
+    /**
+     * const byte Const_AsduType_1_M_SP_NA_1 = 1;        // 单点信息
+     * const byte Const_AsduType_9_M_ME_NA_1 = 9;        // 测量值，规一化值
+     * const byte Const_AsduType_15_M_IT_NA_1 = 15;      // 累计量
+     * const byte Const_AsduType_30_M_SP_TB_1 = 30;      // 带时标CP56Time2a的单点信息
+     * const byte Const_AsduType_45_C_SC_NA_1 = 45;      // 单命令
+     * const byte Const_AsduType_46_C_DC_NA_1 = 46;      // 双命令
+     * const byte Const_AsduType_48_C_SE_NA_1 = 48;      // 设定命令，归一化值
+     * const byte Const_AsduType_70_M_EI_NA_1 = 70;      // 初始化结束
+     * const byte Const_AsduType_100_C_IC_NA_1 = 100;    // 总召唤命令
+     * const byte Const_AsduType_101_C_CI_NA_1 = 101;    // 电能脉冲召唤命令
+     * const byte Const_AsduType_102_C_SE_NA_1 = 102;    // 单个参数读取命令
+     * const byte Const_AsduType_103_C_CS_NA_1 = 103;    // 时钟同步命令
+     * const byte Const_AsduType_104_C_TS_NA_1 = 104;    // 测试命令
+     * const byte Const_AsduType_105_C_PC_NA_1 = 105;    // 复位进程命令
+     * const byte Const_AsduType_50_C_SE_NC_1 = 50;      // 设定值命令
+     * const byte Const_AsduType_55_C_YT_NA_1 = 55;      // 遥调定值修改命令
+     * const byte Const_AsduType_108_C_YT_NA_1 = 108;    // 定值召唤召唤命令
+     */
+    public static ReportValueDtoList toRealData(ProtocolImpl protocol, ASdu asdu){
+        ReportValueDtoList dtos = new ReportValueDtoList();
+        long ts = System.currentTimeMillis();
+        for(int i = 0; i < asdu.getInformationObjects().length; ++i){
+            InformationObject informationObject = asdu.getInformationObjects()[i];
+            toRealData(dtos, ts, protocol, asdu.getTypeIdentification(), informationObject);
+        }
+        return dtos;
+    }
+
+    public static void toRealData(ReportValueDtoList values, long tm, ProtocolImpl protocol, ASduType aSduType, InformationObject informationObject){
+
+        switch (aSduType){
+            case M_SP_NA_1, M_SP_TB_1 ->
+                    torealData(values, tm, PointType.YX, protocol, informationObject, (ReportValueDto value, InformationElement[] elements) ->{
+                        if(elements[0] instanceof IeSinglePointWithQuality ieValue){
+                            value.setIValue(ieValue.isOn()? 1 :0);
+                            return true;
+                        }
+                        return false;
+                    });
+            case M_ME_NA_1 ->
+                    torealData(values, tm, PointType.YC, protocol, informationObject, (ReportValueDto value, InformationElement[] elements) ->{
+                        if(elements[0] instanceof IeNormalizedValue ieValue){
+                            value.setDValue(ieValue.getNormalizedValue());
+                            return true;
+                        }
+                        return false;
+                    });
+            case M_ME_NC_1 ->
+                    torealData(values, tm, PointType.YC, protocol, informationObject, (ReportValueDto value, InformationElement[] elements) ->{
+                        if(elements[0] instanceof IeShortFloat ieValue){
+                            value.setDValue(ieValue.getValue());
+                            return true;
+                        }
+                        return false;
+                    });
+            default -> {
+                if(aSduType.getId() < 145) {
+                    log.info("can not convert ti={}-{}", aSduType.getId(), aSduType);
+                }
+            }
+        }
+    }
+
+    private static void torealData(ReportValueDtoList valueDtos,
+                                   long tm,
+                                   PointType pointType,
+                                   ProtocolImpl protocol,
+                                   InformationObject informationObject,
+                                   BiFunction<ReportValueDto, InformationElement[], Boolean> converter) {
+        int address = informationObject.getInformationObjectAddress();
+        InformationElement[][] informationElements = informationObject.getInformationElements();
+
+        for(int i = 0; i < informationElements.length; ++i) {
+            AbstractProtocol.PointAddrCode pointAddrCode = protocol.getByAddr(address + i, pointType);
+            if (pointAddrCode == null) {
+                if (log.isDebugEnabled()) {
+                    log.debug("can not find PointAddrCode for addr={} pointType={}", address + i, pointType);
+                }
+                continue;
+            }
+
+            ReportValueDto val = new ReportValueDto();
+            val.setChannelId(pointAddrCode.getChannelId());
+            val.setDeviceId(pointAddrCode.getDeviceId());
+            val.setPointCode(pointAddrCode.getCode());
+            val.setPointType(pointType);
+            val.setTmMs(tm);
+            if(converter.apply(val, informationElements[i])) {
+                valueDtos.add(val);
+            }
+        }
+    }
+}

gateway/forecast-transmit/pom.xml

@@ -0,0 +1,31 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>forecast-transmit</artifactId>
+    <description>传输功能模块</description>
+
+    <parent>
+        <artifactId>forecast-gateway</artifactId>
+        <groupId>cn.oleauto</groupId>
+        <version>${revision}</version>
+    </parent>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-all</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>cn.hutool</groupId>
+            <artifactId>hutool-all</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>cn.oleauto</groupId>
+            <artifactId>agvc-protocol-common</artifactId>
+        </dependency>
+    </dependencies>
+
+</project>

gateway/forecast-transmit/src/main/java/cn/oleauto/transmit/tcp/TcpClient.java

@@ -0,0 +1,204 @@
+package cn.oleauto.transmit.tcp;
+
+import cn.hutool.core.thread.ThreadUtil;
+import cn.oleauto.common.core.text.Convert;
+import cn.oleauto.common.utils.StringUtils;
+import cn.oleauto.protocol.common.IChannel;
+import cn.oleauto.protocol.common.event.*;
+import cn.oleauto.protocol.common.transmit.AbstractTransmit;
+import cn.oleauto.protocol.common.transmit.TransmitState;
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufUtil;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.*;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.timeout.IdleStateHandler;
+import io.netty.util.ReferenceCountUtil;
+import io.netty.util.concurrent.DefaultThreadFactory;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import java.io.StringReader;
+import java.util.Properties;
+
+/**
+ * @author oleauto
+ * @date
+ */
+@Slf4j
+public class TcpClient extends AbstractTransmit {
+
+    @Getter
+    private String host;
+    @Getter
+    private int port;
+    private final IChannel channel;
+
+    private volatile boolean isStop = false;
+    private final Bootstrap bootstrap;
+    private final EventLoopGroup workEventLoop;
+    private ChannelHandlerContext channelHandlerContext;
+    private EventCenter<AbstractEventArgs> eventCenter;
+
+    public TcpClient(IChannel channel)
+    {
+        this.channel = channel;
+
+//        DefaultThreadFactory workerThreadFactory = new DefaultThreadFactory("channel-worker-" + channel.getName());
+//        workEventLoop = new NioEventLoopGroup(workerThreadFactory);
+        workEventLoop = new NioEventLoopGroup(1);
+        bootstrap = new Bootstrap();
+        // bootstrap.option(ChannelOption.SO_RCVBUF, 1024 * 1024);//缓冲区大小
+        bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 2000);
+        TcpClient that = this;
+        this.bootstrap.group(workEventLoop)
+                .channel(NioSocketChannel.class)
+                .option(ChannelOption.SO_KEEPALIVE, true)
+                .handler(new IdleStateHandler(0, 0, 60))
+                .handler(new ChannelInitializer<Channel>() {
+                    @Override
+                    protected void initChannel(Channel ch) {
+                        ch.pipeline().addLast(new ProtocolTcpClientHandler(that));
+                    }
+                });
+    }
+
+    @Override
+    public int start(String config) {
+        log.info("{} channel start", channel.getName());
+
+        Properties properties = new Properties();
+        try {
+            properties.load(new StringReader(config));
+            host = properties.getProperty("host");
+            if(StringUtils.isBlank(host)){
+                throw new IllegalArgumentException("not exist host");
+            }
+            port = Convert.toInt(properties.getProperty("port"), 0);
+            if(port == 0){
+                throw new IllegalArgumentException("not exist port");
+            }
+            isStop = false;
+        } catch (Exception ex){
+            log.error("channel={} config is invalid, content={}, ex={}", channel.getName(), config, ex);
+            return -1;
+        }
+        state = TransmitState.STARTED;
+        return 0;
+    }
+
+    @Override
+    public void stop() {
+        log.info("{} channel stop", channel.getName());
+        isStop = true;
+        disconnect();
+        state = TransmitState.STOPPED;
+    }
+
+    @Override
+    public int connect(){
+        ChannelFuture channelFuture = null;
+        try {
+            disconnect();
+            // bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000);// 没有效果
+            channelFuture = bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
+                    .connect(host, port)
+                    .sync();
+            return 0;
+        } catch (Exception e) {
+            log.error("channel={} connect failed, host={}, port={}, ex={}", channel.getName(), host, port, e);
+            if(channelFuture != null){
+                channelFuture.channel().close();
+            }
+        }
+        return -1;
+    }
+
+    @Override
+    public void disconnect(){
+        ChannelHandlerContext ctx = channelHandlerContext;
+        if(ctx != null){
+            ctx.disconnect();
+        }
+    }
+
+    @Override
+    public int send(byte[] data) {
+        if(data != null || data.length > 0){
+            ChannelHandlerContext context = channelHandlerContext;
+            if(context != null) {
+                ByteBuf buf = Unpooled.wrappedBuffer(data);
+                context.channel().writeAndFlush(buf);
+                if(log.isDebugEnabled()) {
+                    log.debug("channel={}, send bytes={} ", channel.getName(), ByteBufUtil.hexDump(data));
+                }
+                notifySent(data);
+                return data.length;
+            }
+        }
+        return 0;
+    }
+
+    @Override
+    public int send(byte[] data, int tmOUtMs) {
+        return send(data);
+    }
+
+    @Override
+    public byte[] receive() {
+        return new byte[0];
+    }
+
+    @Override
+    public byte[] receive(int tmOutMs) {
+        return new byte[0];
+    }
+
+    /**
+     * ProtocolTcpServerHandler 类继承自 ChannelInboundHandlerAdapter，
+     * 用于处理 TCP 服务端的入站事件。这个类是一个空实现，继承者可以根据需要重写相应的方法来处理事件。
+     */
+    static class ProtocolTcpClientHandler extends ChannelInboundHandlerAdapter {
+
+        private final TcpClient client;
+        public ProtocolTcpClientHandler(TcpClient client) {
+            this.client = client;
+        }
+
+        @Override
+        public void channelActive(ChannelHandlerContext ctx) {
+            ChannelHandlerContext oldCtx = client.channelHandlerContext;
+            if(oldCtx != null){
+                oldCtx.channel().close();
+            }
+
+            client.channelHandlerContext = ctx;
+            log.info("channel={} connected, ctx={}", client.channel.getName(), ctx.channel().id());
+            client.notifyConnected();
+        }
+
+        @Override
+        public void channelInactive(ChannelHandlerContext ctx){
+            if(client.channelHandlerContext == ctx) {
+                client.channelHandlerContext = null;
+            }
+            log.info("channel={} disconnected, ctx={}", client.channel.getName(), ctx.channel().id());
+            client.notifyDisconnected();
+        }
+
+        @Override
+        public void channelRead(ChannelHandlerContext ctx, Object msg) {
+            ByteBuf byteBuf = (ByteBuf) msg;
+            if (byteBuf.readableBytes() > 0) {
+                byte[] bytes = new byte[byteBuf.readableBytes()];
+                byteBuf.readBytes(bytes);
+                if(log.isDebugEnabled()) {
+                    log.debug("channel={}, receive bytes={}", client.channel.getName(), ByteBufUtil.hexDump(bytes));
+                }
+                client.notifyReceived(bytes);
+            }
+            ReferenceCountUtil.release(byteBuf);
+        }
+    }
+}