WAREAMS/warewms-ams/src/main/java/com/ruoyi/ams/agv/ndc/AciService.java

package com.ruoyi.ams.agv.ndc;

import cn.hutool.core.date.DateUnit;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.thread.ThreadUtil;
import com.alibaba.fastjson.JSON;
import com.ruoyi.ams.agv.ndc.common.Aci;
import com.ruoyi.ams.agv.ndc.common.ByteUtil;
import com.ruoyi.ams.agv.ndc.domain.AmsHexdefineDetail;
import com.ruoyi.ams.agv.ndc.domain.AmsTask;
import com.ruoyi.ams.agv.ndc.entity.Order;
import com.ruoyi.ams.agv.ndc.service.IAmsHexdefineDetailService;
import com.ruoyi.ams.agv.ndc.service.IAmsTaskService;
import com.ruoyi.ams.agv.ndc.service.SocketBufferService;
import com.ruoyi.base.constant.Constant;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.Date;
import java.util.List;

/**
 * Created by IntelliJ IDEA.
 * User: andy.qu
 * Date: 2022/8/5
 */
@Slf4j
@Service
public class AciService {

    private static final int TIMEOUT = 3000;

    private static final int BUFF_SIZE = 2048;

    private static final String IP_ADDR = "10.95.2.66";

    private static final int IP_PORT = 50001;

    @Autowired
    private SocketBufferService socketBufferService;
    @Autowired
    private IAmsTaskService amsTaskService;
    @Autowired
    private IAmsHexdefineDetailService amsHexdefineDetailService;

    @SuppressWarnings("InfiniteLoopStatement")
    public void start() throws IOException {

        String ip_addr = IP_ADDR; //redisUtils.get(RedisEnum.NDC_IP.getMessage());
        int ip_port = IP_PORT; //redisUtils.getInt(RedisEnum.NDC_PORT.getMessage());

        SocketChannel socketChannel = SocketChannel.open();
        socketChannel.configureBlocking(false); // 解决了IO阻塞
        while (true) {

            try {
                //链接服务端Socket
                if (!socketChannel.isConnected()) {
                    if (!socketChannel.isOpen()) {
                        socketChannel = SocketChannel.open();
                        socketChannel.configureBlocking(false); // 解决了IO阻塞
                    }
                    // 非阻塞模式下 这段代码并不是等到连接建立之后再往下执行
                    SocketAddress socketAddress = new InetSocketAddress(ip_addr, ip_port);
                    socketChannel.connect(socketAddress);
                    Thread.sleep(2000);
                    if (socketChannel.isConnectionPending()) {
                        socketChannel.finishConnect();
                    }
                }

                ByteBuffer buffer = ByteBuffer.allocate(BUFF_SIZE);
                buffer.clear();

                //删除指令
                AmsTask taskBean = new AmsTask();
                taskBean.setIsDelete(1);
                taskBean.setAciAccept(0);
                taskBean = amsTaskService.selectAmsTaskByModel(taskBean);
                String s = "87CD000800060001006E0002%04X";
                if (taskBean != null) {
                    if (taskBean.getAciIndex() == null) {
                        //还没有提交给aci系统
                        taskBean.setAciAccept(1);
                        amsTaskService.updateAmsTask(taskBean);
                    } else {
                        //已经提交给aci系统
                        s = String.format(s, taskBean.getAciIndex());
                        buffer.put(ByteUtil.hexString2ByteArray(s));
                        buffer.flip();
                        socketChannel.write(buffer);
                        buffer.clear();
                        log.info("cancelOrder: " + s);
                    }

                    if (DateUtil.between(taskBean.getAddtime(), new Date(), DateUnit.MINUTE) > 1) {
                        taskBean.setAciAccept(2);
                        taskBean.setRemark("下发超时1分钟未接收自动取消");
                        amsTaskService.updateAmsTask(taskBean);
                    }
                }

                //充电指令
                taskBean = new AmsTask();
                taskBean.setIsDelete(0);
                taskBean.setAciAccept(0);
                taskBean.setBusinessType(Constant.TASK_BUSINESS_TYPE.TASK_POWER.getValue());
                taskBean = amsTaskService.selectAmsTaskByModel(taskBean);
                s = "87CD0008000C00010071000663010001%04X";
                if (taskBean != null) {

                    s = String.format(s, Integer.parseInt(taskBean.getDeviceName()));
                    buffer.put(ByteUtil.hexString2ByteArray(s));
                    buffer.flip();
                    socketChannel.write(buffer);
                    buffer.clear();
                    log.info("AGV-Power: " + s);
                    taskBean.setServerAck(0);
                    taskBean.setAciAccept(1);
                    taskBean.setIsDelete(null);
                    amsTaskService.updateAmsTask(taskBean);
                }

                //添加指令
                List<AmsTask> taskBeans = amsTaskService.selectTaskNotIssued();

                if (taskBeans != null && taskBeans.size() > 0) {

                    for (AmsTask t : taskBeans) {
                        t.init();
                        List<AmsHexdefineDetail> basTask = amsHexdefineDetailService.selectAmsHexdefineDetailByBusinessType(t.getBusinessType());
                        if (basTask == null || basTask.size() == 0) {
                            t.setIsDelete(1);
                            t.setRemark("该任务类型未定义");
                            log.info("任务类型未定义------" + JSON.toJSONString(t));
                            continue;
                        }
//                        try {
                        String order = amsHexdefineDetailService.concatOrder(t, basTask);
                        log.info("writeOrder:" + order);
                        if (t.getBusinessType().equals("01")) {

                            buffer.put(ByteUtil.hexString2ByteArray(order));
                            buffer.flip();
                            socketChannel.write(buffer);
                            buffer.clear();
                            if (DateUtil.between(t.getAddtime(), new Date(), DateUnit.MINUTE) > 2) {
                                t.setAciAccept(2);
                                t.setRoadway("下发超时2分钟未接收自动取消");
                                amsTaskService.updateAmsTask(t);
                            }
                        } else {
                            buffer.put(ByteUtil.hexString2ByteArray(order));
                            buffer.flip();
                            socketChannel.write(buffer);
                            buffer.clear();
                            // 修改已发送
                            t.setAciAccept(1);
                            t.setRoadway("TS请求反馈任务发送成功！");
                            amsTaskService.updateAmsTask(t);
                        }
//                        } catch (Exception e) {
//                            e.printStackTrace();
//                            t.setServerAck(0);
//                            t.setAciAccept(1);
//                            t.setRoadway("程序异常自动ack");
//                            t.setIsDelete(null);
//                            amsTaskService.updateAmsTask(t);
//                        }
                    }
                }

                //从服务端读取
                buffer.clear();
                //因为设置了异步 这边不会阻塞 测试了一下 服务器断开这边会抛出异常 但是写的时候不会抛出异常 可能是设置了异步的原因
                int readLenth = socketChannel.read(buffer);
                if (readLenth == -1) {
//                    throw new IOException();
                }
                if (readLenth > 0) {
                    buffer.flip();
                    byte[] bytes = new byte[readLenth];
                    buffer.get(bytes);
                    List<Order> bufferSeparators = socketBufferService.bufferSeparator(bytes);
                    for (Order order : bufferSeparators) {

//                    log.info("消息类型：" + order.getMessageType() + "，消息参数：" + ByteUtil.byteArray2HexString(order.getParameters()));
                        switch (order.getMessageType()) {

                            case Aci.MSG_ORDER_STATUS://车体状态信息、位置信息
                                socketBufferService.handleMsgOrderStatus(order.getParameters());
                                break;
                            case Aci.MSG_EVENT://任务反馈、自定义消息
                                socketBufferService.handleMsgEvent(order.getParameters());
                                break;
                            case Aci.MSG_STOP://交通管制暂停
                                socketBufferService.handleMsgStopStart(order.getParameters(), true);
                                break;
                            case Aci.MSG_START://启动信息
                                socketBufferService.handleMsgStopStart(order.getParameters(), false);
                                break;
                            case Aci.MSG_ORDER_ACK://信息接收反馈
                                socketBufferService.handleMsgOrderAck(order.getParameters());
                                break;
                        }
                    }
                }
                buffer.clear();

                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            } catch (IOException i) {
                log.error("NDC通讯异常！", i);
                try {
                    socketChannel.close();
                } catch (Exception exception) {
                    i.printStackTrace();
                }
            } catch (Exception e) {
                e.printStackTrace();
                try {
                    socketChannel.close();
                } catch (Exception exception) {
                    e.printStackTrace();
                }
            }
        }
    }
}