text: // Copyright (c) 2019, FlightAware LLC.
  // All rights reserved.
  // Licensed under the 2-clause BSD license; see the LICENSE file
  
  #include "sample_source.h"
  
  #include <chrono>
  #include <iostream>
  
  using namespace airnav::uat;
  
  void FileSampleSource::Start() {
      stream_.open(path_.native());
      if (!stream_.good()) {
          auto ec = boost::system::error_code(errno, boost::system::system_category());
          stream_.close();
          DispatchError(ec);
          return;
      }
  
      next_block_ = std::chrono::steady_clock::now();
      timestamp_ = 1; // always use synthetic timestamps for file sources
  
      auto self = std::static_pointer_cast<FileSampleSource>(shared_from_this());
      service_.post(std::bind(&FileSampleSource::ReadBlock, self, boost::system::error_code()));
  }
  
  void FileSampleSource::Stop() {
      timer_.cancel();
      if (stream_.is_open()) {
          stream_.close();
          DispatchError(boost::asio::error::eof);
      }
  }
  
  void FileSampleSource::ReadBlock(const boost::system::error_code &ec) {
      if (ec) {
          if (ec == boost::asio::error::operation_aborted) {
              return;
          }
  
          stream_.close();
          DispatchError(ec);
          return;
      }
  
      if (!stream_.is_open()) {
          return;
      }
  
      block_.resize(block_.capacity());
      stream_.read(reinterpret_cast<char *>(block_.data()), block_.size());
  
      if (stream_.bad()) {
          auto ec = boost::system::error_code(errno, boost::system::system_category());
          stream_.close();
          DispatchError(ec);
          return;
      }
  
      block_.resize(stream_.gcount() - (stream_.gcount() % alignment_));
      if (!block_.empty()) {
          DispatchBuffer(timestamp_, block_);
          timestamp_ += (block_.size() * 1000ULL / bytes_per_second_);
      }
  
      if (stream_.eof()) {
          stream_.close();
          DispatchError(boost::asio::error::eof);
          return;
      }
  
      auto self = std::static_pointer_cast<FileSampleSource>(shared_from_this());
      if (throttle_) {
          auto delay = std::chrono::nanoseconds(1000000000ULL * block_.size() / bytes_per_second_);
          next_block_ += delay;
          timer_.expires_at(next_block_);
          timer_.async_wait(std::bind(&FileSampleSource::ReadBlock, self, std::placeholders::_1));
      } else {
          service_.post(std::bind(&FileSampleSource::ReadBlock, self, boost::system::error_code()));
      }
  }
  
  //
  //
  //
  
  void StdinSampleSource::Start() {
      stream_.assign(::dup(STDIN_FILENO));
      ScheduleRead();
  }
  
  void StdinSampleSource::Stop() {
      if (stream_.is_open()) {
          stream_.close();
          DispatchError(boost::asio::error::eof);
      }
  }
  
  void StdinSampleSource::ScheduleRead() {
      if (!stream_.is_open()) {
          return;
      }
  
      block_.resize(block_.capacity());
  
      auto self = shared_from_this();
      stream_.async_read_some(boost::asio::buffer(block_.data() + used_, block_.size() - used_), [this, self](const boost::system::error_code &ec, std::size_t bytes_transferred) {
          if (ec) {
              if (ec == boost::asio::error::operation_aborted) {
                  return;
              }
  
              stream_.close();
              DispatchError(ec);
              return;
          }
  
          used_ += bytes_transferred;
  
          // work out a starting timestamp
          static auto unix_epoch = std::chrono::system_clock::from_time_t(0);
          auto end_of_block = std::chrono::system_clock::now();
          auto start_of_block = end_of_block - (std::chrono::milliseconds(1000) * bytes_transferred / samples_per_second_ / alignment_);
          std::uint64_t timestamp = std::chrono::duration_cast<std::chrono::milliseconds>(start_of_block - unix_epoch).count();
  
          // fixme, don't copy!
          auto trailing_bytes = used_ % alignment_;
          auto leading_bytes = used_ - trailing_bytes;
  
          Bytes buffer;
          buffer.resize(leading_bytes);
          std::copy(block_.begin(), block_.begin() + leading_bytes, buffer.begin());
          std::copy(block_.begin() + leading_bytes, block_.end(), block_.begin());
          used_ = trailing_bytes;
          DispatchBuffer(timestamp, buffer);
          ScheduleRead();
      });
  }
  