file_manager_strategy.h

Go to the documentation of this file.

/*
 * Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License").
 * You may not use this file except in compliance with the License.
 * A copy of the License is located at
 *
 *  http://aws.amazon.com/apache2.0
 *
 * or in the "license" file accompanying this file. This file 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.
 */

#pragma once

#include <iostream>
#include <fstream>
#include <list>
#include <unordered_map>
#include <set>
#include <memory>
#include <random>
#include <mutex>
#include <experimental/filesystem>
#include <dataflow_lite/utils/service.h>
#include <file_management/file_manager_options.h>
#include <aws/core/utils/json/JsonSerializer.h>

namespace Aws {
namespace FileManagement {

enum TokenStatus {
  ACTIVE,
  INACTIVE
};

using DataToken = uint64_t;

static constexpr const char* kPositionKey = "position";
static constexpr const char* kEofKey = "eof";
static constexpr const char* kFilePathKey = "file_path";

void sanitizePath(std::string & path);

class FileTokenInfo {
public:

  FileTokenInfo() = default;

  // NOLINTNEXTLINE(google-runtime-int)
  explicit FileTokenInfo(const std::string &file_path, const long position, const bool eof) :
  file_path_{file_path},
  position_(position),
  eof_(eof)
  {

  };

  // NOLINTNEXTLINE(google-runtime-int)
  explicit FileTokenInfo(std::string &&file_path, const long position, const bool eof) :
      file_path_{std::move(file_path)},
      position_(position),
      eof_(eof)
  {

  };

  FileTokenInfo(const FileTokenInfo &info) = default;

  FileTokenInfo & operator=(const FileTokenInfo & other) = default;

  ~FileTokenInfo() = default;

  std::string serialize() const {
    Aws::Utils::Json::JsonValue json_value;
    const Aws::String file_path(file_path_.c_str());
    json_value
        .WithInt64(kPositionKey, position_)
        .WithBool(kEofKey, eof_)
        .WithString(kFilePathKey, file_path);
    return std::string(json_value.View().WriteCompact().c_str());
  }

  void deserialize(const std::string& token_info_json) {
    const Aws::String aws_str(token_info_json.c_str());
    const Aws::Utils::Json::JsonValue json_value(aws_str);
    if (!json_value.WasParseSuccessful()) {
      throw std::runtime_error("Unable to parse JSON");
    }
    auto view = json_value.View();
    position_ = view.GetInt64(kPositionKey);
    eof_ = view.GetBool(kEofKey);
    file_path_ = view.GetString(kFilePathKey).c_str();
  }

  std::string file_path_;

  int64_t position_ = 0;

  bool eof_{};
};

inline bool operator==(const FileTokenInfo& lhs, const FileTokenInfo& rhs){
  return lhs.eof_ == rhs.eof_ && lhs.position_ == rhs.position_ && lhs.file_path_ == rhs.file_path_;
}

inline bool operator!=(const FileTokenInfo& lhs, const FileTokenInfo& rhs){ return !(lhs == rhs); }

class DataManagerStrategy : public Service {
public:
  DataManagerStrategy() = default;
  ~DataManagerStrategy() override = default;

  virtual bool isDataAvailable() = 0;

  virtual bool isDeleteStaleData() = 0;

  virtual DataToken read(std::string &data) = 0;

  virtual void write(const std::string &data) = 0;

  virtual void resolve(const DataToken &token, bool is_success) = 0;
};

class TokenStore {
public:

  TokenStore() = default;

  explicit TokenStore(const TokenStoreOptions &options);

  bool isTokenAvailable(const std::string &file_name) const;

  FileTokenInfo popAvailableToken(const std::string &file_name);

  // NOLINTNEXTLINE(google-runtime-int)
  DataToken createToken(const std::string &file_name, const long streampos, bool is_eof);

  FileTokenInfo fail(const DataToken &token);

  FileTokenInfo resolve(const DataToken &token);

  std::vector<FileTokenInfo> backup();

  void backupToDisk();

  void restore(const std::vector<FileTokenInfo> &file_tokens);

  void restoreFromDisk();



private:
  void validateOptions();
  
  void initializeBackupDirectory();

  std::unordered_map<DataToken, FileTokenInfo> token_store_;

  std::unordered_map<std::string, std::list<DataToken>> file_tokens_;

  std::unordered_map<std::string, FileTokenInfo> staged_tokens_;

  TokenStoreOptions options_;

  std::random_device rand_device;
};

class FileManagerStrategy : public DataManagerStrategy {
public:
  explicit FileManagerStrategy(const FileManagerStrategyOptions &options);

  ~FileManagerStrategy() override = default;

  bool start() override;

  bool isDataAvailable() override;

  bool isDeleteStaleData() override;

  DataToken read(std::string &data) override;

  void write(const std::string &data) override;

  void resolve(const DataToken &token, bool is_success) override;

  bool shutdown() override;

protected: // functions exposed for testing
  std::string getFileToRead();

  std::string getActiveWriteFile() {
    return active_write_file_;
  }

private:
  void validateOptions();

  void initializeStorage();

  void initializeTokenStore();

  void discoverStoredFiles();

  void deleteFile(const std::string &file_path);

  void checkIfWriteFileShouldRotate(const uintmax_t &new_data_size);

  void rotateWriteFile();

  void checkIfStorageLimitHasBeenReached(const uintmax_t &new_data_size);

  void deleteOldestFile();

  void addFilePathToStorage(const std::experimental::filesystem::path &file_path);

  std::list<std::string> stored_files_;

  std::atomic<size_t> stored_files_size_{};

  std::string active_write_file_;

  std::atomic<size_t> active_write_file_size_{};

  std::mutex active_write_file_mutex_;

  std::string active_read_file_;

  std::unique_ptr<std::ifstream> active_read_file_stream_;

  std::mutex active_read_file_mutex_;

  FileManagerStrategyOptions options_;

  std::unique_ptr<TokenStore> token_store_;
};

}  // namespace FileManagement
}  // namespace Aws