arpadboda commented on a change in pull request #605: MINIFICPP-550 - Implement 
RocksDB controller service and component st…
URL: https://github.com/apache/nifi-minifi-cpp/pull/605#discussion_r393646027
 
 

 ##########
 File path: extensions/sftp/processors/ListSFTP.cpp
 ##########
 @@ -760,140 +713,83 @@ void ListSFTP::listByTrackingTimestamps(
   }
 }
 
-bool ListSFTP::persistTrackingEntitiesCache(const std::string& hostname, const 
std::string& username, const std::string& remote_path) {
-  std::ofstream file(tracking_entities_state_filename_);
-  if (!file.is_open()) {
-    logger_->log_error("Failed to store Tracking Entities state to state file 
\"%s\"", tracking_entities_state_filename_.c_str());
-    return false;
-  }
-  file << "hostname=" << hostname << "\n";
-  file << "username=" << username << "\n";
-  file << "remote_path=" << remote_path << "\n";
-  file << "json_state_file=" << tracking_entities_state_json_filename_ << "\n";
-  file.close();
-
-  std::ofstream json_file(tracking_entities_state_json_filename_);
-  if (!json_file.is_open()) {
-    logger_->log_error("Failed to store Tracking Entities state to state json 
file \"%s\"", tracking_entities_state_json_filename_.c_str());
-    return false;
-  }
-
-  rapidjson::Document entities(rapidjson::kObjectType);
-  rapidjson::Document::AllocatorType& alloc = entities.GetAllocator();
-  for (const auto& already_listed_entity : already_listed_entities_) {
-    rapidjson::Value entity(rapidjson::kObjectType);
-    entity.AddMember("timestamp", already_listed_entity.second.timestamp, 
alloc);
-    entity.AddMember("size", already_listed_entity.second.size, alloc);
-    entities.AddMember(rapidjson::Value(already_listed_entity.first.c_str(), 
alloc), std::move(entity), alloc);
+bool ListSFTP::persistTrackingEntitiesCache(const 
std::shared_ptr<core::ProcessContext>& context, const std::string& hostname, 
const std::string& username, const std::string& remote_path) {
+  std::unordered_map<std::string, std::string> state;
+  state["listing_strategy"] = listing_strategy_;
+  state["hostname"] = hostname;
+  state["username"] = username;
+  state["remote_path"] = remote_path;
+  size_t i = 0;
+  for (const auto &already_listed_entity : already_listed_entities_) {
+    state["entity." + std::to_string(i) + ".name"] = 
already_listed_entity.first;
+    state["entity." + std::to_string(i) + ".timestamp"] = 
std::to_string(already_listed_entity.second.timestamp);
+    state["entity." + std::to_string(i) + ".size"] = 
std::to_string(already_listed_entity.second.size);
+    ++i;
   }
-
-  rapidjson::OStreamWrapper osw(json_file);
-  rapidjson::Writer<rapidjson::OStreamWrapper> writer(osw);
-  entities.Accept(writer);
-
-  return true;
+  return state_manager_->set(state);
 }
 
-bool ListSFTP::updateFromTrackingEntitiesCache(const std::string& hostname, 
const std::string& username, const std::string& remote_path) {
-  std::ifstream file(tracking_entities_state_filename_);
-  if (!file.is_open()) {
-    logger_->log_error("Failed to open Tracking Entities state file \"%s\"", 
tracking_entities_state_filename_.c_str());
-    return false;
-  }
+bool ListSFTP::updateFromTrackingEntitiesCache(const 
std::shared_ptr<core::ProcessContext>& context, const std::string& hostname, 
const std::string& username, const std::string& remote_path) {
+  std::string state_listing_strategy;
   std::string state_hostname;
   std::string state_username;
   std::string state_remote_path;
-  std::string state_json_state_file;
-
-  std::string line;
-  while (std::getline(file, line)) {
-    size_t separator_pos = line.find('=');
-    if (separator_pos == std::string::npos) {
-      logger_->log_warn("None key-value line found in Tracking Entities state 
file \"%s\": \"%s\"", tracking_entities_state_filename_.c_str(), line.c_str());
-      continue;
-    }
-    std::string key = line.substr(0, separator_pos);
-    std::string value = line.substr(separator_pos + 1);
-    if (key == "hostname") {
-      state_hostname = std::move(value);
-    } else if (key == "username") {
-      state_username = std::move(value);
-    } else if (key == "remote_path") {
-      state_remote_path = std::move(value);
-    } else if (key == "json_state_file") {
-      state_json_state_file = std::move(value);
-    } else {
-      logger_->log_warn("Unknown key found in Tracking Entities state file 
\"%s\": \"%s\"", tracking_entities_state_filename_.c_str(), key.c_str());
-    }
-  }
-  file.close();
-
-  if (state_hostname != hostname ||
-      state_username != username ||
-      state_remote_path != remote_path) {
-    logger_->log_error("Tracking Entities state file \"%s\" was created with 
different settings than the current ones, ignoring. "
-                       "Hostname: \"%s\" vs. \"%s\", "
-                       "Username: \"%s\" vs. \"%s\", "
-                       "Remote Path: \"%s\" vs. \"%s\"",
-                       tracking_entities_state_filename_.c_str(),
-                       state_hostname, hostname,
-                       state_username, username,
-                       state_remote_path, remote_path);
-    return false;
-  }
+  std::unordered_map<std::string, ListedEntity> new_already_listed_entities;
 
-  if (state_json_state_file.empty()) {
-    logger_->log_error("Could not found json state file path in Tracking 
Entities state file \"%s\"", tracking_entities_state_filename_.c_str());
+  std::unordered_map<std::string, std::string> state_map;
+  if (!state_manager_->get(state_map)) {
+    logger_->log_debug("Failed to get state from StateManager");
 
 Review comment:
   That's completely reasonable, I put my focus on the error case: for eg. 
persistency is corrupted/deleted on the disk, so processors unexpectedly start 
from a clean state. Which worth at least an info level log message in my 
opinion. 
   Yes, I also expect the state manager to log something in such scenario, 
although loading persistency shouldn't happen too frequently (onschedule 
phase), so I don't mind being quite verbose here. 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

Reply via email to