package com.xcong.excoin.websocket;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.huobi.client.model.Candlestick;
import com.xcong.excoin.common.contants.AppContants;
import com.xcong.excoin.modules.symbols.constants.SymbolsConstats;
import com.xcong.excoin.trade.CoinTraderFactory;
import com.xcong.excoin.utils.CoinTypeConvert;
import com.xcong.excoin.utils.RedisUtils;
import com.xcong.excoin.utils.SpringContextHolder;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@ServerEndpoint(value = "/trade/market")
@Component
public class TradePlateSendWebSocket {

    /**
     * 记录当前在线连接数
     */
    private static AtomicInteger onlineCount = new AtomicInteger(0);


    private static Map<String, Map<String, Session>> tradeplateClients = new ConcurrentHashMap<>();

    private static Map<String, Map<String, Session>> klineClients = new ConcurrentHashMap<>();

    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session) {
        onlineCount.incrementAndGet(); // 在线数加1
       // log.info("有新连接加入：{}，当前在线人数为：{}", session.getId(), onlineCount.get());
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose(Session session) {
        onlineCount.decrementAndGet(); // 在线数减1
        Collection<Map<String, Session>> values = tradeplateClients.values();
        if (CollectionUtils.isNotEmpty(values)) {
            for (Map<String, Session> map : values) {
                map.remove(session.getId());
            }
        }

        Collection<Map<String, Session>> klineClientsValues = klineClients.values();
        if (CollectionUtils.isNotEmpty(klineClientsValues)) {
            for (Map<String, Session> map : klineClientsValues) {
                map.remove(session.getId());
            }
        }
        //log.info("有一连接关闭：{}，当前在线人数为：{}", session.getId(), onlineCount.get());
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message 客户端发送过来的消息
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        // 盘口订阅方法 {sub: 'market.btcusdt.depth.10,id: symbol}
        JSONObject jsonObject = JSON.parseObject(message);
//        log.info("订阅参数:{}", jsonObject);
        // 盘口的判断
        if (jsonObject.containsKey("sub") && jsonObject.get("sub").toString().contains("depth")) {

            String sub = jsonObject.get("sub").toString();
            String symbol = sub.split("\\.")[1];
            symbol = CoinTypeConvert.convert(symbol);
            if (tradeplateClients.containsKey(symbol)) {
                tradeplateClients.get(symbol).put(session.getId(), session);
            } else {
                Map<String, Session> map = new HashMap<>();
                map.put(session.getId(), session);
                tradeplateClients.put(symbol, map);
            }
            // 发送一次盘口
            CoinTraderFactory factory = SpringContextHolder.getBean(CoinTraderFactory.class);
            // 发送订阅消息
            String nekk = factory.getTrader(SymbolsConstats.ROC).sendTradePlateMessage();
            SubResultModel subResultModel = new SubResultModel();
            subResultModel.setId("griceusdt");
            subResultModel.setSubbed(sub);
            synchronized (session) {
                try {
                    session.getBasicRemote().sendText(JSONObject.toJSONString(subResultModel));
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            synchronized (session) {
                try {
                    session.getBasicRemote().sendText(nekk);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }


        }
        // 取消盘口订阅
        if (jsonObject.containsKey("unsub") && jsonObject.get("unsub").toString().contains("depth")) {
            // {unsub:'market.btcusdt.kline.1min'}
            String unsub = jsonObject.get("unsub").toString();
            String[] split = unsub.split("\\.");
            String symbol = split[1];
            symbol = CoinTypeConvert.convert(symbol);
            String key = symbol;
            if (tradeplateClients.containsKey(key)) {
                tradeplateClients.get(key).remove(session.getId());
            }
        }

        // 最新K线订阅
        if (jsonObject.containsKey("sub") && jsonObject.get("sub").toString().contains("kline")) {
            // 订阅
            // {sub: 'market.btcusdt.kline.1min'}
            String sub = jsonObject.get("sub").toString();
            String[] split = sub.split("\\.");
            String symbol = split[1];
            symbol = CoinTypeConvert.convert(symbol);
            String period = split[3];
            if ("60min".equals(period)) {
                period = "1hour";
            }
            String key = symbol + "-" + period;
            if (klineClients.containsKey(key)) {
                // 有这个币种K线
                Map<String, Session> stringSessionMap = klineClients.get(key);
                if (!stringSessionMap.containsKey(session.getId())) {
                    stringSessionMap.put(session.getId(), session);
                }
            } else {
                Map<String, Session> stringSessionMap = new HashMap<>();
                stringSessionMap.put(session.getId(), session);
                klineClients.put(key, stringSessionMap);
            }
            // 给他发送最新K线 TODO
            String newKline = "NEW_KINE_{}";
            key = StrUtil.format(newKline, symbol);
            RedisUtils redisUtils = SpringContextHolder.getBean(RedisUtils.class);
            Object o = redisUtils.get(key);
            Map<String, Candlestick> currentKlineMap = (Map<String, Candlestick>) o;
            String ch = "market.{}.kline.{}";
            String chKey = period;
            if (period.equals("60min")) {
                chKey = "1hour";
            }
            if(currentKlineMap!=null && currentKlineMap.containsKey(chKey)){
                Candlestick value = currentKlineMap.get(chKey);
                // 转换
                NewCandlestick newCandlestick = new NewCandlestick();
                String nekkusdt = split[1];
                ch = StrUtil.format(ch, nekkusdt, period);
                newCandlestick.setCh(ch);
                CandlestickModel model = new CandlestickModel();
                model.setVol(value.getVolume());
                model.setLow(value.getLow());
                model.setOpen(value.getOpen());
                model.setHigh(value.getHigh());
                model.setCount(value.getCount());
                model.setAmount(value.getAmount());
                model.setId(value.getTimestamp() / 1000);
                model.setTimestamp(value.getTimestamp() / 1000);
                model.setClose(value.getClose());
                newCandlestick.setTick(model);
                sendMessageKlineNow(JSONObject.toJSONString(newCandlestick), session);
            }
        }

        // 取消订阅
        if (jsonObject.containsKey("unsub") && jsonObject.get("unsub").toString().contains("kline")) {
            // {unsub:'market.${symbol}.kline.${strPeriod}'}
            String unsub = jsonObject.get("unsub").toString();
            String[] split = unsub.split("\\.");
            String strPeriod = split[3];
            if ("60min".equals(strPeriod)) {
                strPeriod = "1hour";
            }
            String symbol = split[1];
            symbol = CoinTypeConvert.convert(symbol);
            String key = symbol + "-" + strPeriod;
            if (klineClients.containsKey(key)) {
                klineClients.get(key).remove(session.getId());
            }
        }

        // 历史K线订阅
        // {req: "market.nekkusdt.kline.1min", symbol: "nekkusdt", period: "1min"}
        if (jsonObject.containsKey("req") && jsonObject.get("req").toString().contains("kline")) {
            String sub = jsonObject.get("req").toString();
            String[] split = sub.split("\\.");
            String symbol = split[1];
            symbol = CoinTypeConvert.convert(symbol);
            String period = split[3];
            if ("60min".equals(period)) {
                period = "1hour";
            }
            //String key = symbol+"-"+period;
            // String key = "KINE_BCH/USDT_1week";
            String key = "KINE_{}_{}";
            // 币币k线数据
            //key = StrUtil.format(key, symbol, period);
            key = StrUtil.format(key, "GRICE/USDT", period);
            RedisUtils bean = SpringContextHolder.getBean(RedisUtils.class);
            Object o = bean.get(key);
            List<CandlestickModel> candlestickModels = new ArrayList<>();
            CandlestickResult result = new CandlestickResult();
            result.setRep(sub);
            if (o != null) {
                List<Candlestick> list = (List<Candlestick>) o;

                if(list!=null && list.size()>300){
                    int size = list.size();
                    list = list.subList(size-300,size);
                }
                CandlestickModel model = null;
                for (Candlestick candlestick : list) {
                    model = new CandlestickModel();
                    model.setAmount(candlestick.getAmount());
                    model.setClose(candlestick.getClose());
                    model.setCount(candlestick.getCount());
                    model.setHigh(candlestick.getHigh());
                    model.setId(candlestick.getTimestamp() / 1000);
                    model.setOpen(candlestick.getOpen());
                    model.setLow(candlestick.getLow());
                    model.setVol(candlestick.getVolume());
                    candlestickModels.add(model);
                }
                result.setData(candlestickModels);
            }

            sendMessageHistory(JSON.toJSONString(result), session);
        }
    }

    @OnError
    public void onError(Session session, Throwable error) {
       // log.error("发生错误");
        //error.printStackTrace();
    }

    /**
     * 发送盘口消息
     *
     * @param message 消息内容
     */
    public void sendMessagePlate(String symbol, String message, Session fromSession) {
        if (tradeplateClients.containsKey(symbol)) {
            Map<String, Session> nekk = tradeplateClients.get(symbol);
            for (Map.Entry<String, Session> sessionEntry : nekk.entrySet()) {
                Session toSession = sessionEntry.getValue();
                // 排除掉自己
                boolean open = toSession.isOpen();
                if (open) {
                    toSession.getAsyncRemote().sendText(message);
                }
            }
        }

    }

    /**
     * 发送历史K线
     *
     * @param message, toSession
     * @return void
     */
    public void sendMessageHistory(String message, Session toSession) {
        boolean open = toSession.isOpen();
        if (open) {
            toSession.getAsyncRemote().sendText(message);
        }
    }

    /**
     * 发送最新K线
     *
     * @param symbol, period, message, fromSession
     * @return void
     */
    public void sendMessageKline(String symbol, String period, String message, Session fromSession) {
        String key = symbol + "-" + period;
        if (klineClients.containsKey(key)) {
            Map<String, Session> stringSessionMap = klineClients.get(key);
            for (Map.Entry<String, Session> sessionEntry : stringSessionMap.entrySet()) {
                Session toSession = sessionEntry.getValue();
                boolean open = toSession.isOpen();
                if (open) {
                    toSession.getAsyncRemote().sendText(message);
                }
            }
        }
    }

    public void sendMessageKlineNow(String message, Session toSession) {
        boolean open = toSession.isOpen();
        if (open) {
            toSession.getAsyncRemote().sendText(message);
        }
    }
}