package com.fastbee.base.session;

import com.fastbee.common.core.protocol.Message;
import com.fastbee.common.enums.ServerType;
import com.fastbee.common.exception.ServiceException;
import com.fastbee.common.utils.DateUtils;
import com.fastbee.common.utils.gateway.mq.Topics;
import com.fastbee.base.core.model.Response;
import com.fasterxml.jackson.annotation.JsonFormat;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttVersion;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoSink;

import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;

/**
 * 会话
 *
 * @Author guanshubiao
 * @Date 2022/9/12 20:22
 */
@Data
@Slf4j
public class Session {

    private  boolean udp;
    private  Function<Session, Boolean> remover;
    protected Channel channel;
    /**
     * 客户端id
     */
    private String clientId;
    private String productId;
    private SessionManager sessionManager;
    private InetSocketAddress remoteAddress;
    private String remoteAddressStr;

    private long creationTime;
    private long lastAccessTime;
    private Map<Object, Object> attributes;

    private String sessionId;
    //原子计数器，报文没有消息流水号的，充当流水号
    private AtomicInteger serialNo = new AtomicInteger(0);
    private int keepAlive = 60;


    /*mqtt版本号*/
    private MqttVersion version;
    /*是否清楚会话*/
    private Boolean cleanSession = false;
    /*mqtt账号*/
    private String username;
    /*是否链接*/
    private Boolean connected = false;
    /*mqtt消息类型*/
    private MqttMessageType mqttMessageType;
    private int keepAliveMax = 120;
    /*主题*/
    private String topicName;
    /*Channel处理类上下文*/
    private ChannelHandlerContext handlerContext;

    private List<Topics> topics;
    private Integer topicCount;
    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
    private Date connected_at;
    private String ip;
    /*服务协议类型 MQTT，TCP UDP COAP*/
    private ServerType serverType;

    public Session(){

    }

    public Session(SessionManager sessionManager,
                   Channel channel,
                   InetSocketAddress remoteAddress,
                   Function<Session, Boolean> remover,
                   boolean udp,
                   ServerType serverType) {
        this.channel = channel;
        this.creationTime = DateUtils.getTimestamp();
        this.lastAccessTime = creationTime;
        this.sessionManager = sessionManager;
        this.remoteAddress = remoteAddress;
        this.remoteAddressStr = remoteAddress.toString();
        this.remover = remover;
        this.udp = udp;
        this.serverType = serverType;
        this.ip = remoteAddress.getHostName();
        this.connected = true;
        this.connected_at = DateUtils.getNowDate();
        if (sessionManager != null && sessionManager.getSessionKeys() != null) {
            this.attributes = new EnumMap(sessionManager.getSessionKeys());
        } else {
            this.attributes = new TreeMap<>();
        }
    }

    /**
     * 判断设备是否已经注册上线
     */
    public boolean isRegistered() {
        return clientId != null;
    }

    /*设备端注册*/
    public void register(Message message) {
        register(message.getClientId(), message);
    }

    public void register(String clientId, Message message) {
        //已经注册，不再发送注册包数据
        if (isRegistered()){
            return;
        }
        if (clientId == null) {
            throw new ServiceException("客户端注册clientId不能为空");
        }
        this.clientId = clientId.toUpperCase();
        if (sessionManager != null) {
            sessionManager.add(this);
        }
    }

    public Object getAttribute(Object name) {
        return attributes.get(name);
    }

    public void setAttribute(Object name, Object value) {
        attributes.put(name, value);
    }

    /**
     * 获取最后上线时间
     */
    public long access() {
        return lastAccessTime = DateUtils.getTimestamp();
    }

    /**
     * 获取远程端口
     */
    public InetSocketAddress remoteAddress() {
        return remoteAddress;
    }

    /**
     * 获取流水号
     */
    public int nextSerialNO() {
        int current;
        int next;
        do {
            current = serialNo.get();
            next = current > 0xffff ? 0 : current;
        } while (!serialNo.compareAndSet(current, next + 1));
        return next;
    }

    /**
     * 处理离线方法
     */
    public void invalidate() {
        if (isRegistered() && sessionManager != null) {
            sessionManager.remove(this);
        }
        remover.apply(this);
    }

    public boolean isUdp() {
        return udp;
    }

    private final Map<String, MonoSink> topicSubscribers = new HashMap<>();

    private static final Mono rejected = Mono.error(new RejectedExecutionException("设备端暂无响应"));

    /**
     * 异步发送通知类消息
     * 同步发送 mono.block()
     * 订阅回调 mono.doOnSuccess(处理成功).doOnError(处理异常).subscribe(开始订阅)
     */
    public Mono<Void> notify(Message message) {
        return mono(channel.writeAndFlush(Packet.of(this, message)));
    }

    public Mono<Void> notify(ByteBuf message) {
        return mono(channel.writeAndFlush(Packet.of(this, message)));
    }

    public static Mono<Void> mono(ChannelFuture channelFuture) {
        return Mono.create(sink -> channelFuture.addListener(future -> {
            if (future.isSuccess()) {
                sink.success();
            } else {
                sink.error(future.cause());
            }
        }));
    }

    /**
     * 异步发送消息，接收响应
     * 同步接收 mono.block()
     * 订阅回调 mono.doOnSuccess(处理成功).doOnError(处理异常).subscribe(开始订阅)
     */
    public <T> Mono<T> request(Message message, Class<T> responseClass) {
        String key = requestKey(message, responseClass);
        Mono<T> subscribe = this.subscribe(key);
        if (subscribe == null) {
            return rejected;
        }
        ChannelFuture channelFuture = channel.writeAndFlush(Packet.of(this, message));
        return Mono.create(sink -> channelFuture.addListener(future -> {
            if (future.isSuccess()) {
                sink.success(future);
            } else {
                sink.error(future.cause());
            }
        })).then(subscribe).doFinally(signal -> unsubscribe(key));
    }

    /**
     * 消息响应
     */
    public boolean response(Message message){
        MonoSink monoSink = topicSubscribers.get(responseKey(message));
        if (monoSink != null){
            monoSink.success(message);
            return true;
        }
        return false;
    }

    /**
     * 订阅
     */
    private Mono subscribe(String key) {
        synchronized (topicSubscribers) {
            if (!topicSubscribers.containsKey(key)) {
                return Mono.create(sink -> topicSubscribers.put(key, sink));
            }
        }
        return null;
    }

    /**
     * 取消订阅
     */
    private void unsubscribe(String key) {
        topicSubscribers.remove(key);
    }

    /*生成流水号*/
    private static String requestKey(Message request, Class responseClass) {
        String className = responseClass.getName();
        if (Response.class.isAssignableFrom(responseClass)) {
            String serNo = request.getSerNo();
            return new StringBuffer(34).append(className).append('.').append(serNo).toString();
        }
        return className;
    }

    /*返回流水号*/
    private static String responseKey(Object response) {
        String className = response.getClass().getName();
        if (response instanceof Response) {
            int serialNo = ((Response) response).getResponseSerialNo();
            return new StringBuffer(34).append(className).append('.').append(serialNo).toString();
        }
        return className;
    }

}