szaszm commented on code in PR #1703: URL: https://github.com/apache/nifi-minifi-cpp/pull/1703#discussion_r1414718702
########## extensions/standard-processors/processors/AttributeRollingWindow.h: ########## @@ -0,0 +1,117 @@ +/** + * 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 <chrono> +#include <memory> +#include <mutex> +#include <optional> +#include <queue> +#include <string_view> + +#include "core/AbstractProcessor.h" +#include "core/Annotation.h" +#include "core/logging/LoggerFactory.h" +#include "core/PropertyDefinitionBuilder.h" +#include "core/PropertyType.h" +#include "core/RelationshipDefinition.h" +#include "RollingWindow.h" +#include "StateManager.h" + +namespace org::apache::nifi::minifi::processors { + +class AttributeRollingWindow final : public core::AbstractProcessor<AttributeRollingWindow> { + public: + using core::AbstractProcessor<AttributeRollingWindow>::AbstractProcessor; + + EXTENSIONAPI static constexpr auto Description = "Track a Rolling Window based on evaluating an Expression Language " + "expression on each FlowFile. Each FlowFile will be emitted with the count of FlowFiles and total aggregate value" + "of values processed in the current window."; + + EXTENSIONAPI static constexpr auto ValueToTrack = core::PropertyDefinitionBuilder<>::createProperty("Value to track") + .withDescription("The expression on which to evaluate each FlowFile. The result of the expression will be added " + "to the rolling window value.") + .isRequired(true) + .supportsExpressionLanguage(true) + .build(); + EXTENSIONAPI static constexpr auto TimeWindow = core::PropertyDefinitionBuilder<>::createProperty("Time window") + .withDescription("The amount of time for a rolling window. The format of the value is expected to be a " + "count followed by a time unit. For example 5 millis, 10 secs, 1 min, 3 hours, 2 days, etc.") + .withPropertyType(core::StandardPropertyTypes::TIME_PERIOD_TYPE) + .build(); + EXTENSIONAPI static constexpr auto WindowLength = core::PropertyDefinitionBuilder<>::createProperty("Window length") + .withDescription("The window length in number of values. Takes precedence over 'Time window'. If set to zero, " + "the 'Time window' property is used instead.") + .isRequired(true) + .withDefaultValue("0") + .withPropertyType(core::StandardPropertyTypes::UNSIGNED_INT_TYPE) + .build(); + EXTENSIONAPI static constexpr auto AttributeNamePrefix = core::PropertyDefinitionBuilder<>::createProperty("Attribute name prefix") + .withDescription("The prefix to add to the generated attribute names. For example, if this is set to 'rolling.window.', " + "then the full attribute names will be 'rolling.window.value', 'rolling.window.count', etc.") + .isRequired(true) + .withDefaultValue("rolling.window.") + .build(); + EXTENSIONAPI static constexpr auto Properties = std::array<core::PropertyReference, 4>{ + ValueToTrack, + TimeWindow, + WindowLength, + AttributeNamePrefix + }; + + EXTENSIONAPI static constexpr auto Success = core::RelationshipDefinition{"success", "All FlowFiles that are " + "successfully processed are routed to this relationship."}; + EXTENSIONAPI static constexpr auto Failure = core::RelationshipDefinition{"failure", "When a FlowFile fails, " + "it is routed here."}; + EXTENSIONAPI static constexpr auto Relationships = std::array{Success, Failure}; + + EXTENSIONAPI static constexpr auto Count = core::OutputAttributeDefinition<1>{"<prefix>count", {Success}, "Number of the values in the rolling window"}; + EXTENSIONAPI static constexpr auto Value = core::OutputAttributeDefinition<1>{"<prefix>value", {Success}, "Sum of the values in the rolling window"}; + EXTENSIONAPI static constexpr auto Mean = core::OutputAttributeDefinition<1>{"<prefix>mean", {Success}, "Mean of the values in the rolling window"}; + EXTENSIONAPI static constexpr auto Median = core::OutputAttributeDefinition<1>{"<prefix>median", {Success}, "Median of the values in the rolling window"}; + EXTENSIONAPI static constexpr auto Variance = core::OutputAttributeDefinition<1>{"<prefix>variance", {Success}, "Variance of the values in the rolling window"}; + EXTENSIONAPI static constexpr auto Stddev = core::OutputAttributeDefinition<1>{"<prefix>stddev", {Success}, "Standard deviation of the values in the rolling window"}; + EXTENSIONAPI static constexpr auto Min = core::OutputAttributeDefinition<1>{"<prefix>min", {Success}, "Smallest value in the rolling window"}; + EXTENSIONAPI static constexpr auto Max = core::OutputAttributeDefinition<1>{"<prefix>max", {Success}, "Largest value in the rolling window"}; Review Comment: fixed in [bdf1cd3](https://github.com/apache/nifi-minifi-cpp/pull/1703/commits/bdf1cd3b5f27b9c4d759ebc23b1a05c259905ef9) ########## extensions/standard-processors/processors/AttributeRollingWindow.cpp: ########## @@ -0,0 +1,122 @@ +/** + * 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 "AttributeRollingWindow.h" +#include <algorithm> +#include <numeric> +#include "fmt/format.h" +#include "core/ProcessContext.h" +#include "core/ProcessSession.h" +#include "core/Resource.h" +#include "utils/expected.h" +#include "utils/OptionalUtils.h" + +namespace org::apache::nifi::minifi::processors { + +void AttributeRollingWindow::onSchedule(core::ProcessContext* context, core::ProcessSessionFactory*) { + gsl_Expects(context); + time_window_ = context->getProperty<core::TimePeriodValue>(TimeWindow) + | utils::transform(&core::TimePeriodValue::getMilliseconds); + window_length_ = context->getProperty<uint64_t>(WindowLength) + | utils::filter([](uint64_t value) { return value > 0; }) + | utils::transform([](uint64_t value) { return size_t{value}; }); Review Comment: fixed in [bdf1cd3](https://github.com/apache/nifi-minifi-cpp/pull/1703/commits/bdf1cd3b5f27b9c4d759ebc23b1a05c259905ef9) ########## extensions/standard-processors/processors/AttributeRollingWindow.cpp: ########## @@ -0,0 +1,122 @@ +/** + * 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 "AttributeRollingWindow.h" +#include <algorithm> +#include <numeric> +#include "fmt/format.h" +#include "core/ProcessContext.h" +#include "core/ProcessSession.h" +#include "core/Resource.h" +#include "utils/expected.h" +#include "utils/OptionalUtils.h" + +namespace org::apache::nifi::minifi::processors { + +void AttributeRollingWindow::onSchedule(core::ProcessContext* context, core::ProcessSessionFactory*) { + gsl_Expects(context); + time_window_ = context->getProperty<core::TimePeriodValue>(TimeWindow) + | utils::transform(&core::TimePeriodValue::getMilliseconds); + window_length_ = context->getProperty<uint64_t>(WindowLength) + | utils::filter([](uint64_t value) { return value > 0; }) + | utils::transform([](uint64_t value) { return size_t{value}; }); + if (!time_window_ && !window_length_) { + throw minifi::Exception{ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "Either 'Time window' or 'Window length' must be set"}; + } + attribute_name_prefix_ = (context->getProperty(AttributeNamePrefix) + | utils::orElse([] { + throw minifi::Exception{ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "'Attribute name prefix' must be set"}; + })).value(); Review Comment: fixed in [bdf1cd3](https://github.com/apache/nifi-minifi-cpp/pull/1703/commits/bdf1cd3b5f27b9c4d759ebc23b1a05c259905ef9) -- 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. To unsubscribe, e-mail: issues-unsubscr...@nifi.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
