| /** |
| * @file FlowFileRecord.cpp |
| * Flow file record class implementation |
| * |
| * 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. |
| */ |
| #include <time.h> |
| #include <cstdio> |
| #include <vector> |
| #include <queue> |
| #include <map> |
| #include <memory> |
| #include <string> |
| #include <iostream> |
| #include <fstream> |
| #include <cinttypes> |
| #include "FlowFileRecord.h" |
| #include "core/logging/LoggerConfiguration.h" |
| #include "core/Relationship.h" |
| #include "core/Repository.h" |
| |
| namespace org { |
| namespace apache { |
| namespace nifi { |
| namespace minifi { |
| |
| std::shared_ptr<logging::Logger> FlowFileRecord::logger_ = logging::LoggerFactory<FlowFileRecord>::getLogger(); |
| std::atomic<uint64_t> FlowFileRecord::local_flow_seq_number_(0); |
| |
| FlowFileRecord::FlowFileRecord() { |
| // TODO(adebreceni): |
| // we should revisit if we need these in a follow-up ticket |
| id_ = local_flow_seq_number_++; |
| addAttribute(core::SpecialFlowAttribute::FILENAME, std::to_string(utils::timeutils::getTimeNano())); |
| } |
| |
| std::shared_ptr<FlowFileRecord> FlowFileRecord::DeSerialize(const std::string& key, const std::shared_ptr<core::Repository>& flowRepository, |
| const std::shared_ptr<core::ContentRepository>& content_repo, utils::Identifier& container) { |
| std::string value; |
| |
| if (!flowRepository->Get(key, value)) { |
| logger_->log_error("NiFi FlowFile Store event %s can not found", key); |
| return nullptr; |
| } |
| io::BufferStream stream((const uint8_t*) value.data(), value.length()); |
| |
| auto record = DeSerialize(stream, content_repo, container); |
| |
| if (record) { |
| logger_->log_debug("NiFi FlowFile retrieve uuid %s size " "%" PRIu64 " connection %s success", record->getUUIDStr(), stream.size(), container.to_string()); |
| } else { |
| logger_->log_debug("Couldn't deserialize FlowFile %s from the stream of size " "%" PRIu64, key, stream.size()); |
| } |
| |
| return record; |
| } |
| |
| bool FlowFileRecord::Serialize(io::OutputStream &outStream) { |
| int ret; |
| |
| ret = outStream.write(event_time_); |
| if (ret != 8) { |
| return false; |
| } |
| |
| ret = outStream.write(entry_date_); |
| if (ret != 8) { |
| return false; |
| } |
| |
| ret = outStream.write(lineage_start_date_); |
| if (ret != 8) { |
| return false; |
| } |
| |
| ret = outStream.write(uuid_); |
| if (ret <= 0) { |
| return false; |
| } |
| |
| utils::Identifier containerId; |
| if (connection_) { |
| containerId = connection_->getUUID(); |
| } |
| ret = outStream.write(containerId); |
| if (ret <= 0) { |
| return false; |
| } |
| // write flow attributes |
| uint32_t numAttributes = attributes_.size(); |
| ret = outStream.write(numAttributes); |
| if (ret != 4) { |
| return false; |
| } |
| |
| for (auto& itAttribute : attributes_) { |
| ret = outStream.write(itAttribute.first, true); |
| if (ret <= 0) { |
| return false; |
| } |
| ret = outStream.write(itAttribute.second, true); |
| if (ret <= 0) { |
| return false; |
| } |
| } |
| |
| ret = outStream.write(getContentFullPath()); |
| if (ret <= 0) { |
| return false; |
| } |
| |
| ret = outStream.write(size_); |
| if (ret != 8) { |
| return false; |
| } |
| |
| ret = outStream.write(offset_); |
| if (ret != 8) { |
| return false; |
| } |
| |
| return true; |
| } |
| |
| bool FlowFileRecord::Persist(const std::shared_ptr<core::Repository>& flowRepository) { |
| if (flowRepository->isNoop()) { |
| return true; |
| } |
| |
| io::BufferStream outStream; |
| |
| if (!Serialize(outStream)) { |
| return false; |
| } |
| |
| if (flowRepository->Put(getUUIDStr(), const_cast<uint8_t*>(outStream.getBuffer()), outStream.size())) { |
| logger_->log_debug("NiFi FlowFile Store event %s size " "%" PRIu64 " success", getUUIDStr(), outStream.size()); |
| // on behalf of the persisted record instance |
| if (claim_) claim_->increaseFlowFileRecordOwnedCount(); |
| return true; |
| } else { |
| logger_->log_error("NiFi FlowFile Store failed %s size " "%" PRIu64, getUUIDStr(), outStream.size()); |
| return false; |
| } |
| |
| return true; |
| } |
| |
| std::shared_ptr<FlowFileRecord> FlowFileRecord::DeSerialize(io::InputStream& inStream, const std::shared_ptr<core::ContentRepository>& content_repo, utils::Identifier& container) { |
| int ret; |
| |
| auto file = std::make_shared<FlowFileRecord>(); |
| |
| ret = inStream.read(file->event_time_); |
| if (ret != 8) { |
| return {}; |
| } |
| |
| ret = inStream.read(file->entry_date_); |
| if (ret != 8) { |
| return {}; |
| } |
| |
| ret = inStream.read(file->lineage_start_date_); |
| if (ret != 8) { |
| return {}; |
| } |
| |
| ret = inStream.read(file->uuid_); |
| if (ret <= 0) { |
| return {}; |
| } |
| |
| ret = inStream.read(container); |
| if (ret <= 0) { |
| return {}; |
| } |
| |
| // read flow attributes |
| uint32_t numAttributes = 0; |
| ret = inStream.read(numAttributes); |
| if (ret != 4) { |
| return {}; |
| } |
| |
| for (uint32_t i = 0; i < numAttributes; i++) { |
| std::string key; |
| ret = inStream.read(key, true); |
| if (ret <= 0) { |
| return {}; |
| } |
| std::string value; |
| ret = inStream.read(value, true); |
| if (ret <= 0) { |
| return {}; |
| } |
| file->attributes_[key] = value; |
| } |
| |
| std::string content_full_path; |
| ret = inStream.read(content_full_path); |
| if (ret <= 0) { |
| return {}; |
| } |
| |
| ret = inStream.read(file->size_); |
| if (ret != 8) { |
| return {}; |
| } |
| |
| ret = inStream.read(file->offset_); |
| if (ret != 8) { |
| return {}; |
| } |
| |
| file->claim_ = std::make_shared<ResourceClaim>(content_full_path, content_repo); |
| |
| return file; |
| } |
| |
| } /* namespace minifi */ |
| } /* namespace nifi */ |
| } /* namespace apache */ |
| } /* namespace org */ |