/********************************************************************
 * 2014 -
 * open source under Apache License Version 2.0
 ********************************************************************/
/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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.
 */
#include "Atomic.h"
#include "DateTime.h"
#include "Exception.h"
#include "ExceptionInternal.h"
#include "FileSystemInter.h"
#include "HWCrc32c.h"
#include "LeaseRenewer.h"
#include "Logger.h"
#include "OutputStream.h"
#include "OutputStreamImpl.h"
#include "Packet.h"
#include "PacketHeader.h"
#include "SWCrc32c.h"

#include <cassert>
#include <inttypes.h>

namespace Hdfs {
namespace Internal {

OutputStreamImpl::OutputStreamImpl() :
/*heartBeatStop(true),*/ closed(true), isAppend(false), syncBlock(false), checksumSize(0), chunkSize(
        0), chunksPerPacket(0), closeTimeout(0), heartBeatInterval(0), packetSize(0), position(
            0), replication(0), blockSize(0), bytesWritten(0), cursor(0), lastFlushed(
                0), nextSeqNo(0), packets(0), cryptoCodec(NULL), kcp(NULL) {
    if (HWCrc32c::available()) {
        checksum = shared_ptr < Checksum > (new HWCrc32c());
    } else {
        checksum = shared_ptr < Checksum > (new SWCrc32c());
    }

    checksumSize = sizeof(int32_t);
    lastSend = steady_clock::now();
#ifdef MOCK
    stub = NULL;
#endif
}

OutputStreamImpl::~OutputStreamImpl() {
    if (!closed) {
        try {
            close();
        } catch (...) {
        }
    }
}

void OutputStreamImpl::checkStatus() {
    if (closed) {
        THROW(HdfsIOException, "OutputStreamImpl: stream is not opened.");
    }

    lock_guard < mutex > lock(mut);

    if (lastError != exception_ptr()) {
        rethrow_exception(lastError);
    }
}

void OutputStreamImpl::setError(const exception_ptr & error) {
    try {
        lock_guard < mutex > lock(mut);
        lastError = error;
    } catch (...) {
    }
}

shared_ptr<CryptoCodec> OutputStreamImpl::getCryptoCodec() {
    return cryptoCodec;
}

void OutputStreamImpl::setCryptoCodec(shared_ptr<CryptoCodec> cryptoCodec) {
    this->cryptoCodec = cryptoCodec;
}

shared_ptr<KmsClientProvider> OutputStreamImpl::getKmsClientProvider() {
    return kcp;
}

void OutputStreamImpl::setKmsClientProvider(shared_ptr<KmsClientProvider> kcp) {
    this->kcp = kcp;
}
/**
 * To create or append a file.
 * @param fs hdfs file system.
 * @param path the file path.
 * @param flag creation flag, can be Create, Append or Create|Overwrite.
 * @param permission create a new file with given permission.
 * @param createParent if the parent does not exist, create it.
 * @param replication create a file with given number of replication.
 * @param blockSize  create a file with given block size.
 */
void OutputStreamImpl::open(shared_ptr<FileSystemInter> fs, const char * path, int flag,
                            const Permission & permission, bool createParent, int replication,
                            int64_t blockSize) {
    if (NULL == path || 0 == strlen(path) || replication < 0 || blockSize < 0) {
        THROW(InvalidParameter, "Invalid parameter.");
    }

    if (!(flag == Create || flag == (Create | SyncBlock) || flag == Overwrite
            || flag == (Overwrite | SyncBlock) || flag == Append
            || flag == (Append | SyncBlock) || flag == (Create | Overwrite)
            || flag == (Create | Overwrite | SyncBlock)
            || flag == (Create | Append)
            || flag == (Create | Append | SyncBlock))) {
        THROW(InvalidParameter, "Invalid flag.");
    }

    try {
        openInternal(fs, path, flag, permission, createParent, replication,
                     blockSize);
    } catch (...) {
        reset();
        throw;
    }
}

void OutputStreamImpl::computePacketChunkSize() {
    int chunkSizeWithChecksum = chunkSize + checksumSize;
    static const int packetHeaderSize = PacketHeader::GetPkgHeaderSize();
    chunksPerPacket =
        (packetSize - packetHeaderSize + chunkSizeWithChecksum - 1)
        / chunkSizeWithChecksum;
    chunksPerPacket = chunksPerPacket > 1 ? chunksPerPacket : 1;
    packetSize = chunksPerPacket * chunkSizeWithChecksum + packetHeaderSize;
    buffer.resize(chunkSize);
}

void OutputStreamImpl::initAppend() {
    FileStatus fileInfo;
    std::pair<shared_ptr<LocatedBlock>, shared_ptr<FileStatus> > lastBlockWithStatus;
    lastBlockWithStatus = filesystem->append(this->path);
    lastBlock = lastBlockWithStatus.first;

    if (lastBlockWithStatus.second) {
        fileInfo = *lastBlockWithStatus.second;
    } else {
        fileInfo = filesystem->getFileStatus(this->path.c_str());
    }

    closed = false;

    try {
        this->blockSize = fileInfo.getBlockSize();
        cursor = fileInfo.getLength();

        if (lastBlock) {
            isAppend = true;
            bytesWritten = lastBlock->getNumBytes();
            int64_t usedInLastBlock = fileInfo.getLength() % blockSize;
            int64_t freeInLastBlock = blockSize - usedInLastBlock;

            if (freeInLastBlock == this->blockSize) {
                THROW(HdfsIOException,
                      "OutputStreamImpl: the last block for file %s is full.",
                      this->path.c_str());
            }

            int usedInCksum = cursor % chunkSize;
            int freeInCksum = chunkSize - usedInCksum;

            if (usedInCksum > 0 && freeInCksum > 0) {
                /*
                 * if there is space in the last partial chunk, then
                 * setup in such a way that the next packet will have only
                 * one chunk that fills up the partial chunk.
                 */
                packetSize = 0;
                chunkSize = freeInCksum;
            } else {
                /*
                 * if the remaining space in the block is smaller than
                 * that expected size of of a packet, then create
                 * smaller size packet.
                 */
                packetSize =
                    packetSize < freeInLastBlock ?
                    packetSize : static_cast<int>(freeInLastBlock);
            }
        }
    } catch (...) {
        completeFile(false);
        reset();
        throw;
    }

    computePacketChunkSize();
}

void OutputStreamImpl::openInternal(shared_ptr<FileSystemInter> fs, const char * path,
                                    int flag, const Permission & permission, bool createParent,
                                    int replication, int64_t blockSize) {
    filesystem = fs;
    this->path = fs->getStandardPath(path);
    this->replication = replication;
    this->blockSize = blockSize;
    syncBlock = flag & SyncBlock;
    conf = shared_ptr < SessionConfig > (new SessionConfig(fs->getConf()));
    LOG(DEBUG2, "open file %s for %s", this->path.c_str(), (flag & Append ? "append" : "write"));
    packets.setMaxSize(conf->getPacketPoolSize());

    if (0 == replication) {
        this->replication = conf->getDefaultReplica();
    } else {
        this->replication = replication;
    }

    if (0 == blockSize) {
        this->blockSize = conf->getDefaultBlockSize();
    } else {
        this->blockSize = blockSize;
    }

    chunkSize = conf->getDefaultChunkSize();
    packetSize = conf->getDefaultPacketSize();
    heartBeatInterval = conf->getHeartBeatInterval();
    closeTimeout = conf->getCloseFileTimeout();

    if (packetSize < chunkSize) {
        THROW(InvalidParameter,
              "OutputStreamImpl: packet size %d is less than the chunk size %d.",
              packetSize, chunkSize);
    }

    if (0 != this->blockSize % chunkSize) {
        THROW(InvalidParameter,
              "OutputStreamImpl: block size %" PRId64 " is not the multiply of chunk size %d.",
              this->blockSize, chunkSize);
    }

    try {
        if (flag & Append) {
            fileStatus = fs->getFileStatus(this->path.c_str());
            FileEncryptionInfo *fileEnInfo = fileStatus.getFileEncryption();
            if (fileStatus.isFileEncrypted()) {
                if (cryptoCodec == NULL) {
                    auth = shared_ptr<RpcAuth> (
                            new RpcAuth(fs->getUserInfo(), RpcAuth::ParseMethod(conf->getKmsMethod())));
                    kcp = shared_ptr<KmsClientProvider> (
                            new KmsClientProvider(auth, conf));
                    cryptoCodec = shared_ptr<CryptoCodec> (
                            new CryptoCodec(fileEnInfo, kcp, conf->getCryptoBufferSize()));

                    int64_t file_length = fileStatus.getLength();
                    int ret = cryptoCodec->init(CryptoMethod::ENCRYPT, file_length);
                    if (ret < 0) {
                        THROW(HdfsIOException, "init CryptoCodec failed, file:%s", this->path.c_str());
                    }
                }
            }
            initAppend();
            LeaseRenewer::GetLeaseRenewer().StartRenew(filesystem);
            return;
        }
    } catch (const FileNotFoundException & e) {
        if (!(flag & Create)) {
            throw;
        }
    }

    assert((flag & Create) || (flag & Overwrite));
    fs->create(this->path, permission, flag, createParent, this->replication,
            this->blockSize);
    fileStatus = fs->getFileStatus(this->path.c_str());
    FileEncryptionInfo *fileEnInfo = fileStatus.getFileEncryption();
    if (fileStatus.isFileEncrypted()) {
        if (cryptoCodec == NULL) {
            auth = shared_ptr<RpcAuth>(
                    new RpcAuth(fs->getUserInfo(), RpcAuth::ParseMethod(conf->getKmsMethod())));
            kcp = shared_ptr<KmsClientProvider>(
                    new KmsClientProvider(auth, conf));
            cryptoCodec = shared_ptr<CryptoCodec>(
                    new CryptoCodec(fileEnInfo, kcp, conf->getCryptoBufferSize()));

            int64_t file_length = fileStatus.getLength();
            assert(file_length == 0);
            int ret = cryptoCodec->init(CryptoMethod::ENCRYPT, file_length);
            if (ret < 0) {
                THROW(HdfsIOException, "init CryptoCodec failed, file:%s", this->path.c_str());
            }
        }
    }
    closed = false;
    computePacketChunkSize();
    LeaseRenewer::GetLeaseRenewer().StartRenew(filesystem);
}

/**
 * To append data to file.
 * @param buf the data used to append.
 * @param size the data size.
 */
void OutputStreamImpl::append(const char * buf, int64_t size) {
    LOG(DEBUG3, "append file %s size is %" PRId64 ", offset %" PRId64 " next pos %" PRId64, path.c_str(), size, cursor, size + cursor);

    if (NULL == buf || size < 0) {
        THROW(InvalidParameter, "Invalid parameter.");
    }

    checkStatus();

    try {
        appendInternal(buf, size);
    } catch (...) {
        setError(current_exception());
        throw;
    }
}

void OutputStreamImpl::appendInternal(const char * buf, int64_t size) {
    int64_t todo = size;
	std::string bufEncode;

    if (fileStatus.isFileEncrypted()) {
        //encrypt buf
        bufEncode = cryptoCodec->cipher_wrap(buf, size);
        buf = bufEncode.c_str();

    }
    while (todo > 0) {
        int batch = buffer.size() - position;
        batch = batch < todo ? batch : static_cast<int>(todo);

        /*
         * bypass buffer.
         */
        if (0 == position && todo >= static_cast<int64_t>(buffer.size())) {
            checksum->update(buf + size - todo, batch);
            appendChunkToPacket(buf + size - todo, batch);
            bytesWritten += batch;
            checksum->reset();
        } else {
            checksum->update(buf + size - todo, batch);
            memcpy(&buffer[position], buf + size - todo, batch);
            position += batch;

            if (position == static_cast<int>(buffer.size())) {
                appendChunkToPacket(&buffer[0], buffer.size());
                bytesWritten += buffer.size();
                checksum->reset();
                position = 0;
            }
        }

        todo -= batch;

        if (currentPacket
                && (currentPacket->isFull() || bytesWritten == blockSize)) {
            sendPacket(currentPacket);

            if (isAppend) {
                isAppend = false;
                chunkSize = conf->getDefaultChunkSize();
                packetSize = conf->getDefaultPacketSize();
                computePacketChunkSize();
            }

            if (bytesWritten == blockSize) {
                closePipeline();
            }
        }
    }

    cursor += size;
}

void OutputStreamImpl::appendChunkToPacket(const char * buf, int size) {
    assert(NULL != buf && size > 0);

    if (!currentPacket) {
        currentPacket = packets.getPacket(packetSize, chunksPerPacket, bytesWritten,
                                          nextSeqNo++, checksumSize);
    }

    currentPacket->addChecksum(checksum->getValue());
    currentPacket->addData(buf, size);
    currentPacket->increaseNumChunks();
}

void OutputStreamImpl::sendPacket(shared_ptr<Packet> packet) {
    if (!pipeline) {
        setupPipeline();
    }

    pipeline->send(currentPacket);
    currentPacket.reset();
    lastSend = steady_clock::now();
}

void OutputStreamImpl::setupPipeline() {
    assert(currentPacket);
#ifdef MOCK
    pipeline = stub->getPipeline();
#else
    pipeline = shared_ptr<Pipeline>(new PipelineImpl(isAppend, path.c_str(), *conf, filesystem,
                                    CHECKSUM_TYPE_CRC32C, conf->getDefaultChunkSize(), replication,
                                    currentPacket->getOffsetInBlock(), packets, lastBlock));
#endif
    lastSend = steady_clock::now();
    /*
     * start heart beat beat thread
     */
    /*if (heartBeatStop) {
        if (heartBeatSender.joinable()) {
            heartBeatSender.join();
        }

        heartBeatStop = false;
        heartBeatSender = thread(&OutputStreamImpl::heartBeatSenderRoutine, this);
    }*/
}

/**
 * Flush all data in buffer and waiting for ack.
 * Will block until get all acks.
 */
void OutputStreamImpl::flush() {
    LOG(DEBUG3, "flush file %s at offset %" PRId64, path.c_str(), cursor);
    checkStatus();

    try {
        flushInternal(false);
    } catch (...) {
        setError(current_exception());
        throw;
    }
}

void OutputStreamImpl::flushInternal(bool needSync) {
    if (lastFlushed == cursor && !needSync) {
        return;
    } else {
        lastFlushed = cursor;
    }

    if (position > 0) {
        appendChunkToPacket(&buffer[0], position);
    }

    /*
     * if the pipeline and currentPacket are both NULL,
     * that means the pipeline has been closed and no more data in buffer/packet.
     * already synced when closing pipeline.
     */
    if (!currentPacket && needSync && pipeline) {
        currentPacket = packets.getPacket(packetSize, chunksPerPacket, bytesWritten,
                                          nextSeqNo++, checksumSize);
    }

    lock_guard < mutex > lock(mut);

    if (currentPacket) {
        currentPacket->setSyncFlag(needSync);
        sendPacket(currentPacket);
    }

    if (pipeline) {
        pipeline->flush();
    }
}

/**
 * return the current file length.
 * @return current file length.
 */
int64_t OutputStreamImpl::tell() {
    checkStatus();
    return cursor;
}

/**
 * @ref OutputStream::sync
 */
void OutputStreamImpl::sync() {
    LOG(DEBUG3, "sync file %s at offset %" PRId64, path.c_str(), cursor);
    checkStatus();

    try {
        flushInternal(true);
    } catch (...) {
        setError(current_exception());
        throw;
    }
}

void OutputStreamImpl::completeFile(bool throwError) {
    steady_clock::time_point start = steady_clock::now();

    while (true) {
        try {
            bool success;
            success = filesystem->complete(path, lastBlock.get());

            if (success) {
                return;
            }
        } catch (HdfsIOException & e) {
            if (throwError) {
                NESTED_THROW(HdfsIOException,
                             "OutputStreamImpl: failed to complete file %s.",
                             path.c_str());
            } else {
                return;
            }
        }

        if (closeTimeout > 0) {
            steady_clock::time_point end = steady_clock::now();

            if (ToMilliSeconds(start, end) >= closeTimeout) {
                if (throwError) {
                    THROW(HdfsIOException,
                          "OutputStreamImpl: timeout when complete file %s, timeout interval %d ms.",
                          path.c_str(), closeTimeout);
                } else {
                    return;
                }
            }
        }

        try {
            sleep_for(milliseconds(400));
        } catch (...) {
        }
    }
}

/**
 * close the stream.
 */
void OutputStreamImpl::closePipeline() {
    lock_guard < mutex > lock(mut);

    if (!pipeline) {
        return;
    }

    if (currentPacket) {
        sendPacket(currentPacket);
    }

    currentPacket = packets.getPacket(packetSize, chunksPerPacket, bytesWritten, nextSeqNo++,
                                      checksumSize);

    if (syncBlock) {
        currentPacket->setSyncFlag(syncBlock);
    }

    lastBlock = pipeline->close(currentPacket);
    assert(lastBlock);
    currentPacket.reset();
    pipeline.reset();
    filesystem->fsync(path);
    bytesWritten = 0;
}

void OutputStreamImpl::close() {
    exception_ptr e;

    if (closed) {
        return;
    }

    try {
        //pipeline may be broken
        if (!lastError) {
            if (lastFlushed != cursor && position > 0) {
                appendChunkToPacket(&buffer[0], position);
            }

            if (lastFlushed != cursor && currentPacket) {
                sendPacket(currentPacket);
            }

            closePipeline();
            /*heartBeatStop = true;
             condHeartBeatSender.notify_all();

             if (heartBeatSender.joinable()) {
             heartBeatSender.join();
             }*/
            completeFile(true);
        }
    } catch (...) {
        e = current_exception();
    }

    LeaseRenewer::GetLeaseRenewer().StopRenew(filesystem);
    LOG(DEBUG3, "close file %s for write with length %" PRId64, path.c_str(), cursor);
    reset();

    if (e) {
        rethrow_exception(e);
    }
}

void OutputStreamImpl::reset() {
    blockSize = 0;
    bytesWritten = 0;
    checksum->reset();
    chunkSize = 0;
    chunksPerPacket = 0;
    closed = true;
    closeTimeout = 0;
    conf.reset();
    currentPacket.reset();
    cursor = 0;
    filesystem.reset();
    heartBeatInterval = 0;
    isAppend = false;
    lastBlock.reset();
    lastError = exception_ptr();
    lastFlushed = 0;
    nextSeqNo = 0;
    packetSize = 0;
    path.clear();
    pipeline.reset();
    position = 0;
    replication = 0;
    syncBlock = false;
}

std::string OutputStreamImpl::toString() {
    if (path.empty()) {
        return std::string("OutputStream for path ") + path;
    } else {
        return std::string("OutputStream (not opened)");
    }
}

/*void OutputStreamImpl::heartBeatSenderRoutine() {
    assert(heartBeatStop == false);

    while (!heartBeatStop) {
        try {
            unique_lock < mutex > lock(mut);
            condHeartBeatSender.wait_for(lock, milliseconds(1000));

            try {
                try {
                    if (pipeline
                            && ToMilliSeconds(lastSend, steady_clock::now())
                            >= heartBeatInterval) {
                        pipeline->send(shared_ptr < Packet > (new Packet()));
                        lastSend = steady_clock::now();
                    }
                } catch (...) {
                    NESTED_THROW(Hdfs::HdfsIOException, "Failed to send heart beat, path: %s",
                                 path.c_str());
                }
            } catch (...) {
                lastError = current_exception();
                throw;
            }
        } catch (const std::bad_alloc & e) {

             * keep quiet if we run out of memory, since writing log need memory,
             * that may cause the process terminated.

            break;
        } catch (const Hdfs::HdfsException & e) {
            LOG(LOG_ERROR, "Heart beat thread exit since %s",
                GetExceptionDetail(e));
        } catch (const std::exception & e) {
            LOG(LOG_ERROR, "Heart beat thread exit since %s",
                e.what());
        }
    }

    heartBeatStop = true;
}*/

}
}