adamdebreceni commented on a change in pull request #887:
URL: https://github.com/apache/nifi-minifi-cpp/pull/887#discussion_r481953824
##########
File path: extensions/rocksdb-repos/DatabaseContentRepository.cpp
##########
@@ -64,13 +65,53 @@ void DatabaseContentRepository::stop() {
db_.reset();
}
+DatabaseContentRepository::Session::Session(std::shared_ptr<ContentRepository>
repository) : ContentSession(std::move(repository)) {}
+
+std::shared_ptr<ContentSession> DatabaseContentRepository::createSession() {
+ return std::make_shared<Session>(shared_from_this());
+}
+
+void DatabaseContentRepository::Session::commit() {
+ auto dbContentRepository =
std::static_pointer_cast<DatabaseContentRepository>(repository_);
+ auto opendb = dbContentRepository->db_->open();
+ if (!opendb) {
+ throw Exception(GENERAL_EXCEPTION, "Couldn't open rocksdb database to
commit content changes");
+ }
+ rocksdb::WriteBatch batch;
+ for (const auto& resource : managedResources_) {
+ auto outStream = dbContentRepository->write(resource.first, false, &batch);
+ if (outStream == nullptr) {
+ throw Exception(GENERAL_EXCEPTION, "Couldn't open the underlying
resource for write: " + resource.first->getContentFullPath());
+ }
+ const auto size = resource.second->getSize();
+ if (outStream->write(const_cast<uint8_t*>(resource.second->getBuffer()),
size) != size) {
+ throw Exception(GENERAL_EXCEPTION, "Failed to write new resource: " +
resource.first->getContentFullPath());
+ }
+ }
+ for (const auto& resource : extendedResources_) {
+ auto outStream = dbContentRepository->write(resource.first, true, &batch);
+ if (outStream == nullptr) {
+ throw Exception(GENERAL_EXCEPTION, "Couldn't open the underlying
resource for append: " + resource.first->getContentFullPath());
+ }
+ const auto size = resource.second->getSize();
+ if (outStream->write(const_cast<uint8_t*>(resource.second->getBuffer()),
size) != size) {
+ throw Exception(GENERAL_EXCEPTION, "Failed to append to resource: " +
resource.first->getContentFullPath());
+ }
+ }
+
+ rocksdb::WriteOptions options;
+ options.sync = true;
+ rocksdb::Status status = opendb->Write(options, &batch);
+ if (!status.ok()) {
+ throw std::runtime_error("Batch write failed: " + status.ToString());
Review comment:
changed it to one
----------------------------------------------------------------
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:
[email protected]
