/*
 * 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 <thrift/thrift-config.h>

#include <thrift/transport/TFileTransport.h>
#include <thrift/transport/TTransportUtils.h>
#include <thrift/transport/PlatformSocket.h>
#include <thrift/concurrency/FunctionRunner.h>

#include <boost/version.hpp>

#ifdef HAVE_SYS_TIME_H
#include <sys/time.h>
#else
#include <time.h>
#endif
#include <fcntl.h>
#ifdef HAVE_UNISTD_H
#include <unistd.h>
#endif
#ifdef HAVE_STRINGS_H
#include <strings.h>
#endif
#include <cstdlib>
#include <cstring>
#include <iostream>
#include <limits>
#include <memory>
#ifdef HAVE_SYS_STAT_H
#include <sys/stat.h>
#endif

#ifdef _WIN32
#include <io.h>
#endif

namespace apache {
namespace thrift {
namespace transport {

using std::shared_ptr;
using std::cerr;
using std::cout;
using std::endl;
using std::string;
using namespace apache::thrift::protocol;
using namespace apache::thrift::concurrency;

TFileTransport::TFileTransport(string path, bool readOnly)
  : readState_(),
    readBuff_(nullptr),
    currentEvent_(nullptr),
    readBuffSize_(DEFAULT_READ_BUFF_SIZE),
    readTimeout_(NO_TAIL_READ_TIMEOUT),
    chunkSize_(DEFAULT_CHUNK_SIZE),
    eventBufferSize_(DEFAULT_EVENT_BUFFER_SIZE),
    flushMaxUs_(DEFAULT_FLUSH_MAX_US),
    flushMaxBytes_(DEFAULT_FLUSH_MAX_BYTES),
    maxEventSize_(DEFAULT_MAX_EVENT_SIZE),
    maxCorruptedEvents_(DEFAULT_MAX_CORRUPTED_EVENTS),
    eofSleepTime_(DEFAULT_EOF_SLEEP_TIME_US),
    corruptedEventSleepTime_(DEFAULT_CORRUPTED_SLEEP_TIME_US),
    writerThreadIOErrorSleepTime_(DEFAULT_WRITER_THREAD_SLEEP_TIME_US),
    dequeueBuffer_(nullptr),
    enqueueBuffer_(nullptr),
    notFull_(&mutex_),
    notEmpty_(&mutex_),
    closing_(false),
    flushed_(&mutex_),
    forceFlush_(false),
    filename_(path),
    fd_(0),
    bufferAndThreadInitialized_(false),
    offset_(0),
    lastBadChunk_(0),
    numCorruptedEventsInChunk_(0),
    readOnly_(readOnly) {
  threadFactory_.setDetached(false);
  openLogFile();
}

void TFileTransport::resetOutputFile(int fd, string filename, off_t offset) {
  filename_ = filename;
  offset_ = offset;

  // check if current file is still open
  if (fd_ > 0) {
    // flush any events in the queue
    flush();
    GlobalOutput.printf("error, current file (%s) not closed", filename_.c_str());
    if (-1 == ::THRIFT_CLOSE(fd_)) {
      int errno_copy = THRIFT_ERRNO;
      GlobalOutput.perror("TFileTransport: resetOutputFile() ::close() ", errno_copy);
      throw TTransportException(TTransportException::UNKNOWN,
                                "TFileTransport: error in file close",
                                errno_copy);
    } else {
      // successfully closed fd
      fd_ = 0;
    }
  }

  if (fd) {
    fd_ = fd;
  } else {
    // open file if the input fd is 0
    openLogFile();
  }
}

TFileTransport::~TFileTransport() {
  // flush the buffer if a writer thread is active
  if (writerThread_.get()) {
    // set state to closing
    closing_ = true;

    // wake up the writer thread
    // Since closing_ is true, it will attempt to flush all data, then exit.
    notEmpty_.notify();

    writerThread_->join();
    writerThread_.reset();
  }

  if (dequeueBuffer_) {
    delete dequeueBuffer_;
    dequeueBuffer_ = nullptr;
  }

  if (enqueueBuffer_) {
    delete enqueueBuffer_;
    enqueueBuffer_ = nullptr;
  }

  if (readBuff_) {
    delete[] readBuff_;
    readBuff_ = nullptr;
  }

  if (currentEvent_) {
    delete currentEvent_;
    currentEvent_ = nullptr;
  }

  // close logfile
  if (fd_ > 0) {
    if (-1 == ::THRIFT_CLOSE(fd_)) {
      GlobalOutput.perror("TFileTransport: ~TFileTransport() ::close() ", THRIFT_ERRNO);
    } else {
      // successfully closed fd
      fd_ = 0;
    }
  }
}

bool TFileTransport::initBufferAndWriteThread() {
  if (bufferAndThreadInitialized_) {
    T_ERROR("%s", "Trying to double-init TFileTransport");
    return false;
  }

  if (!writerThread_.get()) {
    writerThread_ = threadFactory_.newThread(
        apache::thrift::concurrency::FunctionRunner::create(startWriterThread, this));
    writerThread_->start();
  }

  dequeueBuffer_ = new TFileTransportBuffer(eventBufferSize_);
  enqueueBuffer_ = new TFileTransportBuffer(eventBufferSize_);
  bufferAndThreadInitialized_ = true;

  return true;
}

void TFileTransport::write(const uint8_t* buf, uint32_t len) {
  if (readOnly_) {
    throw TTransportException("TFileTransport: attempting to write to file opened readonly");
  }

  enqueueEvent(buf, len);
}

template <class _T>
struct uniqueDeleter
{
  void operator()(_T *ptr) const { delete ptr; }
};

void TFileTransport::enqueueEvent(const uint8_t* buf, uint32_t eventLen) {
  // can't enqueue more events if file is going to close
  if (closing_) {
    return;
  }

  // make sure that event size is valid
  if ((maxEventSize_ > 0) && (eventLen > maxEventSize_)) {
    T_ERROR("msg size is greater than max event size: %u > %u\n", eventLen, maxEventSize_);
    return;
  }

  if (eventLen == 0) {
    T_ERROR("%s", "cannot enqueue an empty event");
    return;
  }

  std::unique_ptr<eventInfo, uniqueDeleter<eventInfo> > toEnqueue(new eventInfo());
  toEnqueue->eventBuff_ = new uint8_t[(sizeof(uint8_t) * eventLen) + 4];

  // first 4 bytes is the event length
  memcpy(toEnqueue->eventBuff_, (void*)(&eventLen), 4);
  // actual event contents
  memcpy(toEnqueue->eventBuff_ + 4, buf, eventLen);
  toEnqueue->eventSize_ = eventLen + 4;

  // lock mutex
  Guard g(mutex_);

  // make sure that enqueue buffer is initialized and writer thread is running
  if (!bufferAndThreadInitialized_) {
    if (!initBufferAndWriteThread()) {
      return;
    }
  }

  // Can't enqueue while buffer is full
  while (enqueueBuffer_->isFull()) {
    notFull_.wait();
  }

  // We shouldn't be trying to enqueue new data while a forced flush is
  // requested.  (Otherwise the writer thread might not ever be able to finish
  // the flush if more data keeps being enqueued.)
  assert(!forceFlush_);

  // add to the buffer
  eventInfo* pEvent = toEnqueue.release();
  if (!enqueueBuffer_->addEvent(pEvent)) {
    delete pEvent;
    return;
  }

  // signal anybody who's waiting for the buffer to be non-empty
  notEmpty_.notify();

  // this really should be a loop where it makes sure it got flushed
  // because condition variables can get triggered by the os for no reason
  // it is probably a non-factor for the time being
}

bool TFileTransport::swapEventBuffers(const std::chrono::time_point<std::chrono::steady_clock> *deadline) {
  bool swap;
  Guard g(mutex_);

  if (!enqueueBuffer_->isEmpty()) {
    swap = true;
  } else if (closing_) {
    // even though there is no data to write,
    // return immediately if the transport is closing
    swap = false;
  } else {
    if (deadline != nullptr) {
      // if we were handed a deadline time struct, do a timed wait
      notEmpty_.waitForTime(*deadline);
    } else {
      // just wait until the buffer gets an item
      notEmpty_.wait();
    }

    // could be empty if we timed out
    swap = enqueueBuffer_->isEmpty();
  }

  if (swap) {
    TFileTransportBuffer* temp = enqueueBuffer_;
    enqueueBuffer_ = dequeueBuffer_;
    dequeueBuffer_ = temp;
  }

  if (swap) {
    notFull_.notify();
  }

  return swap;
}

void TFileTransport::writerThread() {
  bool hasIOError = false;

  // open file if it is not open
  if (!fd_) {
    try {
      openLogFile();
    } catch (...) {
      int errno_copy = THRIFT_ERRNO;
      GlobalOutput.perror("TFileTransport: writerThread() openLogFile() ", errno_copy);
      fd_ = 0;
      hasIOError = true;
    }
  }

  // set the offset to the correct value (EOF)
  if (!hasIOError) {
    try {
      seekToEnd();
      // throw away any partial events
      offset_ += readState_.lastDispatchPtr_;
      if (0 == THRIFT_FTRUNCATE(fd_, offset_)) {
        readState_.resetAllValues();
      } else {
        int errno_copy = THRIFT_ERRNO;
        GlobalOutput.perror("TFileTransport: writerThread() truncate ", errno_copy);
        hasIOError = true;
      }
    } catch (...) {
      int errno_copy = THRIFT_ERRNO;
      GlobalOutput.perror("TFileTransport: writerThread() initialization ", errno_copy);
      hasIOError = true;
    }
  }

  // Figure out the next time by which a flush must take place
  auto ts_next_flush = getNextFlushTime();
  uint32_t unflushed = 0;

  while (1) {
    // this will only be true when the destructor is being invoked
    if (closing_) {
      if (hasIOError) {
        return;
      }

      // Try to empty buffers before exit
      if (enqueueBuffer_->isEmpty() && dequeueBuffer_->isEmpty()) {
        ::THRIFT_FSYNC(fd_);
        if (-1 == ::THRIFT_CLOSE(fd_)) {
          int errno_copy = THRIFT_ERRNO;
          GlobalOutput.perror("TFileTransport: writerThread() ::close() ", errno_copy);
        } else {
          // fd successfully closed
          fd_ = 0;
        }
        return;
      }
    }

    if (swapEventBuffers(&ts_next_flush)) {
      eventInfo* outEvent;
      while (nullptr != (outEvent = dequeueBuffer_->getNext())) {
        // Remove an event from the buffer and write it out to disk. If there is any IO error, for
        // instance,
        // the output file is unmounted or deleted, then this event is dropped. However, the writer
        // thread
        // will: (1) sleep for a short while; (2) try to reopen the file; (3) if successful then
        // start writing
        // from the end.

        while (hasIOError) {
          T_ERROR(
              "TFileTransport: writer thread going to sleep for %u microseconds due to IO errors",
              writerThreadIOErrorSleepTime_);
          THRIFT_SLEEP_USEC(writerThreadIOErrorSleepTime_);
          if (closing_) {
            return;
          }
          if (!fd_) {
            ::THRIFT_CLOSE(fd_);
            fd_ = 0;
          }
          try {
            openLogFile();
            seekToEnd();
            unflushed = 0;
            hasIOError = false;
            T_LOG_OPER(
                "TFileTransport: log file %s reopened by writer thread during error recovery",
                filename_.c_str());
          } catch (...) {
            T_ERROR("TFileTransport: unable to reopen log file %s during error recovery",
                    filename_.c_str());
          }
        }

        // sanity check on event
        if ((maxEventSize_ > 0) && (outEvent->eventSize_ > maxEventSize_)) {
          T_ERROR("msg size is greater than max event size: %u > %u\n",
                  outEvent->eventSize_,
                  maxEventSize_);
          continue;
        }

        // If chunking is required, then make sure that msg does not cross chunk boundary
        if ((outEvent->eventSize_ > 0) && (chunkSize_ != 0)) {
          // event size must be less than chunk size
          if (outEvent->eventSize_ > chunkSize_) {
            T_ERROR("TFileTransport: event size(%u) > chunk size(%u): skipping event",
                    outEvent->eventSize_,
                    chunkSize_);
            continue;
          }

          int64_t chunk1 = offset_ / chunkSize_;
          int64_t chunk2 = (offset_ + outEvent->eventSize_ - 1) / chunkSize_;

          // if adding this event will cross a chunk boundary, pad the chunk with zeros
          if (chunk1 != chunk2) {
            // refetch the offset to keep in sync
            offset_ = THRIFT_LSEEK(fd_, 0, SEEK_CUR);
            auto padding = (int32_t)((offset_ / chunkSize_ + 1) * chunkSize_ - offset_);

            auto* zeros = new uint8_t[padding];
            memset(zeros, '\0', padding);
            boost::scoped_array<uint8_t> array(zeros);
            if (-1 == ::THRIFT_WRITE(fd_, zeros, padding)) {
              int errno_copy = THRIFT_ERRNO;
              GlobalOutput.perror("TFileTransport: writerThread() error while padding zeros ",
                                  errno_copy);
              hasIOError = true;
              continue;
            }
            unflushed += padding;
            offset_ += padding;
          }
        }

        // write the dequeued event to the file
        if (outEvent->eventSize_ > 0) {
          if (-1 == ::THRIFT_WRITE(fd_, outEvent->eventBuff_, outEvent->eventSize_)) {
            int errno_copy = THRIFT_ERRNO;
            GlobalOutput.perror("TFileTransport: error while writing event ", errno_copy);
            hasIOError = true;
            continue;
          }
          unflushed += outEvent->eventSize_;
          offset_ += outEvent->eventSize_;
        }
      }
      dequeueBuffer_->reset();
    }

    if (hasIOError) {
      continue;
    }

    // Local variable to cache the state of forceFlush_.
    //
    // We only want to check the value of forceFlush_ once each time around the
    // loop.  If we check it more than once without holding the lock the entire
    // time, it could have changed state in between.  This will result in us
    // making inconsistent decisions.
    bool forced_flush = false;
    {
      Guard g(mutex_);
      if (forceFlush_) {
        if (!enqueueBuffer_->isEmpty()) {
          // If forceFlush_ is true, we need to flush all available data.
          // If enqueueBuffer_ is not empty, go back to the start of the loop to
          // write it out.
          //
          // We know the main thread is waiting on forceFlush_ to be cleared,
          // so no new events will be added to enqueueBuffer_ until we clear
          // forceFlush_.  Therefore the next time around the loop enqueueBuffer_
          // is guaranteed to be empty.  (I.e., we're guaranteed to make progress
          // and clear forceFlush_ the next time around the loop.)
          continue;
        }
        forced_flush = true;
      }
    }

    // determine if we need to perform an fsync
    bool flush = false;
    if (forced_flush || unflushed > flushMaxBytes_) {
      flush = true;
    } else {
      if (std::chrono::steady_clock::now() > ts_next_flush) {
        if (unflushed > 0) {
          flush = true;
        } else {
          // If there is no new data since the last fsync,
          // don't perform the fsync, but do reset the timer.
          ts_next_flush = getNextFlushTime();
        }
      }
    }

    if (flush) {
      // sync (force flush) file to disk
      THRIFT_FSYNC(fd_);
      unflushed = 0;
      ts_next_flush = getNextFlushTime();

      // notify anybody waiting for flush completion
      if (forced_flush) {
        Guard g(mutex_);
        forceFlush_ = false;
        assert(enqueueBuffer_->isEmpty());
        assert(dequeueBuffer_->isEmpty());
        flushed_.notifyAll();
      }
    }
  }
}

void TFileTransport::flush() {
  // file must be open for writing for any flushing to take place
  if (!writerThread_.get()) {
    return;
  }
  // wait for flush to take place
  Guard g(mutex_);

  // Indicate that we are requesting a flush
  forceFlush_ = true;
  // Wake up the writer thread so it will perform the flush immediately
  notEmpty_.notify();

  while (forceFlush_) {
    flushed_.wait();
  }
}

uint32_t TFileTransport::readAll(uint8_t* buf, uint32_t len) {
  uint32_t have = 0;
  uint32_t get = 0;

  while (have < len) {
    get = read(buf + have, len - have);
    if (get <= 0) {
      throw TEOFException();
    }
    have += get;
  }

  return have;
}

bool TFileTransport::peek() {
  // check if there is an event ready to be read
  if (!currentEvent_) {
    currentEvent_ = readEvent();
  }

  // did not manage to read an event from the file. This could have happened
  // if the timeout expired or there was some other error
  if (!currentEvent_) {
    return false;
  }

  // check if there is anything to read
  return (currentEvent_->eventSize_ - currentEvent_->eventBuffPos_) > 0;
}

uint32_t TFileTransport::read(uint8_t* buf, uint32_t len) {
  // check if there an event is ready to be read
  if (!currentEvent_) {
    currentEvent_ = readEvent();
  }

  // did not manage to read an event from the file. This could have happened
  // if the timeout expired or there was some other error
  if (!currentEvent_) {
    return 0;
  }

  // read as much of the current event as possible
  int32_t remaining = currentEvent_->eventSize_ - currentEvent_->eventBuffPos_;
  if (remaining <= (int32_t)len) {
    // copy over anything thats remaining
    if (remaining > 0) {
      memcpy(buf, currentEvent_->eventBuff_ + currentEvent_->eventBuffPos_, remaining);
    }
    delete (currentEvent_);
    currentEvent_ = nullptr;
    return remaining;
  }

  // read as much as possible
  memcpy(buf, currentEvent_->eventBuff_ + currentEvent_->eventBuffPos_, len);
  currentEvent_->eventBuffPos_ += len;
  return len;
}

// note caller is responsible for freeing returned events
eventInfo* TFileTransport::readEvent() {
  int readTries = 0;

  if (!readBuff_) {
    readBuff_ = new uint8_t[readBuffSize_];
  }

  while (1) {
    // read from the file if read buffer is exhausted
    if (readState_.bufferPtr_ == readState_.bufferLen_) {
      // advance the offset pointer
      offset_ += readState_.bufferLen_;
      readState_.bufferLen_ = static_cast<uint32_t>(::THRIFT_READ(fd_, readBuff_, readBuffSize_));
      //       if (readState_.bufferLen_) {
      //         T_DEBUG_L(1, "Amount read: %u (offset: %lu)", readState_.bufferLen_, offset_);
      //       }
      readState_.bufferPtr_ = 0;
      readState_.lastDispatchPtr_ = 0;

      // read error
      if (readState_.bufferLen_ == -1) {
        readState_.resetAllValues();
        GlobalOutput("TFileTransport: error while reading from file");
        throw TTransportException("TFileTransport: error while reading from file");
      } else if (readState_.bufferLen_ == 0) { // EOF
        // wait indefinitely if there is no timeout
        if (readTimeout_ == TAIL_READ_TIMEOUT) {
          THRIFT_SLEEP_USEC(eofSleepTime_);
          continue;
        } else if (readTimeout_ == NO_TAIL_READ_TIMEOUT) {
          // reset state
          readState_.resetState(0);
          return nullptr;
        } else if (readTimeout_ > 0) {
          // timeout already expired once
          if (readTries > 0) {
            readState_.resetState(0);
            return nullptr;
          } else {
            THRIFT_SLEEP_USEC(readTimeout_ * 1000);
            readTries++;
            continue;
          }
        }
      }
    }

    readTries = 0;

    // attempt to read an event from the buffer
    while (readState_.bufferPtr_ < readState_.bufferLen_) {
      if (readState_.readingSize_) {
        if (readState_.eventSizeBuffPos_ == 0) {
          if ((offset_ + readState_.bufferPtr_) / chunkSize_
              != ((offset_ + readState_.bufferPtr_ + 3) / chunkSize_)) {
            // skip one byte towards chunk boundary
            //            T_DEBUG_L(1, "Skipping a byte");
            readState_.bufferPtr_++;
            continue;
          }
        }

        readState_.eventSizeBuff_[readState_.eventSizeBuffPos_++]
            = readBuff_[readState_.bufferPtr_++];

        if (readState_.eventSizeBuffPos_ == 4) {
          if (readState_.getEventSize() == 0) {
            // 0 length event indicates padding
            //            T_DEBUG_L(1, "Got padding");
            readState_.resetState(readState_.lastDispatchPtr_);
            continue;
          }
          // got a valid event
          readState_.readingSize_ = false;
          if (readState_.event_) {
            delete (readState_.event_);
          }
          readState_.event_ = new eventInfo();
          readState_.event_->eventSize_ = readState_.getEventSize();

          // check if the event is corrupted and perform recovery if required
          if (isEventCorrupted()) {
            performRecovery();
            // start from the top
            break;
          }
        }
      } else {
        if (!readState_.event_->eventBuff_) {
          readState_.event_->eventBuff_ = new uint8_t[readState_.event_->eventSize_];
          readState_.event_->eventBuffPos_ = 0;
        }
        // take either the entire event or the remaining bytes in the buffer
        int reclaimBuffer = (std::min)((uint32_t)(readState_.bufferLen_ - readState_.bufferPtr_),
                                       readState_.event_->eventSize_ - readState_.event_->eventBuffPos_);

        // copy data from read buffer into event buffer
        memcpy(readState_.event_->eventBuff_ + readState_.event_->eventBuffPos_,
               readBuff_ + readState_.bufferPtr_,
               reclaimBuffer);

        // increment position ptrs
        readState_.event_->eventBuffPos_ += reclaimBuffer;
        readState_.bufferPtr_ += reclaimBuffer;

        // check if the event has been read in full
        if (readState_.event_->eventBuffPos_ == readState_.event_->eventSize_) {
          // set the completed event to the current event
          eventInfo* completeEvent = readState_.event_;
          completeEvent->eventBuffPos_ = 0;

          readState_.event_ = nullptr;
          readState_.resetState(readState_.bufferPtr_);

          // exit criteria
          return completeEvent;
        }
      }
    }
  }
}

bool TFileTransport::isEventCorrupted() {
  // an error is triggered if:
  if ((maxEventSize_ > 0) && (readState_.event_->eventSize_ > maxEventSize_)) {
    // 1. Event size is larger than user-speficied max-event size
    T_ERROR("Read corrupt event. Event size(%u) greater than max event size (%u)",
            readState_.event_->eventSize_,
            maxEventSize_);
    return true;
  } else if (readState_.event_->eventSize_ > chunkSize_) {
    // 2. Event size is larger than chunk size
    T_ERROR("Read corrupt event. Event size(%u) greater than chunk size (%u)",
            readState_.event_->eventSize_,
            chunkSize_);
    return true;
  } else if (((offset_ + readState_.bufferPtr_ - 4) / chunkSize_)
             != ((offset_ + readState_.bufferPtr_ + readState_.event_->eventSize_ - 1)
                 / chunkSize_)) {
    // 3. size indicates that event crosses chunk boundary
    T_ERROR("Read corrupt event. Event crosses chunk boundary. Event size:%u  Offset:%lu",
            readState_.event_->eventSize_,
            static_cast<unsigned long>(offset_ + readState_.bufferPtr_ + 4));

    return true;
  }

  return false;
}

void TFileTransport::performRecovery() {
  // perform some kickass recovery
  uint32_t curChunk = getCurChunk();
  if (lastBadChunk_ == curChunk) {
    numCorruptedEventsInChunk_++;
  } else {
    lastBadChunk_ = curChunk;
    numCorruptedEventsInChunk_ = 1;
  }

  if (numCorruptedEventsInChunk_ < maxCorruptedEvents_) {
    // maybe there was an error in reading the file from disk
    // seek to the beginning of chunk and try again
    seekToChunk(curChunk);
  } else {

    // just skip ahead to the next chunk if we not already at the last chunk
    if (curChunk != (getNumChunks() - 1)) {
      seekToChunk(curChunk + 1);
    } else if (readTimeout_ == TAIL_READ_TIMEOUT) {
      // if tailing the file, wait until there is enough data to start
      // the next chunk
      while (curChunk == (getNumChunks() - 1)) {
        THRIFT_SLEEP_USEC(corruptedEventSleepTime_);
      }
      seekToChunk(curChunk + 1);
    } else {
      // pretty hosed at this stage, rewind the file back to the last successful
      // point and punt on the error
      readState_.resetState(readState_.lastDispatchPtr_);
      currentEvent_ = nullptr;
      char errorMsg[1024];
      sprintf(errorMsg,
              "TFileTransport: log file corrupted at offset: %lu",
              static_cast<unsigned long>(offset_ + readState_.lastDispatchPtr_));

      GlobalOutput(errorMsg);
      throw TTransportException(errorMsg);
    }
  }
}

void TFileTransport::seekToChunk(int32_t chunk) {
  if (fd_ <= 0) {
    throw TTransportException("File not open");
  }

  int32_t numChunks = getNumChunks();

  // file is empty, seeking to chunk is pointless
  if (numChunks == 0) {
    return;
  }

  // negative indicates reverse seek (from the end)
  if (chunk < 0) {
    chunk += numChunks;
  }

  // too large a value for reverse seek, just seek to beginning
  if (chunk < 0) {
    T_DEBUG("%s", "Incorrect value for reverse seek. Seeking to beginning...");
    chunk = 0;
  }

  // cannot seek past EOF
  bool seekToEnd = false;
  off_t minEndOffset = 0;
  if (chunk >= numChunks) {
    T_DEBUG("%s", "Trying to seek past EOF. Seeking to EOF instead...");
    seekToEnd = true;
    chunk = numChunks - 1;
    // this is the min offset to process events till
    minEndOffset = ::THRIFT_LSEEK(fd_, 0, SEEK_END);
  }

  off_t newOffset = off_t(chunk) * chunkSize_;
  offset_ = ::THRIFT_LSEEK(fd_, newOffset, SEEK_SET);
  readState_.resetAllValues();
  currentEvent_ = nullptr;
  if (offset_ == -1) {
    GlobalOutput("TFileTransport: lseek error in seekToChunk");
    throw TTransportException("TFileTransport: lseek error in seekToChunk");
  }

  // seek to EOF if user wanted to go to last chunk
  if (seekToEnd) {
    uint32_t oldReadTimeout = getReadTimeout();
    setReadTimeout(NO_TAIL_READ_TIMEOUT);
    // keep on reading unti the last event at point of seekChunk call
    shared_ptr<eventInfo> event;
    while ((offset_ + readState_.bufferPtr_) < minEndOffset) {
      event.reset(readEvent());
      if (event.get() == nullptr) {
        break;
      }
    }
    setReadTimeout(oldReadTimeout);
  }
}

void TFileTransport::seekToEnd() {
  seekToChunk(getNumChunks());
}

uint32_t TFileTransport::getNumChunks() {
  if (fd_ <= 0) {
    return 0;
  }

  struct THRIFT_STAT f_info;
  int rv = ::THRIFT_FSTAT(fd_, &f_info);

  if (rv < 0) {
    int errno_copy = THRIFT_ERRNO;
    throw TTransportException(TTransportException::UNKNOWN,
                              "TFileTransport::getNumChunks() (fstat)",
                              errno_copy);
  }

  if (f_info.st_size > 0) {
    size_t numChunks = ((f_info.st_size) / chunkSize_) + 1;
    if (numChunks > (std::numeric_limits<uint32_t>::max)())
      throw TTransportException("Too many chunks");
    return static_cast<uint32_t>(numChunks);
  }

  // empty file has no chunks
  return 0;
}

uint32_t TFileTransport::getCurChunk() {
  return static_cast<uint32_t>(offset_ / chunkSize_);
}

// Utility Functions
void TFileTransport::openLogFile() {
#ifndef _WIN32
  mode_t mode = readOnly_ ? S_IRUSR | S_IRGRP | S_IROTH : S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH;
  int flags = readOnly_ ? O_RDONLY : O_RDWR | O_CREAT | O_APPEND;
#else
  int mode = readOnly_ ? _S_IREAD : _S_IREAD | _S_IWRITE;
  int flags = readOnly_ ? _O_RDONLY : _O_RDWR | _O_CREAT | _O_APPEND;
#endif
  fd_ = ::THRIFT_OPEN(filename_.c_str(), flags, mode);
  offset_ = 0;

  // make sure open call was successful
  if (fd_ == -1) {
    int errno_copy = THRIFT_ERRNO;
    GlobalOutput.perror("TFileTransport: openLogFile() ::open() file: " + filename_, errno_copy);
    throw TTransportException(TTransportException::NOT_OPEN, filename_, errno_copy);
  }
}

std::chrono::time_point<std::chrono::steady_clock> TFileTransport::getNextFlushTime() {
  return std::chrono::steady_clock::now() + std::chrono::microseconds(flushMaxUs_);
}

TFileTransportBuffer::TFileTransportBuffer(uint32_t size)
  : bufferMode_(WRITE), writePoint_(0), readPoint_(0), size_(size) {
  buffer_ = new eventInfo* [size];
}

TFileTransportBuffer::~TFileTransportBuffer() {
  if (buffer_) {
    for (uint32_t i = 0; i < writePoint_; i++) {
      delete buffer_[i];
    }
    delete[] buffer_;
    buffer_ = nullptr;
  }
}

bool TFileTransportBuffer::addEvent(eventInfo* event) {
  if (bufferMode_ == READ) {
    GlobalOutput("Trying to write to a buffer in read mode");
  }
  if (writePoint_ < size_) {
    buffer_[writePoint_++] = event;
    return true;
  } else {
    // buffer is full
    return false;
  }
}

eventInfo* TFileTransportBuffer::getNext() {
  if (bufferMode_ == WRITE) {
    bufferMode_ = READ;
  }
  if (readPoint_ < writePoint_) {
    return buffer_[readPoint_++];
  } else {
    // no more entries
    return nullptr;
  }
}

void TFileTransportBuffer::reset() {
  if (bufferMode_ == WRITE || writePoint_ > readPoint_) {
    T_DEBUG("%s", "Resetting a buffer with unread entries");
  }
  // Clean up the old entries
  for (uint32_t i = 0; i < writePoint_; i++) {
    delete buffer_[i];
  }
  bufferMode_ = WRITE;
  writePoint_ = 0;
  readPoint_ = 0;
}

bool TFileTransportBuffer::isFull() {
  return writePoint_ == size_;
}

bool TFileTransportBuffer::isEmpty() {
  return writePoint_ == 0;
}

TFileProcessor::TFileProcessor(shared_ptr<TProcessor> processor,
                               shared_ptr<TProtocolFactory> protocolFactory,
                               shared_ptr<TFileReaderTransport> inputTransport)
  : processor_(processor),
    inputProtocolFactory_(protocolFactory),
    outputProtocolFactory_(protocolFactory),
    inputTransport_(inputTransport) {

  // default the output transport to a null transport (common case)
  outputTransport_ = std::make_shared<TNullTransport>();
}

TFileProcessor::TFileProcessor(shared_ptr<TProcessor> processor,
                               shared_ptr<TProtocolFactory> inputProtocolFactory,
                               shared_ptr<TProtocolFactory> outputProtocolFactory,
                               shared_ptr<TFileReaderTransport> inputTransport)
  : processor_(processor),
    inputProtocolFactory_(inputProtocolFactory),
    outputProtocolFactory_(outputProtocolFactory),
    inputTransport_(inputTransport) {

  // default the output transport to a null transport (common case)
  outputTransport_ = std::make_shared<TNullTransport>();
}

TFileProcessor::TFileProcessor(shared_ptr<TProcessor> processor,
                               shared_ptr<TProtocolFactory> protocolFactory,
                               shared_ptr<TFileReaderTransport> inputTransport,
                               shared_ptr<TTransport> outputTransport)
  : processor_(processor),
    inputProtocolFactory_(protocolFactory),
    outputProtocolFactory_(protocolFactory),
    inputTransport_(inputTransport),
    outputTransport_(outputTransport) {
}

void TFileProcessor::process(uint32_t numEvents, bool tail) {
  shared_ptr<TProtocol> inputProtocol = inputProtocolFactory_->getProtocol(inputTransport_);
  shared_ptr<TProtocol> outputProtocol = outputProtocolFactory_->getProtocol(outputTransport_);

  // set the read timeout to 0 if tailing is required
  int32_t oldReadTimeout = inputTransport_->getReadTimeout();
  if (tail) {
    // save old read timeout so it can be restored
    inputTransport_->setReadTimeout(TFileTransport::TAIL_READ_TIMEOUT);
  }

  uint32_t numProcessed = 0;
  while (1) {
    // bad form to use exceptions for flow control but there is really
    // no other way around it
    try {
      processor_->process(inputProtocol, outputProtocol, nullptr);
      numProcessed++;
      if ((numEvents > 0) && (numProcessed == numEvents)) {
        return;
      }
    } catch (TEOFException&) {
      if (!tail) {
        break;
      }
    } catch (TException& te) {
      cerr << te.what() << endl;
      break;
    }
  }

  // restore old read timeout
  if (tail) {
    inputTransport_->setReadTimeout(oldReadTimeout);
  }
}

void TFileProcessor::processChunk() {
  shared_ptr<TProtocol> inputProtocol = inputProtocolFactory_->getProtocol(inputTransport_);
  shared_ptr<TProtocol> outputProtocol = outputProtocolFactory_->getProtocol(outputTransport_);

  uint32_t curChunk = inputTransport_->getCurChunk();

  while (1) {
    // bad form to use exceptions for flow control but there is really
    // no other way around it
    try {
      processor_->process(inputProtocol, outputProtocol, nullptr);
      if (curChunk != inputTransport_->getCurChunk()) {
        break;
      }
    } catch (TEOFException&) {
      break;
    } catch (TException& te) {
      cerr << te.what() << endl;
      break;
    }
  }
}
}
}
} // apache::thrift::transport