package com.business.socket;


import com.business.socket.bean.PushModel;
import com.sooka.common.utils.StringUtils;

import javax.websocket.Session;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/******************************
 * @description 发送消息
 * @author yanhongliang
 * @date 2019-08-09 15:40
 ******************************/

public class MessageManager {

    private static LinkedBlockingQueue<BatchBean> mMessageQueue = new LinkedBlockingQueue<BatchBean>();

    private static ThreadPoolExecutor mPoolExecutor = new ThreadPoolExecutor(
            5,
            10,
            15,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<Runnable>(),
            new ThreadPoolExecutor.AbortPolicy());

    /**
     * 添加消息到消息队列
     * @param receiverSession 接收者 Session
     * @param pushModel 发送的信息
     * @return 添加是否成功
     */
    public static boolean add(Session receiverSession, PushModel pushModel) {

        if (receiverSession == null|| pushModel==null
                || StringUtils.isEmpty(pushModel.getPushString())) {
            return false;
        }

        try {
            mMessageQueue.put(new BatchBean(receiverSession,pushModel.getPushString()));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return true;
    }

    public static void submit() {

        if (mMessageQueue.size() == 0) {
            return;
        }

        try {
            BatchBean batchBean = mMessageQueue.take();
            mPoolExecutor.execute(new SendMessageTask(batchBean, () -> submit()));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static int getCount() {
        return mMessageQueue.size();
    }

    public static void stop(){
        mMessageQueue.clear();
        mPoolExecutor.shutdownNow();
    }

    public interface OnComplete {
        void onComplete();
    }

    static class SendMessageTask implements Runnable{

        private BatchBean bean;
        private OnComplete onComplete;

        SendMessageTask(BatchBean bean, OnComplete onComplete){
            this.bean = bean;
            this.onComplete = onComplete;
        }
        @Override
        public void run() {
            try {

                bean.receiverSession.getBasicRemote().sendText("");
                if (bean.receiverSession != null) {
                    bean.receiverSession.getBasicRemote().sendText(bean.message);
                }

            }catch (Exception e){
                e.printStackTrace();
            }finally {
                onComplete.onComplete();
            }
        }
    }

    /**
     * 给每个人发送消息的bean的封装
     */
    private static class BatchBean {
        /**
         * 消息
         */
        private String message;

        /**
         * 接收者
         */
        private Session receiverSession;

        BatchBean( Session session,String message) {
            this.message = message;
            this.receiverSession = session;
        }
    }

}
