package org.bdware.sc.crdt.proxy;

import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import org.bdware.crdt.basic.JoinableCRDT;
import org.bdware.sc.crdt.SharableVarManager;
import org.bdware.sc.crdt.SharableVarState;
import org.bdware.sc.util.JsonUtil;

import java.util.Collections;
import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public abstract class SharableVar<T extends JoinableCRDT> {
    // 用于writer节点的变量副本
    public T writerVar;
    // 用于reader节点的变量副本
    public T readerVar;

    public Queue<JoinableCRDT> writerVarDeltaQueue;
    public Queue<JoinableCRDT> readerVarDeltaQueue;

    public String varId;
    public SharableVarState sharableVarState;

    public HashedWheelTimer readerTimer;
    public ReaderSyncTimerTask readerSyncTimerTask;
    public HashedWheelTimer writerTimer;
    public WriterSyncTimerTask writerSyncTimerTask;

    public SharableVar(String varId, String cpId,
            SharableVarState.SharableVarConfiguration resolveResult) {
        this.varId = varId;
        this.sharableVarState = new SharableVarState(cpId, resolveResult);

        if (this.sharableVarState.isReaderFlag()) {
            this.readerVar = createDeltaCrdt(cpId, varId);
            this.readerVarDeltaQueue = new LinkedList<>();
            if (this.sharableVarState.getReaderChildren() != null
                    && this.sharableVarState.getReaderChildren().size() > 0
                    && this.sharableVarState.getReaderInterval() != null
                    && this.sharableVarState.getReaderInterval() > 0) {
                this.readerTimer = new HashedWheelTimer(r -> {
                    Thread t = Executors.defaultThreadFactory().newThread(r);
                    t.setDaemon(true);
                    return t;
                }, 5, TimeUnit.MILLISECONDS, 2);
                this.readerTimer.newTimeout(timeout -> {
                    readerSyncTimerTask = new ReaderSyncTimerTask();
                    readerTimer.newTimeout(readerSyncTimerTask,
                            sharableVarState.getReaderInterval(), TimeUnit.SECONDS);
                }, this.sharableVarState.getReaderInterval(), TimeUnit.SECONDS);
            }
        }
        if (this.sharableVarState.isWriterFlag()) {
            this.writerVar = createDeltaCrdt(cpId, varId);
            this.writerVarDeltaQueue = new LinkedList<>();
            if (this.sharableVarState.getWriteInterval() != null
                    && this.sharableVarState.getWriteInterval() > 0) {
                this.writerTimer = new HashedWheelTimer(r -> {
                    Thread t = Executors.defaultThreadFactory().newThread(r);
                    t.setDaemon(true);
                    return t;
                }, 5, TimeUnit.MILLISECONDS, 2);
                this.writerTimer.newTimeout(timeout -> {
                    writerSyncTimerTask = new WriterSyncTimerTask();
                    writerTimer.newTimeout(writerSyncTimerTask, sharableVarState.getWriteInterval(),
                            TimeUnit.SECONDS);
                }, this.sharableVarState.getWriteInterval(), TimeUnit.SECONDS);
            }
        }
    }

    private void syncReaderVar() {
        if (readerVarDeltaQueue.isEmpty()) {
            return;
        }
        JoinableCRDT joinedDelta = createDeltaCrdt(null, varId);
        synchronized (readerVarDeltaQueue) {
            while (!readerVarDeltaQueue.isEmpty()) {
                JoinableCRDT delta = readerVarDeltaQueue.poll();
                joinedDelta.join(delta);
            }
        }
        String content = JsonUtil.toJson(joinedDelta);
        SharableVarManager.instance.broadcastSyncMessage(varId,
                sharableVarState.getReaderChildren(), content, "r2r");
    }

    private void syncWriterVar() {
        if (writerVarDeltaQueue.isEmpty()) {
            return;
        }
        JoinableCRDT joinedDelta = createDeltaCrdt(null, varId);
        synchronized (writerVarDeltaQueue) {
            while (!writerVarDeltaQueue.isEmpty()) {
                JoinableCRDT delta = writerVarDeltaQueue.poll();
                joinedDelta.join(delta);
            }
        }
        String content = JsonUtil.toJson(joinedDelta);
        if (sharableVarState.getWriterParent() != null) {
            // 父节点是Writer
            SharableVarManager.instance.broadcastSyncMessage(varId,
                    Collections.singletonList(sharableVarState.getWriterParent()), content, "w2w");
        } else if (sharableVarState.getReaderRoots() != null) {
            // 自己是writer根结点 向reader根结点们广播
            SharableVarManager.instance.broadcastSyncMessage(varId,
                    sharableVarState.getReaderRoots(), content, "w2r");
        }

    }

    abstract protected T createDeltaCrdt(String nodeId, String varId);

    class ReaderSyncTimerTask implements TimerTask {
        @Override
        public void run(Timeout timeout) throws Exception {
            try {
                syncReaderVar();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                readerTimer.newTimeout(readerSyncTimerTask, sharableVarState.getReaderInterval(),
                        TimeUnit.SECONDS);
            }
        }
    }

    class WriterSyncTimerTask implements TimerTask {
        @Override
        public void run(Timeout timeout) throws Exception {
            try {
                syncWriterVar();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                writerTimer.newTimeout(writerSyncTimerTask, sharableVarState.getWriteInterval(),
                        TimeUnit.SECONDS);
            }
        }
    }
}
