package com.mes.milo.runner.subscription;

import com.mes.milo.utils.CustomUtil;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
import org.eclipse.milo.opcua.sdk.client.api.subscriptions.UaSubscription;
import org.eclipse.milo.opcua.sdk.client.api.subscriptions.UaSubscriptionManager;
import org.eclipse.milo.opcua.sdk.client.subscriptions.ManagedDataItem;
import org.eclipse.milo.opcua.sdk.client.subscriptions.ManagedSubscription;
import org.eclipse.milo.opcua.stack.core.types.builtin.NodeId;
import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode;
import org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.UInteger;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * 类 SubscriptionRunner 功能描述：
 *
 * @author mes
 * @version 0.0.1
 * @date 2022/01/01 23:49
 */
@Slf4j
public class SubscriptionRunner {
    /**
     * 点位list
     */
    private final List<String> identifiers;

    private final double samplingInterval;

    public SubscriptionRunner(List<String> identifiers) {
        this.identifiers = identifiers;
        this.samplingInterval = 1000.0D;
    }

    public SubscriptionRunner(List<String> identifiers, double samplingInterval) {
        this.identifiers = identifiers;
        this.samplingInterval = samplingInterval;
    }

    public void run(OpcUaClient opcUaClient, SubscriptionCallback callback) {

        final CountDownLatch downLatch = new CountDownLatch(1);

        //添加订阅监听器，用于处理断线重连后的订阅问题
        opcUaClient.getSubscriptionManager().addSubscriptionListener(new CustomSubscriptionListener(opcUaClient, callback));

        //处理订阅逻辑
        handler(opcUaClient, callback);

        try {
            //持续监听
            downLatch.await();
        } catch (Exception e) {
            log.error("订阅时出现了异常：{}", e.getMessage(), e);
        }
    }

    private void handler(OpcUaClient opcUaClient, SubscriptionCallback callback) {
        try {
            //创建订阅
            ManagedSubscription subscription = ManagedSubscription.create(opcUaClient, samplingInterval);
            subscription.setDefaultSamplingInterval(samplingInterval);
            subscription.setDefaultQueueSize(UInteger.valueOf(10));

            List<NodeId> nodeIdList = new ArrayList<>();
            for (String identifier : identifiers) {
                nodeIdList.add(CustomUtil.parseNodeId(identifier));
            }
            List<ManagedDataItem> dataItemList = subscription.createDataItems(nodeIdList);
            for (ManagedDataItem dataItem : dataItemList) {
                dataItem.addDataValueListener((item) -> {
                    try {
                        callback.onSubscribe
                                (dataItem, item);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                });
            }
        } catch (Exception e) {
            log.error("订阅时出现了异常：{}", e.getMessage(), e);
        }
    }

    private class CustomSubscriptionListener implements UaSubscriptionManager.SubscriptionListener {
        private final OpcUaClient client;
        private final SubscriptionCallback callback;

        public CustomSubscriptionListener(OpcUaClient client, SubscriptionCallback callback) {
            this.client = client;
            this.callback = callback;
        }

        /**
         * 重连时 尝试恢复之前的订阅失败时 会调用此方法
         *
         * @param uaSubscription 订阅
         * @param statusCode     状态
         */
        @Override
        public void onSubscriptionTransferFailed(UaSubscription uaSubscription, StatusCode statusCode) {
            log.debug("恢复订阅失败 需要重新订阅");
            //在回调方法中重新订阅
            handler(client, callback);
        }
    }
}