warewms-project-api/warewms-system/src/main/java/com/warewms/hailiang/connect/CodeReader8Connect.java

package com.warewms.hailiang.connect;

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.extra.spring.SpringUtil;
import com.github.rholder.retry.*;
import com.warewms.common.core.redis.RedisCache;
import com.warewms.hailiang.MES.MesService;
import com.warewms.hailiang.config.CodeReadProperties;
import com.warewms.hailiang.connect.base.TCPConnectBase;
import com.warewms.hailiang.domain.Device;
import com.warewms.hailiang.domain.DeviceLog;
import com.warewms.hailiang.domain.RetroactiveNow;
import com.warewms.hailiang.enums.DaSanPanPlcEnum;
import com.warewms.hailiang.enums.DeviceNameEnum;
import com.warewms.hailiang.init.PlcConnectServiceRunner;
import com.warewms.hailiang.service.RetroactiveNowService;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.CharsetUtil;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
/**
 * Created with IntelliJ IDEA.
 *
 * @author: liuzhifei
 * Date: 2023/8/9
 * Time: 15:45
 * To change this template use File | Settings | File Templates.
 * Description:大散盘行车2号读码器
 **/
@Slf4j
public class CodeReader8Connect implements TCPConnectBase {

    private final String IP_ADDR = "172.20.27.8";

    private final int PORT = 51236;

    private final String deviceName ="CodeReader8";

    private boolean enable = true;

    private ChannelFuture future;

    private Bootstrap bootstrap;

    private EventLoopGroup group;

    private ChannelPipeline pipeline;

    private MesService mesService;

    private RetroactiveNowService retroactiveNowService;

    private PlcConnectServiceRunner plcConnectServiceRunner;

    private ScheduledFuture<?> scheduledFuture =null;

    private ScheduledExecutorService scheduledExecutorService;


    {
        mesService = SpringUtil.getBean(MesService.class);
        retroactiveNowService = SpringUtil.getBean(RetroactiveNowService.class);
        plcConnectServiceRunner = SpringUtil.getBean(PlcConnectServiceRunner.class);
        scheduledExecutorService = SpringUtil.getBean(ScheduledExecutorService.class);
    }

    @Override
    public void init() throws InterruptedException {
        if (enable) {
            log.info("ip:{},deviceName:{}正在进行连接", IP_ADDR, deviceName);
            group = new NioEventLoopGroup();
            try {
                bootstrap = new Bootstrap();
                bootstrap.group(group).channel(NioSocketChannel.class)
                        .option(ChannelOption.TCP_NODELAY, true)
                        .handler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception {
                                pipeline = socketChannel.pipeline();
                                pipeline.addLast(new IdleStateHandler(1,0,0,TimeUnit.SECONDS));
                                pipeline.addLast(new ChannelInboundHandlerAdapter(){
                                    @Override
                                    public void channelInactive(ChannelHandlerContext ctx) {
                                        SpringUtil.getApplicationContext().publishEvent(new Device(deviceName, "2"));
                                        log.error("设备：{}连接已断开！",deviceName);
                                        scheduledFuture.cancel(true);
                                        retry();
                                    }
                                    @Override
                                    public void channelRead(ChannelHandlerContext ctx, Object msg) {
                                        ByteBuf byteBuf = (ByteBuf) msg;
                                        processMessages(String.valueOf(byteBuf.toString(CharsetUtil.UTF_8)));
                                    }
                                });
                            }
                        });
                future = bootstrap.connect(IP_ADDR, PORT).sync();
                future.addListener((channelFuture) -> {
                    if (channelFuture.isSuccess()) {
                        SpringUtil.getApplicationContext().publishEvent(new Device(deviceName, "1"));
                        log.info("ip:{},deviceName:{}连接成功", IP_ADDR, deviceName);
                        ConnectsTheHeartbeat();
                    } else {
                        log.info("ip:{},deviceName:{}连接失败等待重试！", IP_ADDR, deviceName);
                        retry();
                    }
                });
            } catch (Exception e) {
                log.error("ip:{},deviceName:{}连接异常,{},准备重试", IP_ADDR, deviceName, e.getMessage());
                retry();
            }
        }
    }

    @Override
    public void retry() {
        Retryer<Boolean> retryer = RetryerBuilder.<Boolean>newBuilder()
                .retryIfResult(Boolean.FALSE::equals)
                .retryIfExceptionOfType(Exception.class)
                .withStopStrategy(StopStrategies.neverStop())
                .withWaitStrategy(WaitStrategies.fixedWait(5, TimeUnit.SECONDS))
                .build();
        try {
            retryer.call(new Callable<Boolean>() {
                AtomicBoolean isSuccess = new AtomicBoolean(false);
                @Override
                public Boolean call() throws Exception {
                    log.info("ip:{},deviceName:{}连接重试",IP_ADDR,deviceName);
                    future = bootstrap.connect(IP_ADDR, PORT).sync();
                    future.addListener((channelFuture) -> {
                        isSuccess.set(channelFuture.isSuccess());
                        if (channelFuture.isSuccess()) {
                            SpringUtil.getApplicationContext().publishEvent(new Device(deviceName,"1"));
                            log.info("ip:{},deviceName:{}连接成功",IP_ADDR,deviceName);
                            ConnectsTheHeartbeat();
                        }else {
                            log.info("ip:{},deviceName:{}连接失败等待重试！",IP_ADDR,deviceName);
                        }
                    });
                    Thread.sleep(3000);
                    return isSuccess.get();
                }
            });
        } catch (RetryException | ExecutionException e) {
            e.printStackTrace();
        }
    }
    /**
     * 连接心跳
     *
     */
    private void ConnectsTheHeartbeat() {
        scheduledFuture = scheduledExecutorService.scheduleWithFixedDelay(() -> {
            try {
//                log.info("发送心跳，设备：{}", deviceName);
                pipeline.writeAndFlush(Unpooled.copiedBuffer(CodeReadProperties.heartbeat, CharsetUtil.UTF_8));
            } catch (Exception e) {
                log.error("设备：{},连接中断", deviceName);
                SpringUtil.getApplicationContext().publishEvent(new Device(deviceName, "2"));
                retry();
            }
        }, 0, 3, TimeUnit.SECONDS);
    }

    @Override
    public void close() {
        group.shutdownGracefully();
    }


    @Override
    public String getDeviceName() {
        return deviceName;
    }

    @Override
    public ChannelPipeline getChannel() {
        return pipeline;
    }

    @Override
    public void processMessages(String message) {
        try {
            log.info("大散盘2号行车读码器消息：{}", message);
            if (CodeReadProperties.failureReturnInstruction.equals(message)) {
                SpringUtil.getApplicationContext().publishEvent(new DeviceLog("Z1_DaSanPan_DMQ-2-27.8", deviceName, "未识别到码", "2"));
                plcConnectServiceRunner.getPlcServer(DeviceNameEnum.DASANPANPLC.getDeviceName()).writeByte(DaSanPanPlcEnum.ISREAD_two.getMetadata(), (byte) 2);

                Thread.sleep(10000L);
                plcConnectServiceRunner.getPlcServer(DeviceNameEnum.DASANPANPLC.getDeviceName()).writeByte(DaSanPanPlcEnum.ISREAD_two.getMetadata(),(byte) 0);

            } else {
                //获取出料位
                float v = plcConnectServiceRunner.getPlcServer(DeviceNameEnum.DASANPANPLC.getDeviceName()).readFloat32(DaSanPanPlcEnum.TASK_ADD_1.getMetadata());
                String deviceId = v == 10 ? "Z1SZ10" : "Z1SP0" + v;
                String batchNo = mesService.bindLotNo(deviceId, message);

                //读取到托盘号保存到数据库中
                RetroactiveNow retroactiveNow = new RetroactiveNow();
                retroactiveNow.setBatchNo(batchNo);
                retroactiveNow.setLotNo(message);
                retroactiveNow.setStatus("5");
                retroactiveNow.setDeviceId("Z1_DaSanPan_DMQ-1-27.8");
                retroactiveNow.setSanPanDevice(deviceId);
                retroactiveNowService.DataTwins(retroactiveNow);
                plcConnectServiceRunner.getPlcServer(DeviceNameEnum.DASANPANPLC.getDeviceName()).writeByte(DaSanPanPlcEnum.ISREAD_two.getMetadata(), (byte) 0);
                //添加设备日志
                SpringUtil.getApplicationContext().publishEvent(new DeviceLog("Z1_DaSanPan_DMQ-2-27.8", deviceName, "读码器识别到码:" + message, "1"));
            }
        }catch (Exception e){
            log.error("大散盘2号行车读码任务执行异常，msg:",e);
            SpringUtil.getApplicationContext().publishEvent(new DeviceLog("Z1_DaSanPan_DMQ-2-27.8", deviceName, "读码任务出错，msg:"+e.getMessage(), "2"));
        }
    }
}