adamdebreceni commented on a change in pull request #1044: URL: https://github.com/apache/nifi-minifi-cpp/pull/1044#discussion_r630039434
########## File path: extensions/systemd/ConsumeJournald.cpp ########## @@ -0,0 +1,262 @@ +/** + * + * 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 "ConsumeJournald.h" + +#include <algorithm> + +#include <date/date.h> +#include "spdlog/spdlog.h" // TODO(szaszm): make fmt directly available +#include "utils/GeneralUtils.h" + +namespace org { namespace apache { namespace nifi { namespace minifi { namespace extensions { namespace systemd { + +constexpr const char* ConsumeJournald::CURSOR_KEY; +const core::Relationship ConsumeJournald::Success("success", "Successfully consumed journal messages."); + +const core::Property ConsumeJournald::BatchSize = core::PropertyBuilder::createProperty("Batch Size") + ->withDescription("The maximum number of entries processed in a single execution.") + ->withDefaultValue<size_t>(1000) + ->isRequired(true) + ->build(); + +const core::Property ConsumeJournald::PayloadFormat = core::PropertyBuilder::createProperty("Payload Format") + ->withDescription("Configures flow file content formatting. Raw: only the message. Syslog: similar to syslog or journalctl output.") + ->withDefaultValue<std::string>(PAYLOAD_FORMAT_SYSLOG) + ->withAllowableValues<std::string>({PAYLOAD_FORMAT_RAW, PAYLOAD_FORMAT_SYSLOG}) + ->isRequired(true) + ->build(); + +const core::Property ConsumeJournald::IncludeTimestamp = core::PropertyBuilder::createProperty("Include Timestamp") + ->withDescription("Include message timestamp in the 'timestamp' attribute.") + ->withDefaultValue<bool>(true) + ->isRequired(true) + ->build(); + +const core::Property ConsumeJournald::JournalType = core::PropertyBuilder::createProperty("Journal Type") + ->withDescription("Type of journal to consume.") + ->withDefaultValue<std::string>(JOURNAL_TYPE_SYSTEM) + ->withAllowableValues<std::string>({JOURNAL_TYPE_USER, JOURNAL_TYPE_SYSTEM, JOURNAL_TYPE_BOTH}) + ->isRequired(true) + ->build(); + +const core::Property ConsumeJournald::ProcessOldMessages = core::PropertyBuilder::createProperty("Process Old Messages") + ->withDescription("Process events created before the first usage (schedule) of the processor instance.") + ->withDefaultValue<bool>(false) + ->isRequired(true) + ->build(); + +const core::Property ConsumeJournald::TimestampFormat = core::PropertyBuilder::createProperty("Timestamp Format") + ->withDescription("Format string to use when creating the timestamp attribute or writing messages in the syslog format.") + ->withDefaultValue("%x %X %Z") + ->isRequired(true) + ->build(); + +ConsumeJournald::ConsumeJournald(const std::string &name, const utils::Identifier &id, std::unique_ptr<libwrapper::LibWrapper>&& libwrapper) + :core::Processor{name, id}, libwrapper_{std::move(libwrapper)} +{} + +void ConsumeJournald::initialize() { + setSupportedProperties({BatchSize, PayloadFormat, IncludeTimestamp, JournalType, ProcessOldMessages, TimestampFormat}); + setSupportedRelationships({Success}); + + worker_ = utils::make_unique<Worker>(); +} + +void ConsumeJournald::notifyStop() { + bool running = true; + if (!running_.compare_exchange_strong(running, false, std::memory_order_acq_rel) || !journal_) return; + worker_->enqueue([this] { + journal_ = nullptr; + }).get(); + worker_ = nullptr; +} + +void ConsumeJournald::onSchedule(core::ProcessContext* const context, core::ProcessSessionFactory* const sessionFactory) { + gsl_Expects(context && sessionFactory && !running_ && worker_); + using JournalTypeEnum = systemd::JournalType; + + const auto parse_payload_format = [](const std::string& property_value) -> utils::optional<systemd::PayloadFormat> { + if (utils::StringUtils::equalsIgnoreCase(property_value, PAYLOAD_FORMAT_RAW)) return systemd::PayloadFormat::Raw; + if (utils::StringUtils::equalsIgnoreCase(property_value, PAYLOAD_FORMAT_SYSLOG)) return systemd::PayloadFormat::Syslog; + return utils::nullopt; + }; + const auto parse_journal_type = [](const std::string& property_value) -> utils::optional<JournalTypeEnum> { + if (utils::StringUtils::equalsIgnoreCase(property_value, JOURNAL_TYPE_USER)) return JournalTypeEnum::User; + if (utils::StringUtils::equalsIgnoreCase(property_value, JOURNAL_TYPE_SYSTEM)) return JournalTypeEnum::System; + if (utils::StringUtils::equalsIgnoreCase(property_value, JOURNAL_TYPE_BOTH)) return JournalTypeEnum::Both; + return utils::nullopt; + }; + batch_size_ = context->getProperty<size_t>(BatchSize).value_or(10); Review comment: it seems that all these properties already have default values in their definition, moreover instead of using a fallback value when a value is invalid (not one of the allowable values, .e.g the user mistypes `User`) we should log an error and throw as other processors do -- 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
