| 76 | } |
| 77 | |
| 78 | void DatabaseContentRepository::Session::commit() { |
| 79 | auto dbContentRepository = std::static_pointer_cast<DatabaseContentRepository>(repository_); |
| 80 | auto opendb = dbContentRepository->db_->open(); |
| 81 | if (!opendb) { |
| 82 | throw Exception(REPOSITORY_EXCEPTION, "Couldn't open rocksdb database to commit content changes"); |
| 83 | } |
| 84 | rocksdb::WriteBatch batch; |
| 85 | for (const auto& resource : managedResources_) { |
| 86 | auto outStream = dbContentRepository->write(*resource.first, false, &batch); |
| 87 | if (outStream == nullptr) { |
| 88 | throw Exception(REPOSITORY_EXCEPTION, "Couldn't open the underlying resource for write: " + resource.first->getContentFullPath()); |
| 89 | } |
| 90 | const int size = gsl::narrow<int>(resource.second->size()); |
| 91 | if (outStream->write(const_cast<uint8_t*>(resource.second->getBuffer()), size) != size) { |
| 92 | throw Exception(REPOSITORY_EXCEPTION, "Failed to write new resource: " + resource.first->getContentFullPath()); |
| 93 | } |
| 94 | } |
| 95 | for (const auto& resource : extendedResources_) { |
| 96 | auto outStream = dbContentRepository->write(*resource.first, true, &batch); |
| 97 | if (outStream == nullptr) { |
| 98 | throw Exception(REPOSITORY_EXCEPTION, "Couldn't open the underlying resource for append: " + resource.first->getContentFullPath()); |
| 99 | } |
| 100 | const int size = gsl::narrow<int>(resource.second->size()); |
| 101 | if (outStream->write(const_cast<uint8_t*>(resource.second->getBuffer()), size) != size) { |
| 102 | throw Exception(REPOSITORY_EXCEPTION, "Failed to append to resource: " + resource.first->getContentFullPath()); |
| 103 | } |
| 104 | } |
| 105 | |
| 106 | rocksdb::WriteOptions options; |
| 107 | options.sync = true; |
| 108 | rocksdb::Status status = opendb->Write(options, &batch); |
| 109 | if (!status.ok()) { |
| 110 | throw Exception(REPOSITORY_EXCEPTION, "Batch write failed: " + status.ToString()); |
| 111 | } |
| 112 | |
| 113 | managedResources_.clear(); |
| 114 | extendedResources_.clear(); |
| 115 | } |
| 116 | |
| 117 | std::shared_ptr<io::BaseStream> DatabaseContentRepository::write(const minifi::ResourceClaim &claim, bool append) { |
| 118 | return write(claim, append, nullptr); |