package com.wecloud.im.server.sender;

import com.wecloud.im.core.common.api.ApiCode;
import com.wecloud.im.core.common.api.ApiResult;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.cluster.router.address.Address;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.client.discovery.DiscoveryClient;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSON;
import com.wecloud.im.executor.SendMsgThreadPool;
import com.wecloud.im.router.RouterSendService;
import com.wecloud.im.server.cache.UserStateCacheManager;
import com.wecloud.im.server.manager.ChannelManager;
import com.wecloud.im.server.model.ClientInfo;
import com.wecloud.im.server.model.WsResponse;
import com.wecloud.im.server.model.redis.ClientChannelInfo;
import com.wecloud.im.server.model.request.ReceiveVO;
import com.wecloud.utils.GetIpUtils;
import com.wecloud.utils.JsonUtils;

/**
 * @Description 下发数据
 * @Author hewei hwei1233@163.com
 * @Date 2019-12-05
 */
@Component
@Slf4j
public class ChannelSender {

    @Autowired
    private UserStateCacheManager userStateCacheManager;

    @Autowired
    private DiscoveryClient discoveryClient;

    /*
     * 指定ip调用,router=address;
     * injvm = false要设置成false,否则会调用到本地提供者
     */
    @DubboReference(injvm = false, check = false, interfaceClass = RouterSendService.class, parameters = {"router", "address"})
    private RouterSendService routerSendService;

    /**
     * 固定"成功"状态码 带data
     *
     * @param receiveVO
     * @param data
     * @param toClientId
     */
    public void sendMsgData(ReceiveVO receiveVO, Object data, Long toClientId) {
        this.sendMsg(receiveVO, ApiCode.SUCCESS, data, toClientId);
    }

    /**
     * 固定"成功"状态码 无data
     *
     * @param receiveVO
     * @param apiCode
     * @param toClientId
     */
    public void sendMsgSucess(ReceiveVO receiveVO, ApiCode apiCode, Long toClientId) {
        this.sendMsg(receiveVO, apiCode, new HashMap<>(1), toClientId);
    }

    /**
     * 固定"参数错误"状态码 无data
     *
     * @param receiveVO
     * @param toClientId
     */
    public void sendMsgIllegeArgs(ReceiveVO receiveVO, Long toClientId) {
//        this.nullDataSuccess(requestModel, ResultStatus.PARAM_ERROR, userId);
    }


    /**
     * 可自定义状态码 带data
     *
     * @param receiveVO
     * @param data
     */
    public void sendMsg(ReceiveVO receiveVO, ApiCode apiCode, Object data, Long toClientId) {
        ApiResult<Boolean> apiResult = ApiResult.result(apiCode);
        WsResponse responseModel = new WsResponse();
        responseModel.setMsg(apiResult.getMessage());
        responseModel.setCmd(receiveVO.getCmd());
        responseModel.setReqId(receiveVO.getReqId());
        responseModel.setData(data);
        responseModel.setCode(apiResult.getCode());
        this.sendMsg(responseModel, toClientId);
    }

    /**
     * 批量发消息
     * @param responseModel
     * @param toIp
     * @param toClientIdAndPlatforms
     */
    public void batchSendMsg(WsResponse responseModel, String toIp, List<String> toClientIdAndPlatforms) {
        // 是否为当前机器的ip
        if (GetIpUtils.getlanIp().equals(toIp)) {
            String msgJson = JsonUtils.encodeJson(responseModel);
            batchSendMsgLocal(toClientIdAndPlatforms, msgJson);
        } else {
            String msgJson = JsonUtils.encodeJson(responseModel);

            // dubbo指定ip调用
            Address address = new Address(toIp, 20881);
            RpcContext.getContext().setObjectAttachment("address", address);
            routerSendService.batchSendMsgRemote(toClientIdAndPlatforms, msgJson);
        }
    }

    /**
     * 调用ws处理响应逻辑
     *
     * @param responseModel
     * @param toClientId
     */
    public void sendMsg(WsResponse responseModel, Long toClientId) {

        String msgJson = JsonUtils.encodeJson(responseModel);

        List<ClientChannelInfo> channelInfos = userStateCacheManager.findOnlineInfosByClientId(toClientId);
        log.info("获取在线用户入参 {}, 结果 {}", toClientId, JSON.toJSONString(channelInfos));

        // 一个用户存在多端的情况，所以先进行分类，key是ip地址，value是channel的列表
        Map<String, List<ClientChannelInfo>> ipChannels = channelInfos.stream().collect(Collectors.groupingBy(ClientChannelInfo::getLanIp));

        log.info("分组在线用户入参 {}, 结果 {}", toClientId, JSON.toJSONString(ipChannels));

        for (Map.Entry<String, List<ClientChannelInfo>> channelInfoEntry : ipChannels.entrySet()) {

            log.info("在线用户入参 {}, 具体ip结果 {}", toClientId, JSON.toJSONString(channelInfoEntry));

            // 是否为当前机器的ip
            if (GetIpUtils.getlanIp().equals(channelInfoEntry.getKey())) {
                // 调用本地下发
                log.info("在线用户入参 {}, 具体ip结果 {}", toClientId, JSON.toJSONString(channelInfoEntry));
                for(ClientChannelInfo clientChannelInfo : channelInfoEntry.getValue()) {
                    log.info("客户端 {}, 推送消息内容 {}", toClientId, msgJson);
                    this.sendMsgLocal(toClientId, clientChannelInfo.getPlatform(), msgJson);
                }
                continue;
            }
            // todo rpc调用下发，需要改成批量
//                RpcContext.getContext().set("ip", channelInfo.getLanIp());
            // 根据provider的ip,port创建Address实例
            for(ClientChannelInfo clientChannelInfo : channelInfoEntry.getValue()) {
                // dubbo指定ip调用
                Address address = new Address(clientChannelInfo.getLanIp(), 20882);
                RpcContext.getContext().setObjectAttachment("address", address);
                log.info("dubbo指定ip调用 {}", JSON.toJSONString(address));
                try {
                    routerSendService.sendMsgRemote(toClientId, clientChannelInfo.getPlatform(), msgJson);
                } catch (RpcException exception) {
                    // do nothing is ok
                    tryFindAndSendAndRefresh(toClientId, clientChannelInfo.getPlatform(), msgJson);
                    log.info("rpc 调用异常 {} ", JSON.toJSONString(clientChannelInfo));
                }
            }
        }

    }

    /**
     *
     * @Author luozh
     * @Date 2022年05月24日 05:30:50
     * @param toClientId
     * @param platform
     * @param msg
     * @Return
     */
    public void tryFindAndSendAndRefresh(Long toClientId, Integer platform, String msg) {
        log.info("触发tryFindAndSendAndRefresh {} {}", toClientId, platform);
        this.sendMsgLocal(toClientId, platform, msg);
    }

    /**
     * 本地直接下发，限定本机有的channel
     *
     * @param nioSocketChannel
     * @param responseModel
     */
    public void sendMsgLocal(NioSocketChannel nioSocketChannel, Object responseModel) {

        String msgJson = JsonUtils.encodeJson(responseModel);

        // 本地直接下发
        nioSocketChannel.writeAndFlush(new TextWebSocketFrame(msgJson));
    }

    /**
     * 本地直接下发，限定本机有的channel
     *
     * @param nioSocketChannel
     * @param responseModel
     */
    public void sendMsgLocal(NioSocketChannel nioSocketChannel, WsResponse responseModel) {

        String msgJson = JsonUtils.encodeJson(responseModel);

        // 本地直接下发
        nioSocketChannel.writeAndFlush(new TextWebSocketFrame(msgJson));
    }

    /**
     * 批量发送
     * @param toClientIdAndPlatforms list里的 结构为  client的主键id:platform
     * @param msgJson
     */
    public void batchSendMsgLocal(List<String> toClientIdAndPlatforms, String msgJson) {
        for(String arrStr : toClientIdAndPlatforms) {
            SendMsgThreadPool.SEND_MSG_THREAD_POOL_EXECUTOR.execute(()->{
                this.sendMsgLocal(arrStr, msgJson);
            });
        }
    }

    /**
     * 向指定channelId下发数据，限定本机有的channel
     *
     * @param clientId
     * @param platform
     * @param msg
     * @return
     */
    public boolean sendMsgLocal(Long clientId, Integer platform, String msg) {
        String key = ChannelManager.genKeyForSessionInfoMap(clientId, platform);
        return sendMsgLocal(key, msg);
    }

    /**
     * 本地的channel 推送
     * @param key 结构为  client的主键id:platform
     * @param msgJsonStr json结构
     * @return
     */
    private boolean sendMsgLocal(String key, String msgJsonStr) {
        log.info("消息推送 key {} 内容 {}", key, msgJsonStr);
        ClientInfo clientInfo = ChannelManager.SESSION_INFO_MAP.get(key);
        if (clientInfo == null) {
            log.info("消息推送 key {} 本地map取设备信息为空", key);
            return false;
        }

        NioSocketChannel nioSocketChannel = clientInfo.getNioSocketChannel();
        if (null == nioSocketChannel) {
            log.info("writeData连接为空: {}", msgJsonStr);
            return false;
        }
        // 判断连接是否断开
        if (nioSocketChannel.isShutdown()) {
            log.info("writeData 已连接断开: {}\n channelId: {}", msgJsonStr, nioSocketChannel.id().asLongText());
            return false;
        }

        log.info("writeData, channelId: {}", nioSocketChannel.id().asLongText());

        nioSocketChannel.writeAndFlush(new TextWebSocketFrame(msgJsonStr));

        return true;
    }

}
