package com.easysocket.connection.dispatcher;

import com.easysocket.EasySocket;
import com.easysocket.entity.OriginReadData;
import com.easysocket.entity.SocketAddress;
import com.easysocket.interfaces.conn.IConnectionManager;
import com.easysocket.interfaces.conn.ISocketActionDispatch;
import com.easysocket.interfaces.conn.ISocketActionListener;
import com.easysocket.utils.Utils;

import java.io.Serializable;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;

import static com.easysocket.connection.action.IOAction.ACTION_READ_COMPLETE;
import static com.easysocket.connection.action.SocketAction.ACTION_CONN_FAIL;
import static com.easysocket.connection.action.SocketAction.ACTION_CONN_SUCCESS;
import static com.easysocket.connection.action.SocketAction.ACTION_DISCONNECTION;

/**
 * Author：Alex
 * Date：2019/6/1
 * Note：socket行为分发器
 */
public class SocketActionDispatcher implements ISocketActionDispatch {
    /**
     * 连接地址
     */
    private SocketAddress socketAddress;
    /**
     * 连接器
     */
    private IConnectionManager connectionManager;
    /**
     * 回调监听集合
     */
    private List<ISocketActionListener> actionListeners = new ArrayList<>();
    /**
     * 处理socket行为的线程
     */
    private Thread actionThread;
    /**
     * 是否停止分发
     */
    private boolean isStop;

    /**
     * 事件消费队列
     */
    private final LinkedBlockingQueue<ActionBean> socketActions = new LinkedBlockingQueue();
    /**
     * 切换到UI线程
     */
    private MainThreadExecutor mainThreadExecutor = new MainThreadExecutor();


    public SocketActionDispatcher(IConnectionManager connectionManager, SocketAddress socketAddress) {
        this.socketAddress = socketAddress;
        this.connectionManager = connectionManager;
    }

    public void setSocketAddress(SocketAddress info) {
        socketAddress = info;
    }


    @Override
    public void dispatchAction(String action) {
        dispatchAction(action, null);
    }

    @Override
    public void dispatchAction(String action, Serializable serializable) {
        // 将接收到的socket行为封装入列
        ActionBean actionBean = new ActionBean(action, serializable, this);
        socketActions.offer(actionBean);
    }

    @Override
    public void subscribe(ISocketActionListener iSocketActionListener) {
        if (iSocketActionListener != null && !actionListeners.contains(iSocketActionListener)) {
            actionListeners.add(iSocketActionListener);
        }
    }

    @Override
    public void unsubscribe(ISocketActionListener iSocketActionListener) {
        actionListeners.remove(iSocketActionListener);
    }

    /**
     * 分发线程
     */
    private class DispatchThread extends Thread {

        public DispatchThread() {
            super("dispatch thread");
        }

        @Override
        public void run() {
            // 循环处理socket的行为信息
            while (!isStop) {
                try {
                    ActionBean actionBean = socketActions.take();
                    if (actionBean != null && actionBean.mDispatcher != null) {
                        SocketActionDispatcher actionDispatcher = actionBean.mDispatcher;
                        List<ISocketActionListener> copyListeners = new ArrayList<>(actionDispatcher.actionListeners);
                        Iterator<ISocketActionListener> listeners = copyListeners.iterator();
                        // 通知所有监听者
                        while (listeners.hasNext()) {
                            ISocketActionListener listener = listeners.next();
                            actionDispatcher.dispatchActionToListener(actionBean.mAction, actionBean.arg, listener);
                        }
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * socket行为的封装
     */
    protected static class ActionBean {

        public ActionBean(String action, Serializable arg, SocketActionDispatcher dispatcher) {
            mAction = action;
            this.arg = arg;
            mDispatcher = dispatcher;
        }

        String mAction = "";
        Serializable arg;
        SocketActionDispatcher mDispatcher;
    }

    /**
     * 分发行为给监听者
     *
     * @param action
     * @param content
     * @param actionListener
     */
    private void dispatchActionToListener(String action, final Serializable content, final ISocketActionListener actionListener) {
        switch (action) {

            case ACTION_CONN_SUCCESS: // 连接成功
                mainThreadExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        actionListener.onSocketConnSuccess(socketAddress);
                    }
                });

                break;

            case ACTION_CONN_FAIL: // 连接失败
                mainThreadExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        actionListener.onSocketConnFail(socketAddress, ((Boolean) content).booleanValue());
                    }
                });

                break;

            case ACTION_DISCONNECTION: // 连接断开
                mainThreadExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        actionListener.onSocketDisconnect(socketAddress, ((Boolean) content).booleanValue());
                        // 不需要重连，则释放资源
                        if (!(Boolean) content) {
                            stopDispatchThread();
                        }
                    }
                });
                break;

            case ACTION_READ_COMPLETE: // 读取数据完成
                mainThreadExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        // response有三种形式
                        actionListener.onSocketResponse(socketAddress, (OriginReadData) content);
                        byte[] data = Utils.concatBytes(((OriginReadData) content).getHeaderData(), ((OriginReadData) content).getBodyBytes());
                        actionListener.onSocketResponse(socketAddress, new String(data, Charset.forName(EasySocket.getInstance().getDefOptions().getCharsetName())));
                        actionListener.onSocketResponse(socketAddress, data);
                    }
                });
                break;
        }
    }

    // 开始分发线程
    @Override
    public void startDispatchThread() {
        isStop = false;
        if (actionThread == null) {
            actionThread = new DispatchThread();
            actionThread.start();
        }
    }

    @Override
    public void stopDispatchThread() {
        if (actionThread != null && actionThread.isAlive() && !actionThread.isInterrupted()) {
            socketActions.clear();
            //actionListeners.clear();
            isStop = true;
            actionThread.interrupt();
            actionThread = null;
        }
    }

}