utils/include/core/Repository.h (98 lines of code) (raw):
/**
* @file Repository
* Repository class declaration
*
* 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.
*/
#pragma once
#include <atomic>
#include <cstdint>
#include <cstring>
#include <map>
#include <memory>
#include <set>
#include <string>
#include <string_view>
#include <utility>
#include <vector>
#include "core/Connectable.h"
#include "core/ContentRepository.h"
#include "core/Property.h"
#include "core/logging/LoggerFactory.h"
#include "core/RepositoryMetricsSource.h"
#include "utils/BackTrace.h"
#include "utils/Literals.h"
#include "utils/StringUtils.h"
#include "utils/TimeUtil.h"
#include "core/Core.h"
#include "minifi-cpp/core/Repository.h"
#ifndef WIN32
#include <sys/stat.h>
#endif
namespace org::apache::nifi::minifi::core {
constexpr auto REPOSITORY_DIRECTORY = "./repo";
constexpr auto MAX_REPOSITORY_STORAGE_SIZE = 10_MiB;
constexpr auto MAX_REPOSITORY_ENTRY_LIFE_TIME = std::chrono::minutes(10);
constexpr auto REPOSITORY_PURGE_PERIOD = std::chrono::milliseconds(2500);
class RepositoryImpl : public core::CoreComponentImpl, public core::RepositoryMetricsSourceImpl, public virtual Repository {
public:
explicit RepositoryImpl(std::string_view repo_name = "Repository",
std::string directory = REPOSITORY_DIRECTORY,
std::chrono::milliseconds maxPartitionMillis = MAX_REPOSITORY_ENTRY_LIFE_TIME,
int64_t maxPartitionBytes = MAX_REPOSITORY_STORAGE_SIZE,
std::chrono::milliseconds purgePeriod = REPOSITORY_PURGE_PERIOD)
: core::CoreComponentImpl(repo_name),
max_partition_millis_(maxPartitionMillis),
max_partition_bytes_(maxPartitionBytes),
purge_period_(purgePeriod),
repo_full_(false),
directory_(std::move(directory)),
logger_(logging::LoggerFactory<Repository>::getLogger()) {
}
bool isNoop() const override {
return true;
}
void flush() override {
}
bool Put(const std::string& /*key*/, const uint8_t* /*buf*/, size_t /*bufLen*/) override {
return true;
}
bool MultiPut(const std::vector<std::pair<std::string, std::unique_ptr<io::BufferStream>>>& /*data*/) override {
return true;
}
bool Delete(const std::string& /*key*/) override {
return true;
}
bool Delete(const std::shared_ptr<core::CoreComponent>& item) override {
return Delete(item->getUUIDStr());
}
bool Delete(std::vector<std::shared_ptr<core::SerializableComponent>> &storedValues) override;
void setConnectionMap(std::map<std::string, core::Connectable*> connectionMap) override {
connection_map_ = std::move(connectionMap);
}
void setContainers(std::map<std::string, core::Connectable*> containers) override {
containers_ = std::move(containers);
}
bool Get(const std::string& /*key*/, std::string& /*value*/) override {
return false;
}
bool getElements(std::vector<std::shared_ptr<core::SerializableComponent>>& /*store*/, size_t& /*max_size*/) override {
return true;
}
bool storeElement(const std::shared_ptr<core::SerializableComponent>& element) override;
void loadComponent(const std::shared_ptr<core::ContentRepository>& /*content_repo*/) override {
}
std::string getDirectory() const override {
return directory_;
}
std::string getRepositoryName() const override {
return getName();
}
RepositoryImpl(const RepositoryImpl &parent) = delete;
RepositoryImpl &operator=(const RepositoryImpl &parent) = delete;
protected:
std::map<std::string, core::Connectable*> containers_;
std::map<std::string, core::Connectable*> connection_map_;
// max db entry lifetime
std::chrono::milliseconds max_partition_millis_;
// max db size
int64_t max_partition_bytes_;
std::chrono::milliseconds purge_period_;
std::atomic<bool> repo_full_;
std::string directory_;
private:
std::shared_ptr<logging::Logger> logger_;
};
} // namespace org::apache::nifi::minifi::core