/*
 * Copyright (C) 2010-2101 Alibaba Group Holding Limited.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.alibaba.otter.shared.arbitrate.impl.setl.zookeeper;

import java.util.Date;

import org.I0Itec.zkclient.exception.ZkException;
import org.I0Itec.zkclient.exception.ZkNoNodeException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;

import com.alibaba.otter.shared.arbitrate.exception.ArbitrateException;
import com.alibaba.otter.shared.arbitrate.impl.config.ArbitrateConfigUtils;
import com.alibaba.otter.shared.arbitrate.impl.setl.ArbitrateFactory;
import com.alibaba.otter.shared.arbitrate.impl.setl.LoadArbitrateEvent;
import com.alibaba.otter.shared.arbitrate.impl.setl.helper.StagePathUtils;
import com.alibaba.otter.shared.arbitrate.impl.setl.monitor.PermitMonitor;
import com.alibaba.otter.shared.arbitrate.impl.setl.zookeeper.monitor.LoadStageListener;
import com.alibaba.otter.shared.arbitrate.impl.zookeeper.ZooKeeperClient;
import com.alibaba.otter.shared.arbitrate.model.EtlEventData;
import com.alibaba.otter.shared.arbitrate.model.TerminEventData;
import com.alibaba.otter.shared.arbitrate.model.TerminEventData.TerminType;
import com.alibaba.otter.shared.common.model.config.channel.ChannelStatus;
import com.alibaba.otter.shared.common.utils.JsonUtils;
import com.alibaba.otter.shared.common.utils.zookeeper.ZkClientx;

/**
 * 关注transformed节点，创建loaded节点
 * 
 * @version 4.0.3
 * 
 * <pre>
 * 1. 去除对应的DistributedLock的操作，减少中美同步时的latency. 每次DistributedLock操作都会涉及中美zk集群的交互，延迟在300ms
 * 
 * </pre>
 * @author jianghang 2011-8-9 下午05:10:50
 */
public class LoadZooKeeperArbitrateEvent implements LoadArbitrateEvent {

    private static final Logger           logger    = LoggerFactory.getLogger(LoadZooKeeperArbitrateEvent.class);
    private ZkClientx                     zookeeper = ZooKeeperClient.getInstance();
    private TerminZooKeeperArbitrateEvent terminEvent;

    // private Map<Long, DistributedLock> locks = new ConcurrentHashMap<Long,
    // DistributedLock>();

    /**
     * <pre>
     * 算法:
     * 1. 检查当前的Permit，阻塞等待其授权(解决Channel的pause状态处理)
     * 2. 开始阻塞获取符合条件的processId
     * 3. 检查当前的即时Permit状态 (在阻塞获取processId过程会出现一些error信号,process节点会被删除)
     * 4. 获取Select传递的EventData数据，添加next node信息后直接返回
     * </pre>
     */
    public EtlEventData await(Long pipelineId) throws InterruptedException {
        Assert.notNull(pipelineId);
        PermitMonitor permitMonitor = ArbitrateFactory.getInstance(pipelineId, PermitMonitor.class);
        permitMonitor.waitForPermit();// 阻塞等待授权

        LoadStageListener loadStageListener = ArbitrateFactory.getInstance(pipelineId, LoadStageListener.class);
        Long processId = loadStageListener.waitForProcess(); // 符合条件的processId

        // DistributedLock lock = getLock(pipelineId);
        try {
            // 使用锁的理由：
            // 1. 针对双向同步时，其中一个方向出现了异常，需要发起另一端的关闭，此时对方正好在执行某个process的load
            // 2. 单向同步时，如果出现node节点异常，此时正常的节点正在执行某个process的load
            // 为避免因load无法中端引起的数据重复录入，所以针对load阶段添加分布式锁。在有process
            // load过程中不允许进行pipeline关闭操作
            // lock.lock();

            ChannelStatus status = permitMonitor.getChannelPermit();
            if (status.isStart()) {// 即时查询一下当前的状态，状态随时可能会变
                // 根据pipelineId+processId构造对应的path
                String path = StagePathUtils.getTransformStage(pipelineId, processId);
                try {
                    byte[] data = zookeeper.readData(path);
                    return JsonUtils.unmarshalFromByte(data, EtlEventData.class);// 反序列化并返回
                } catch (ZkNoNodeException e) {
                    logger.error("pipeline[{}] processId[{}] is invalid , retry again", pipelineId, processId);
                    // try {
                    // lock.unlock();// 出现任何异常解除lock
                    // } catch (KeeperException e1) {
                    // // ignore
                    // }
                    return await(pipelineId);// /出现节点不存在，说明出现了error情况,递归调用重新获取一次
                } catch (ZkException e) {
                    throw e;
                }
            } else {
                logger.warn("pipelineId[{}] load ignore processId[{}] by status[{}]", new Object[] { pipelineId,
                        processId, status });
                // try {
                // lock.unlock();// 出现任何异常解除lock
                // } catch (KeeperException e) {
                // // ignore
                // }
                // 释放下processId，因为load是等待processId最小值完成Tranform才继续，如果这里不释放，会一直卡死等待
                String path = StagePathUtils.getProcess(pipelineId, processId);
                zookeeper.delete(path);
                return await(pipelineId);// 出现rollback情况，递归调用重新获取一次，当前的processId可丢弃
            }
        } catch (InterruptedException e) {
            throw e;
        } catch (Exception e) {
            // try {
            // lock.unlock();// 出现任何异常解除lock
            // } catch (KeeperException e1) {
            // // ignore
            // }

            throw new ArbitrateException(e);
        }
    }

    /**
     * <pre>
     * 算法:
     * 1. 创建对应的loaded节点,标志load已完成
     * </pre>
     * 
     * @param pipelineId 同步流id
     */
    public void single(EtlEventData data) {
        Assert.notNull(data);
        try {
            // String path = StagePathUtils.getLoadStage(data.getPipelineId(),
            // data.getProcessId());
            // // 序列化
            // data.setEndTime(new Date().getTime());// 返回当前时间
            // byte[] bytes = JsonUtils.marshalToByte(data,
            // SerializerFeature.WriteClassName);
            // try {
            // zookeeper.create(path, bytes, CreateMode.PERSISTENT);
            // } catch (NodeExistsException e) {
            // throw new ArbitrateException("Load_single", e.getMessage(), e);
            // } catch (KeeperException e) {
            // throw new ArbitrateException("Load_single", e.getMessage(), e);
            // } catch (InterruptedException e) {
            // // ignore
            // }

            data.setEndTime(new Date().getTime());// 返回当前时间

            // 调用Termin信号
            TerminEventData termin = new TerminEventData();
            termin.setPipelineId(data.getPipelineId());
            termin.setProcessId(data.getProcessId());
            termin.setStartTime(data.getStartTime());
            termin.setEndTime(data.getEndTime());
            termin.setFirstTime(data.getFirstTime());
            termin.setNumber(data.getNumber());
            termin.setBatchId(data.getBatchId());
            termin.setSize(data.getSize());
            termin.setExts(data.getExts());
            termin.setType(TerminType.NORMAL);
            termin.setCode("setl");
            termin.setDesc("");
            termin.setCurrNid(ArbitrateConfigUtils.getCurrentNid());
            terminEvent.single(termin);
        } finally {
            // // 最后一步释放锁
            // DistributedLock lock = getLock(data.getPipelineId());
            // try {
            // lock.unlock();// 解除lock
            // } catch (KeeperException e) {
            // throw new ArbitrateException("Load_single", e.getMessage(), e);
            // }
        }
    }

    /**
     * <pre>
     * 算法:
     * 1. load出现异常，解除load锁定，并发送对应的termin信号
     * </pre>
     * 
     * @param pipelineId 同步流id
     */
    public void release(Long pipelineId) {
        // DistributedLock lock = getLock(pipelineId);
        // try {
        // lock.unlock();// 解除lock
        // } catch (KeeperException e) {
        // throw new ArbitrateException("Load_single", e.getMessage(), e);
        // }
    }

    // private DistributedLock getLock(Long pipelineId) {
    // DistributedLock lock = locks.get(pipelineId);
    // if (lock == null) {
    // synchronized (locks) {
    // if (!locks.containsKey(pipelineId)) {
    // lock = new DistributedLock(StagePathUtils.getLoadLock(pipelineId));
    // locks.put(pipelineId, lock);
    // }
    // }
    // }
    //
    // return lock;
    // }

    // ======================== setter / getter ==========================

    public void setTerminEvent(TerminZooKeeperArbitrateEvent terminEvent) {
        this.terminEvent = terminEvent;
    }

}
