初版本通讯协议

gateway/forecast-modbus-rtu/src/main/java/cn/oleauto/modbus/rtu/MyRtuMaster.java

@@ -0,0 +1,340 @@
+package cn.oleauto.modbus.rtu;
+
+import cn.oleauto.common.core.text.Convert;
+import cn.oleauto.common.utils.StringUtils;
+import cn.oleauto.protocol.common.IChannel;
+import cn.oleauto.protocol.common.ITransmit;
+import cn.oleauto.protocol.common.event.DefaultEventArgs;
+import cn.oleauto.protocol.common.event.EventCenter;
+import cn.oleauto.protocol.common.event.EventType;
+import cn.oleauto.protocol.common.event.TransmitMessageEventArgs;
+import cn.oleauto.protocol.common.transmit.TransmitState;
+import com.serotonin.modbus4j.exception.ModbusInitException;
+import com.serotonin.modbus4j.exception.ModbusTransportException;
+import com.serotonin.modbus4j.msg.ModbusRequest;
+import com.serotonin.modbus4j.msg.ModbusResponse;
+import com.serotonin.modbus4j.serial.SerialMaster;
+import com.serotonin.modbus4j.serial.SerialPortWrapper;
+import com.serotonin.modbus4j.serial.SerialWaitingRoomKeyFactory;
+import com.serotonin.modbus4j.serial.rtu.RtuMessageParser;
+import com.serotonin.modbus4j.serial.rtu.RtuMessageRequest;
+import com.serotonin.modbus4j.serial.rtu.RtuMessageResponse;
+import com.serotonin.modbus4j.sero.NotImplementedException;
+import com.serotonin.modbus4j.sero.ShouldNeverHappenException;
+import com.serotonin.modbus4j.sero.messaging.MessageControl;
+import com.serotonin.modbus4j.sero.messaging.StreamTransport;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.util.Assert;
+
+import java.io.StringReader;
+import java.util.Properties;
+import java.util.function.Consumer;
+
+@Slf4j
+public class MyRtuMaster extends SerialMaster implements ITransmit {
+
+    // Runtime fields.
+    private final MessageControl conn;
+
+    public MyRtuMaster(IChannel channel){
+        super(null);
+        this.channel = channel;
+        setTimeout(10000);
+        this.conn = getMessageControl();
+    }
+
+    @Override
+    synchronized public void init() throws ModbusInitException {
+        try {
+            openConnection();
+        }
+        catch (Exception e) {
+            throw new ModbusInitException(e);
+        }
+        initialized = true;
+    }
+
+    private void openConnection() throws Exception {
+        super.openConnection(null);
+
+        RtuMessageParser rtuMessageParser = new RtuMessageParser(true);
+        this.conn.start(transport, rtuMessageParser, null, new SerialWaitingRoomKeyFactory());
+        if (getePoll() == null) {
+            ((StreamTransport) transport).start("Modbus RTU master");
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    synchronized public void destroy() {
+        closeMessageControl(conn);
+        super.close();
+        initialized = false;
+        setState(TransmitState.DISCONNECTED);
+    }
+
+    @Override
+    synchronized public ModbusResponse sendImpl(ModbusRequest request) throws ModbusTransportException {
+        // Wrap the modbus request in an rtu request.
+        RtuMessageRequest rtuRequest = new RtuMessageRequest(request);
+
+        // Send the request to get the response.
+        RtuMessageResponse rtuResponse;
+        try {
+            notifySent(rtuRequest.getMessageData());
+            rtuResponse = (RtuMessageResponse) conn.send(rtuRequest);
+            if (rtuResponse == null)
+                return null;
+            notifyReceived(rtuResponse.getMessageData());
+            return rtuResponse.getModbusResponse();
+        }
+        catch (Exception e) {
+            try {
+                log.debug("Connection may have been reset. Attempting to re-open.", e);
+                openConnection();
+                rtuResponse = (RtuMessageResponse) conn.send(rtuRequest);
+                if (rtuResponse == null)
+                    return null;
+                return rtuResponse.getModbusResponse();
+            }catch(Exception e2) {
+                closeConnection(conn);
+                log.debug("Failed to re-connect", e2);
+                throw new ModbusTransportException(e2, request.getSlaveId());
+            }
+        }
+    }
+
+    /**
+     * RTU Spec:
+     * For baud greater than 19200
+     * Message Spacing: 1.750uS
+     *
+     * For baud less than 19200
+     * Message Spacing: 3.5 * char time
+     *
+     * @param wrapper a {@link com.serotonin.modbus4j.serial.SerialPortWrapper} object.
+     * @return a long.
+     */
+    public static long computeMessageFrameSpacing(SerialPortWrapper wrapper){
+        //For Modbus Serial Spec, Message Framing rates at 19200 Baud are fixed
+        if (wrapper.getBaudRate() > 19200) {
+            return 1750000l; //Nanoseconds
+        }
+        else {
+            float charTime = computeCharacterTime(wrapper);
+            return (long) (charTime * 3.5f);
+        }
+    }
+
+    /**
+     * RTU Spec:
+     * For baud greater than 19200
+     * Char Spacing: 750uS
+     *
+     * For baud less than 19200
+     * Char Spacing: 1.5 * char time
+     *
+     * @param wrapper a {@link com.serotonin.modbus4j.serial.SerialPortWrapper} object.
+     * @return a long.
+     */
+    public static long computeCharacterSpacing(SerialPortWrapper wrapper){
+        //For Modbus Serial Spec, Message Framing rates at 19200 Baud are fixed
+        if (wrapper.getBaudRate() > 19200) {
+            return 750000l; //Nanoseconds
+        }
+        else {
+            float charTime = computeCharacterTime(wrapper);
+            return (long) (charTime * 1.5f);
+        }
+    }
+
+    /**
+     * Compute the time it takes to transmit 1 character with
+     * the provided Serial Parameters.
+     *
+     * RTU Spec:
+     * For baud greater than 19200
+     * Char Spacing: 750uS
+     * Message Spacing: 1.750uS
+     *
+     * For baud less than 19200
+     * Char Spacing: 1.5 * char time
+     * Message Spacing: 3.5 * char time
+     *
+     * @return time in nanoseconds
+     * @param wrapper a {@link com.serotonin.modbus4j.serial.SerialPortWrapper} object.
+     */
+    public static float computeCharacterTime(SerialPortWrapper wrapper){
+        //Compute the char size
+        float charBits = wrapper.getDataBits();
+        switch (wrapper.getStopBits()) {
+            case 1:
+                //Strangely this results in 0 stop bits.. in JSSC code
+                break;
+            case 2:
+                charBits += 2f;
+                break;
+            case 3:
+                //1.5 stop bits
+                charBits += 1.5f;
+                break;
+            default:
+                throw new ShouldNeverHappenException("Unknown stop bit size: " + wrapper.getStopBits());
+        }
+
+        if (wrapper.getParity() > 0)
+            charBits += 1; //Add another if using parity
+
+        //Compute ns it takes to send one char
+        // ((charSize/symbols per second) ) * ns per second
+        return (charBits / wrapper.getBaudRate()) * 1000000000f;
+    }
+
+    // 可以合并成一个吗？
+    private final EventCenter<DefaultEventArgs> centerConnected = new EventCenter<>();
+    private final EventCenter<DefaultEventArgs> centerDisonnected = new EventCenter<>();
+    private EventCenter<TransmitMessageEventArgs> centerReceived = new EventCenter<>();
+    private EventCenter<TransmitMessageEventArgs> centerSent = new EventCenter<>();
+    private final IChannel channel;
+    protected TransmitState state = TransmitState.STOPPED;
+    private volatile boolean isStop = false;
+
+
+    void setState(TransmitState state){
+        this.state = state;
+        if(state == TransmitState.CONNECTED){
+            notifyConnected();
+        } else {
+            notifyDisconnected();
+        }
+    }
+
+    @Override
+    public int start(String config){
+        Properties properties = new Properties();
+        try {
+            config = config.replace("\\n", "\n");
+            properties.load(new StringReader(config));
+            String comPort = properties.getProperty("port");
+            if(StringUtils.isBlank(comPort)){
+                throw new IllegalArgumentException("not exist port");
+            }
+            int baud = Convert.toInt(properties.getProperty("baud"), 9600);
+            if(baud == 0){
+                throw new IllegalArgumentException("invalid baud");
+            }
+            int dataBit = Convert.toInt(properties.getProperty("dataBit"), 8);
+            if(dataBit == 0){
+                throw new IllegalArgumentException("invalid dataBit");
+            }
+            int stopBit = Convert.toInt(properties.getProperty("stopBit"), 1);
+            if(dataBit == 0){
+                throw new IllegalArgumentException("invalid stopBit");
+            }
+            int parity = Convert.toInt(properties.getProperty("parity"), 0);
+            if(dataBit == 0){
+                throw new IllegalArgumentException("invalid parity");
+            }
+            ComSerialPortWrapper comSerialPortWrapper = new ComSerialPortWrapper(this,
+                    comPort, baud,0, 0, dataBit, stopBit, parity);
+            wrapper = comSerialPortWrapper;
+            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(){
+        isStop = true;
+    }
+
+    @Override
+    public int connect(){
+        if(isStop){
+            return -1;
+        }
+        try {
+            init();
+            return 0;
+        } catch (Exception exception){
+            setState(TransmitState.DISCONNECTED);
+            return -1;
+        }
+    };
+
+    @Override
+    public void disconnect() {
+        closeConnection(conn);
+        setState(TransmitState.DISCONNECTED);
+    };
+    @Override
+    public TransmitState state() {
+        return state;
+    }
+
+    @Override
+    public int send(byte[] data) {
+        throw new NotImplementedException();
+    }
+
+    @Override
+    public int send(byte[] data, int tmOutMs) {
+        throw new NotImplementedException();
+    }
+
+    @Override
+    public byte[] receive() {
+        throw new NotImplementedException();
+    }
+
+    @Override
+    public byte[] receive(int tmOutMs) {
+        throw new NotImplementedException();
+    }
+
+    @Override
+    public void subscribeConnected(Consumer<DefaultEventArgs> consumer){
+        Assert.notNull(consumer, "consumer can not be null.");
+        centerConnected.subscribe(consumer);
+    }
+
+    @Override
+    public void subscribeDisConnected(Consumer<DefaultEventArgs> consumer){
+        Assert.notNull(consumer, "consumer can not be null.");
+        centerDisonnected.subscribe(consumer);
+    }
+
+    @Override
+    public void subscribeSent(Consumer<TransmitMessageEventArgs> consumer){
+        Assert.notNull(consumer, "consumer can not be null.");
+        centerSent.subscribe(consumer);
+    }
+
+    @Override
+    public void subscribeReceived(Consumer<TransmitMessageEventArgs> consumer){
+        Assert.notNull(consumer, "consumer can not be null.");
+        centerReceived.subscribe(consumer);
+    }
+
+    protected void notifyConnected(){
+        state = TransmitState.CONNECTED;
+        centerConnected.publish(new DefaultEventArgs(this, EventType.TRANSMIT_CONNETED));
+    }
+
+    protected void notifyDisconnected(){
+        state = TransmitState.DISCONNECTED;
+        centerDisonnected.publish(new DefaultEventArgs(this, EventType.TRANSMIT_DISCONNECTED));
+    }
+
+    protected void notifySent(byte[] data){
+        centerSent.publish(new TransmitMessageEventArgs(this, EventType.TRANSMIT_SENT, data));
+    }
+
+    protected void notifyReceived(byte[] data){
+        centerReceived.publish(new TransmitMessageEventArgs(this, EventType.TRANSMIT_RECEIVED, data));
+    }
+}