package com.cmvr.edge.client.service.impl;

import com.cmvr.edge.client.manage.GrpcStreamManager;
import com.cmvr.edge.client.service.EdgeStreamService;
import com.cmvr.framework.websocket.service.GrpcClientService;
import io.grpc.Status;
import io.grpc.stub.StreamObserver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.lang.reflect.Method;
import java.util.Map;

/**
 * grpc客户端服务实现类
 * 统一调度grpc流式接口
 */
@Service
public class GrpcClientServiceImpl implements GrpcClientService {

    private static final Logger log = LoggerFactory.getLogger(GrpcClientServiceImpl.class);

    /**
     * 所有流式服务
     */
    private final Map<String, EdgeStreamService> serviceMap;

    @Autowired
    public GrpcClientServiceImpl(Map<String, EdgeStreamService> serviceMap) {
        this.serviceMap = serviceMap;
    }

    @Override
    public void send(boolean isStart, String terminalId, String deviceId, String service, String method) {
        try {
            // 从 Map 中获取服务实例
            EdgeStreamService serviceInstance = serviceMap.get(service);
            if (serviceInstance == null) {
                throw new IllegalArgumentException("未知的service: " + service);
            }

            // 构造流键
            String key = terminalId + deviceId + service + method;

            if (isStart) {
                // 通过反射调用指定方法
                Method streamMethod = serviceInstance.getClass().getMethod(method, String.class, String.class);
                StreamObserver<?> streamObserver = (StreamObserver<?>) streamMethod.invoke(serviceInstance, terminalId, deviceId);
                if (streamObserver != null) {
                    GrpcStreamManager.addStream(key, streamObserver);
                }
            } else {
                if (!GrpcStreamManager.containsKey(key)) {
                    return;
                }
                // 移除并关闭流
                StreamObserver<?> streamObserver = GrpcStreamManager.removeStream(key);
                try {
                    streamObserver.onError(Status.CANCELLED.withDescription("流需要关闭").asException());
                    log.info("流关闭: {}", key);
                } catch (Exception e) {
                    log.info("流结束");
                }
            }
        } catch (Exception e) {
            log.error("调用服务失败{}", e.getMessage(), e);
        }
    }
}
