Repository: nifi-minifi-cpp Updated Branches: refs/heads/master a5495e375 -> ff1050365
MINIFICPP-49 Added initial implementation of NiFi expression language This closes #188. Signed-off-by: Bin Qiu <[email protected]> Project: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/repo Commit: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/commit/ff105036 Tree: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/tree/ff105036 Diff: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/diff/ff105036 Branch: refs/heads/master Commit: ff105036598afe9b5b183ca968c94f04f3b5f5c4 Parents: a5495e3 Author: Andy I. Christianson <[email protected]> Authored: Wed Nov 8 16:33:55 2017 -0500 Committer: Bin Qiu <[email protected]> Committed: Sun Dec 3 19:48:01 2017 -0800 ---------------------------------------------------------------------- .travis.yml | 6 + CMakeLists.txt | 13 + Extensions.md | 7 + README.md | 10 + docker/Dockerfile | 2 + extensions/expression-language/.gitignore | 7 + extensions/expression-language/CMakeLists.txt | 86 +++++++ extensions/expression-language/Driver.h | 61 +++++ extensions/expression-language/Expression.cpp | 164 +++++++++++++ extensions/expression-language/Parser.yy | 202 ++++++++++++++++ .../expression-language/ProcessContextExpr.cpp | 43 ++++ extensions/expression-language/Scanner.ll | 95 ++++++++ .../impl/expression/Expression.h | 151 ++++++++++++ .../expression-language/noop/CMakeLists.txt | 23 ++ .../noop/ProcessContextExprNoOp.cpp | 35 +++ .../noop/expression/Expression.h | 48 ++++ extensions/http-curl/processors/InvokeHTTP.cpp | 24 +- libminifi/CMakeLists.txt | 1 + libminifi/include/core/ProcessContext.h | 6 + libminifi/include/processors/PutFile.h | 7 +- libminifi/src/processors/PutFile.cpp | 153 ++++++------ .../expression-language-tests/CMakeLists.txt | 40 ++++ .../ExpressionLanguageTests.cpp | 237 +++++++++++++++++++ libminifi/test/unit/PutFileTests.cpp | 2 +- 24 files changed, 1329 insertions(+), 94 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/ff105036/.travis.yml ---------------------------------------------------------------------- diff --git a/.travis.yml b/.travis.yml index 6ff8cb2..4917110 100644 --- a/.travis.yml +++ b/.travis.yml @@ -35,6 +35,8 @@ matrix: packages: - gcc-4.8 - g++-4.8 + - bison + - flex - libboost-all-dev - uuid-dev - doxygen @@ -60,6 +62,8 @@ matrix: - package='ossp-uuid'; [[ $(brew ls --versions ${package}) ]] && { brew outdated ${package} || brew upgrade ${package}; } || brew install ${package} - package='boost'; [[ $(brew ls --versions ${package}) ]] && { brew outdated ${package} || brew upgrade ${package}; } || brew install ${package} - package='cmake'; [[ $(brew ls --versions ${package}) ]] && { brew outdated ${package} || brew upgrade ${package}; } || brew install ${package} + - package='bison'; [[ $(brew ls --versions ${package}) ]] && { brew outdated ${package} || brew upgrade ${package}; } || brew install ${package} + - package='flex'; [[ $(brew ls --versions ${package}) ]] && { brew outdated ${package} || brew upgrade ${package}; } || brew install ${package} - package='ccache'; [[ $(brew ls --versions ${package}) ]] && { brew outdated ${package} || brew upgrade ${package}; } || brew install ${package} - package='openssl'; [[ $(brew ls --versions ${package}) ]] && { brew outdated ${package} || brew upgrade ${package}; } || brew install ${package} - package='doxygen'; [[ $(brew ls --versions ${package}) ]] && { brew outdated ${package} || brew upgrade ${package}; } || brew install ${package} @@ -80,6 +84,8 @@ matrix: - package='ossp-uuid'; [[ $(brew ls --versions ${package}) ]] && { brew outdated ${package} || brew upgrade ${package}; } || brew install ${package} - package='boost'; [[ $(brew ls --versions ${package}) ]] && { brew outdated ${package} || brew upgrade ${package}; } || brew install ${package} - package='cmake'; [[ $(brew ls --versions ${package}) ]] && { brew outdated ${package} || brew upgrade ${package}; } || brew install ${package} + - package='bison'; [[ $(brew ls --versions ${package}) ]] && { brew outdated ${package} || brew upgrade ${package}; } || brew install ${package} + - package='flex'; [[ $(brew ls --versions ${package}) ]] && { brew outdated ${package} || brew upgrade ${package}; } || brew install ${package} - package='ccache'; [[ $(brew ls --versions ${package}) ]] && { brew outdated ${package} || brew upgrade ${package}; } || brew install ${package} - package='openssl'; [[ $(brew ls --versions ${package}) ]] && { brew outdated ${package} || brew upgrade ${package}; } || brew install ${package} - package='doxygen'; [[ $(brew ls --versions ${package}) ]] && { brew outdated ${package} || brew upgrade ${package}; } || brew install ${package} http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/ff105036/CMakeLists.txt ---------------------------------------------------------------------- diff --git a/CMakeLists.txt b/CMakeLists.txt index 59253eb..9e01f10 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -106,6 +106,19 @@ add_subdirectory(thirdparty/yaml-cpp-yaml-cpp-20171024) add_subdirectory(thirdparty/civetweb-1.9.1 EXCLUDE_FROM_ALL) include_directories(thirdparty/concurrentqueue) include_directories(thirdparty/yaml-cpp-yaml-cpp-20171024/include) + +## Expression language extensions +option(DISABLE_EXPRESSION_LANGUAGE "Disables the scripting extensions." OFF) +if (DISABLE_EXPRESSION_LANGUAGE) + # Build expression language NoOp implementation, if necessary + include_directories("extensions/expression-language/noop") + add_subdirectory(extensions/expression-language/noop) +else() + include_directories("extensions/expression-language/impl") + createExtension(EXPRESSION-LANGUAGE-EXTENSIONS "EXPRESSION LANGUAGE EXTENSIONS" "This enables NiFi expression language" "extensions/expression-language" "${TEST_DIR}/expression-language-tests") +endif() + + add_subdirectory(libminifi) if (EXCLUDE_BOOST) http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/ff105036/Extensions.md ---------------------------------------------------------------------- diff --git a/Extensions.md b/Extensions.md index 86a4f59..acebf75 100644 --- a/Extensions.md +++ b/Extensions.md @@ -136,3 +136,10 @@ Note that if Batch Directory is not specified, /tmp/ will be used. *Running PcapTests requires root privileges on Linux. +#Expressions + +To evaluate dynamic expressions using the [NiFi Expression Language](https://nifi.apache.org/docs/nifi-docs/html/expression-language-guide.html), +use the `bool getProperty(const std::string &name, std::string &value, const std::shared_ptr<FlowFile> &flow_file);` +method within processor extensions. The expression defined in the property will be compiled if it hasn't already been, and +the expression will be evaluated against the provided flow file (may be `nullptr`). The result of the evaluation is +stored into the `&value` parameter. http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/ff105036/README.md ---------------------------------------------------------------------- diff --git a/README.md b/README.md index ea5e45f..a6286d8 100644 --- a/README.md +++ b/README.md @@ -77,6 +77,10 @@ Perspectives of the role of MiNiFi should be from the perspective of the agent a * 4.8.4 or greater * g++ * 4.8.4 or greater +* bison + * 3.0 or greater +* flex + * 2.5 or greater **NOTE** if Lua support is enabled, then a C++ compiler with support for c++-14 must be used. If using GCC, version 6.x or greater is recommended. @@ -121,6 +125,8 @@ using a devtools-* package from the Software Collections (SCL). # ~/Development/code/apache/nifi-minifi-cpp on git:master $ yum install cmake \ gcc gcc-c++ \ + bison \ + flex \ libcurl-devel \ rocksdb-devel rocksdb \ libuuid libuuid-devel \ @@ -150,6 +156,8 @@ Aptitude based Linux Distributions # ~/Development/code/apache/nifi-minifi-cpp on git:master $ apt-get install cmake \ gcc g++ \ + bison \ + flex \ libcurl-dev \ librocksdb-dev librocksdb4.1 \ uuid-dev uuid \ @@ -176,6 +184,8 @@ OS X Using Homebrew (with XCode Command Line Tools installed) ``` # ~/Development/code/apache/nifi-minifi-cpp on git:master $ brew install cmake \ + bison \ + flex \ rocksdb \ ossp-uuid \ boost \ http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/ff105036/docker/Dockerfile ---------------------------------------------------------------------- diff --git a/docker/Dockerfile b/docker/Dockerfile index 0892c6f..72ca40f 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -30,6 +30,8 @@ ARG MINIFI_SOURCE_CODE RUN apk --update --no-cache upgrade && apk --update --no-cache add gcc \ g++ \ make \ + bison \ + flex \ wget \ gdb \ musl-dev \ http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/ff105036/extensions/expression-language/.gitignore ---------------------------------------------------------------------- diff --git a/extensions/expression-language/.gitignore b/extensions/expression-language/.gitignore new file mode 100644 index 0000000..38744a8 --- /dev/null +++ b/extensions/expression-language/.gitignore @@ -0,0 +1,7 @@ +/Parser.cpp +/Parser.hpp +/Scanner.h +/Scanner.cpp +/location.hh +/position.hh +/stack.hh http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/ff105036/extensions/expression-language/CMakeLists.txt ---------------------------------------------------------------------- diff --git a/extensions/expression-language/CMakeLists.txt b/extensions/expression-language/CMakeLists.txt new file mode 100644 index 0000000..db3bb88 --- /dev/null +++ b/extensions/expression-language/CMakeLists.txt @@ -0,0 +1,86 @@ +# +# 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. +# + +set(CMAKE_EXE_LINKER_FLAGS "-Wl,--export-all-symbols") +set(CMAKE_SHARED_LINKER_FLAGS "-Wl,--export-symbols") + + +find_package(BISON REQUIRED) +find_package(FLEX REQUIRED) + +bison_target( + el-parser + ${CMAKE_CURRENT_SOURCE_DIR}/Parser.yy + ${CMAKE_CURRENT_SOURCE_DIR}/Parser.cpp +) + +flex_target( + el-scanner + ${CMAKE_CURRENT_SOURCE_DIR}/Scanner.ll + ${CMAKE_CURRENT_SOURCE_DIR}/Scanner.cpp +) + +add_flex_bison_dependency(el-scanner el-parser) + +include_directories(../../libminifi/include ../../libminifi/include/core ../../thirdparty/spdlog-20170710/include ../../thirdparty/concurrentqueue ../../thirdparty/yaml-cpp-yaml-cpp-0.5.3/include ../../thirdparty/civetweb-1.9.1/include ../../thirdparty/jsoncpp/include ../../thirdparty/) +include_directories(impl) + +file(GLOB SOURCES "*.cpp") + +add_library(minifi-expression-language-extensions STATIC ${SOURCES} ${BISON_el-parser_OUTPUTS} ${FLEX_el-scanner_OUTPUTS}) +set_property(TARGET minifi-expression-language-extensions PROPERTY POSITION_INDEPENDENT_CODE ON) +if(THREADS_HAVE_PTHREAD_ARG) + target_compile_options(PUBLIC minifi-expression-language-extensions "-pthread") +endif() +if(CMAKE_THREAD_LIBS_INIT) + target_link_libraries(minifi-expression-language-extensions "${CMAKE_THREAD_LIBS_INIT}") +endif() + +find_package(UUID REQUIRED) +target_link_libraries(minifi-expression-language-extensions ${LIBMINIFI} ${UUID_LIBRARIES} ${JSONCPP_LIB}) +add_dependencies(minifi-expression-language-extensions jsoncpp_project) +find_package(OpenSSL REQUIRED) +include_directories(${OPENSSL_INCLUDE_DIR}) +target_link_libraries(minifi-expression-language-extensions ${CMAKE_DL_LIBS}) +find_package(ZLIB REQUIRED) +include_directories(${ZLIB_INCLUDE_DIRS}) +target_link_libraries (minifi-expression-language-extensions ${ZLIB_LIBRARIES}) +find_package(Boost COMPONENTS system filesystem REQUIRED) +include_directories(${Boost_INCLUDE_DIRS}) +target_link_libraries(minifi-expression-language-extensions ${Boost_SYSTEM_LIBRARY}) +target_link_libraries(minifi-expression-language-extensions ${Boost_FILESYSTEM_LIBRARY}) + +if (WIN32) + set_target_properties(minifi-expression-language-extensions PROPERTIES + LINK_FLAGS "/WHOLEARCHIVE" + ) +elseif (APPLE) + set_target_properties(minifi-expression-language-extensions PROPERTIES + LINK_FLAGS "-Wl,-all_load" + ) +else () + set_target_properties(minifi-expression-language-extensions PROPERTIES + LINK_FLAGS "-Wl,--whole-archive" + ) +endif () + +SET (EXPRESSION-LANGUAGE-EXTENSIONS minifi-expression-language-extensions PARENT_SCOPE) + +register_extension(minifi-expression-language-extensions) + http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/ff105036/extensions/expression-language/Driver.h ---------------------------------------------------------------------- diff --git a/extensions/expression-language/Driver.h b/extensions/expression-language/Driver.h new file mode 100644 index 0000000..4e66b4c --- /dev/null +++ b/extensions/expression-language/Driver.h @@ -0,0 +1,61 @@ +/** + * 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. + */ + +#ifndef __EXPRESSION_LANGUAGE_DRIVER_H__ +#define __EXPRESSION_LANGUAGE_DRIVER_H__ + +#include <string> +#include <map> +#include <sstream> +#include <expression/Expression.h> + +#undef yyFlexLexer +#include <FlexLexer.h> +#include "Parser.hpp" + +#undef YY_DECL +#define YY_DECL int org::apache::nifi::minifi::expression::Driver::lex(org::apache::nifi::minifi::expression::Parser::semantic_type* yylval, \ + org::apache::nifi::minifi::expression::Parser::location_type* yylloc) + +namespace org { +namespace apache { +namespace nifi { +namespace minifi { +namespace expression { + +class Driver : public yyFlexLexer { + public: + explicit Driver(std::istream *input = nullptr, std::ostream *output = nullptr) + : yyFlexLexer(input, output), + result("") { + } + ~Driver() override = default; + int lex(Parser::semantic_type *yylval, + Parser::location_type *yylloc); + + std::map<std::string, int> variables; + + Expression result; +}; + +} /* namespace expression */ +} /* namespace minifi */ +} /* namespace nifi */ +} /* namespace apache */ +} /* namespace org */ + +#endif /* __EXPRESSION_LANGUAGE_DRIVER_H__ */ http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/ff105036/extensions/expression-language/Expression.cpp ---------------------------------------------------------------------- diff --git a/extensions/expression-language/Expression.cpp b/extensions/expression-language/Expression.cpp new file mode 100644 index 0000000..9151948 --- /dev/null +++ b/extensions/expression-language/Expression.cpp @@ -0,0 +1,164 @@ +/** + * 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 <utility> +#include <iostream> + +#include <expression/Expression.h> +#include "Driver.h" + +namespace org { +namespace apache { +namespace nifi { +namespace minifi { +namespace expression { + +Expression compile(const std::string &expr_str) { + std::stringstream expr_str_stream(expr_str); + Driver driver(&expr_str_stream); + Parser parser(&driver); + parser.parse(); + return driver.result; +} + +Expression make_static(std::string val) { + return Expression(std::move(val)); +} + +Expression make_dynamic(std::function<std::string(const Parameters ¶ms)> val_fn) { + return Expression("", std::move(val_fn)); +} + +Expression make_dynamic_attr(const std::string &attribute_id) { + return make_dynamic([attribute_id](const Parameters ¶ms) -> std::string { + std::string result; + params.flow_file.lock()->getAttribute(attribute_id, result); + return result; + }); +} + +std::string expr_hostname(const std::vector<std::string> &args) { + char hostname[1024]; + hostname[1023] = '\0'; + gethostname(hostname, 1023); + return std::string(hostname); +} + +std::string expr_toUpper(const std::vector<std::string> &args) { + std::string result = args[0]; + std::transform(result.begin(), result.end(), result.begin(), ::toupper); + return result; +} + +template<std::string T(const std::vector<std::string> &)> +Expression make_dynamic_function_incomplete(const std::string &function_name, + const std::vector<Expression> &args, + std::size_t num_args) { + if (args.size() == num_args) { + return make_dynamic([=](const Parameters ¶ms) -> std::string { + std::vector<std::string> evaluated_args; + + for (const auto &arg : args) { + evaluated_args.emplace_back(arg(params)); + } + + return T(evaluated_args); + }); + } else { + auto result = make_dynamic([](const Parameters ¶ms) -> std::string { + throw std::runtime_error("Attempted to call incomplete function"); + }); + + result.complete = [function_name, args](Expression expr) -> Expression { + std::vector<Expression> complete_args = {expr}; + complete_args.insert(complete_args.end(), args.begin(), args.end()); + return make_dynamic_function(function_name, complete_args); + }; + + return result; + } +} + +Expression make_dynamic_function(const std::string &function_name, const std::vector<Expression> &args) { + if (function_name == "hostname") { + return make_dynamic_function_incomplete<expr_hostname>(function_name, args, 0); + } else if (function_name == "toUpper") { + return make_dynamic_function_incomplete<expr_toUpper>(function_name, args, 1); + } else { + std::string msg("Unknown expression function: "); + msg.append(function_name); + throw std::runtime_error(msg); + } +} + +Expression make_dynamic_function_postfix(const Expression &subject, const Expression &fn) { + return fn.complete(subject); +} + +bool Expression::isDynamic() const { + if (val_fn_) { + return true; + } else { + return false; + } +} + +Expression Expression::operator+(const Expression &other_expr) const { + if (isDynamic() && other_expr.isDynamic()) { + auto val_fn = val_fn_; + auto other_val_fn = other_expr.val_fn_; + return make_dynamic([val_fn, other_val_fn](const Parameters ¶ms) -> std::string { + std::string result = val_fn(params); + result.append(other_val_fn(params)); + return result; + }); + } else if (isDynamic() && !other_expr.isDynamic()) { + auto val_fn = val_fn_; + auto other_val = other_expr.val_; + return make_dynamic([val_fn, other_val](const Parameters ¶ms) -> std::string { + std::string result = val_fn(params); + result.append(other_val); + return result; + }); + } else if (!isDynamic() && other_expr.isDynamic()) { + auto val = val_; + auto other_val_fn = other_expr.val_fn_; + return make_dynamic([val, other_val_fn](const Parameters ¶ms) -> std::string { + std::string result(val); + result.append(other_val_fn(params)); + return result; + }); + } else if (!isDynamic() && !other_expr.isDynamic()) { + std::string result(val_); + result.append(other_expr.val_); + return make_static(result); + } +} + +std::string Expression::operator()(const Parameters ¶ms) const { + if (isDynamic()) { + return val_fn_(params); + } else { + return val_; + } +} + +} /* namespace expression */ +} /* namespace minifi */ +} /* namespace nifi */ +} /* namespace apache */ +} /* namespace org */ http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/ff105036/extensions/expression-language/Parser.yy ---------------------------------------------------------------------- diff --git a/extensions/expression-language/Parser.yy b/extensions/expression-language/Parser.yy new file mode 100644 index 0000000..0ffe536 --- /dev/null +++ b/extensions/expression-language/Parser.yy @@ -0,0 +1,202 @@ +/** + * 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. + */ + +%skeleton "lalr1.cc" +%require "3.0" + +%define api.namespace {org::apache::nifi::minifi::expression} +%parse-param {Driver* driver} +%locations + +%define parser_class_name {Parser} +%define parse.error verbose +%define api.value.type variant +%define parse.assert +%define api.token.prefix {TOK_} + +%code requires +{ + #include <string> + + #include <expression/Expression.h> + #include "location.hh" + + namespace org { + namespace apache { + namespace nifi { + namespace minifi { + namespace expression { + class Driver; + } /* namespace expression */ + } /* namespace minifi */ + } /* namespace nifi */ + } /* namespace apache */ + } /* namespace org */ +} + +%code +{ + #include <utility> + #include <memory> + #include <functional> + + #include <expression/Expression.h> + #include "Driver.h" + + #undef yylex + #define yylex driver->lex +} + +%token + END 0 "eof" + NEWLINE "\n" + DOLLAR "$" + LCURLY "{" + RCURLY "}" + LPAREN "(" + RPAREN ")" + LSQUARE "[" + RSQUARE "]" + PIPE "|" + COMMA "," + COLON ":" + SEMI ";" + FSLASH "/" + BSLASH "\\" + STAR "*" + HASH "#" + SQUOTE "'" + DQUOTE "\"" +; + +%token <std::string> IDENTIFIER "identifier" +%token <std::string> MISC "misc" +%token <std::string> WHITESPACE "whitespace" +%token <int> NUMBER "number" + +%type <std::string> exp_whitespace +%type <std::string> exp_whitespaces +%type <std::string> attr_id +%type <Expression> fn_call +%type <Expression> fn_arg +%type <std::vector<Expression>> fn_args +%type <Expression> exp_content +%type <Expression> exp_content_val +%type <Expression> exp +%type <Expression> exps +%type <std::string> text_no_quote_no_dollar +%type <std::string> text_inc_quote_escaped_dollar +%type <std::string> text_inc_dollar +%type <std::string> quoted_text +%type <std::string> quoted_text_content +%type <Expression> text_or_exp + +%% + +%start root; + +root: exps { driver->result = $1; } + ; + +text_no_quote_no_dollar: IDENTIFIER { std::swap($$, $1); } + | WHITESPACE { std::swap($$, $1); } + | NEWLINE { $$ = "\n"; } + | MISC { std::swap($$, $1); } + | LCURLY { $$ = "{"; } + | RCURLY { $$ = "}"; } + | LPAREN { $$ = "("; } + | RPAREN { $$ = ")"; } + | LSQUARE { $$ = "["; } + | RSQUARE { $$ = "]"; } + | PIPE { $$ = "|"; } + | COMMA { $$ = ","; } + | COLON { $$ = ":"; } + | SEMI { $$ = ";"; } + | FSLASH { $$ = "/"; } + | BSLASH { $$ = "\\"; } + | STAR { $$ = "*"; } + | HASH { $$ = "#"; } + | NUMBER { $$ = std::to_string($1); } + ; + +text_inc_quote_escaped_dollar: text_no_quote_no_dollar { std::swap($$, $1); } + | SQUOTE { $$ = "'"; } + | DQUOTE { $$ = "\""; } + | DOLLAR DOLLAR { $$ = "$"; } + ; + +text_inc_dollar: text_no_quote_no_dollar { std::swap($$, $1); } + | DOLLAR { $$ = "$"; } + ; + +quoted_text_content: %empty {} + | quoted_text_content text_inc_dollar { $$ = $1 + $2; } + ; + +quoted_text: SQUOTE quoted_text_content SQUOTE { std::swap($$, $2); } + | DQUOTE quoted_text_content DQUOTE { std::swap($$, $2); } + ; + +text_or_exp: text_inc_quote_escaped_dollar { $$ = make_static(std::move($1)); } + | exp { $$ = $1; } + ; + +exps: %empty {} + | exps text_or_exp { $$ = $1 + $2; } + ; + +exp_whitespace: WHITESPACE {} + | NEWLINE {} + ; + +exp_whitespaces: %empty {} + | exp_whitespaces exp_whitespace {} + ; + +attr_id: quoted_text exp_whitespaces { std::swap($$, $1); } + | IDENTIFIER exp_whitespaces { std::swap($$, $1); } + ; + +fn_arg: exp_content_val { $$ = $1; } + ; + +fn_args: %empty {} + | fn_args fn_arg { $$.insert($$.end(), $1.begin(), $1.end()); $$.push_back($2); } + +fn_call: attr_id LPAREN fn_args RPAREN exp_whitespaces { $$ = make_dynamic_function(std::move($1), $3); } + +exp_content_val: attr_id { $$ = make_dynamic_attr(std::move($1)); } + | fn_call { $$ = $1; } + ; + +exp_content: exp_content_val { $$ = $1; } + | exp_content_val COLON exp_whitespace fn_call { $$ = make_dynamic_function_postfix($1, $4); } + ; + +exp: DOLLAR LCURLY exp_whitespaces exp_content RCURLY { $$ = $4; } + ; + +%% + +void org::apache::nifi::minifi::expression::Parser::error(const location_type &location, + const std::string &message) { + std::stringstream err_msg; + err_msg << location; + err_msg << ": "; + err_msg << message; + throw std::runtime_error(err_msg.str()); +} http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/ff105036/extensions/expression-language/ProcessContextExpr.cpp ---------------------------------------------------------------------- diff --git a/extensions/expression-language/ProcessContextExpr.cpp b/extensions/expression-language/ProcessContextExpr.cpp new file mode 100644 index 0000000..d442978 --- /dev/null +++ b/extensions/expression-language/ProcessContextExpr.cpp @@ -0,0 +1,43 @@ +/** + * 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 <core/ProcessContext.h> + +namespace org { +namespace apache { +namespace nifi { +namespace minifi { +namespace core { + +bool ProcessContext::getProperty(const std::string &name, std::string &value, + const std::shared_ptr<FlowFile> &flow_file) { + if (expressions_.find(name) == expressions_.end()) { + std::string expression_str; + getProperty(name, expression_str); + logger_->log_info("Compiling expression for %s/%s: %s", getProcessorNode()->getName(), name, expression_str); + expressions_.emplace(name, expression::compile(expression_str)); + } + + value = expressions_[name]({flow_file}); + return true; +} + +} /* namespace core */ +} /* namespace minifi */ +} /* namespace nifi */ +} /* namespace apache */ +} /* namespace org */ \ No newline at end of file http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/ff105036/extensions/expression-language/Scanner.ll ---------------------------------------------------------------------- diff --git a/extensions/expression-language/Scanner.ll b/extensions/expression-language/Scanner.ll new file mode 100644 index 0000000..2660539 --- /dev/null +++ b/extensions/expression-language/Scanner.ll @@ -0,0 +1,95 @@ +/** + * 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 <cerrno> + #include <climits> + #include <cstdlib> + #include <string> + #include <string> + #include <sstream> + #include "Driver.h" + #include "Parser.hpp" +%} + +%option noyywrap +%option nounput +%option batch +%option noinput +%option 8bit +%option c++ + +id [a-zA-Z][a-zA-Z_0-9]* +int [0-9]+ +whitespace [ \r\t]+ + +%{ + #define YY_USER_ACTION yylloc->columns(yyleng); +%} + +%% + +%{ + yylloc->step(); +%} + +"\n" return Parser::token::TOK_NEWLINE; +"$" return Parser::token::TOK_DOLLAR; +"{" return Parser::token::TOK_LCURLY; +"}" return Parser::token::TOK_RCURLY; +"(" return Parser::token::TOK_LPAREN; +")" return Parser::token::TOK_RPAREN; +"[" return Parser::token::TOK_LSQUARE; +"]" return Parser::token::TOK_RSQUARE; +"|" return Parser::token::TOK_PIPE; +"," return Parser::token::TOK_COMMA; +":" return Parser::token::TOK_COLON; +";" return Parser::token::TOK_SEMI; +"/" return Parser::token::TOK_FSLASH; +"\\" return Parser::token::TOK_BSLASH; +"*" return Parser::token::TOK_STAR; +"#" return Parser::token::TOK_HASH; +"'" return Parser::token::TOK_SQUOTE; +"\"" return Parser::token::TOK_DQUOTE; + +{whitespace} { + yylval->build<std::string>(yytext); + return Parser::token::TOK_WHITESPACE; +} + +{int} { + yylval->build<int>(std::atoi(yytext)); + return Parser::token::TOK_NUMBER; +} + +{id} { + yylval->build<std::string>(yytext); + return Parser::token::TOK_IDENTIFIER; +} + +. { + yylval->build<std::string>(yytext); + return Parser::token::TOK_MISC; +} + +<<EOF>> return Parser::token::TOK_END; + +%% + +int yyFlexLexer::yylex() { + throw std::logic_error("Not implemented."); +} http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/ff105036/extensions/expression-language/impl/expression/Expression.h ---------------------------------------------------------------------- diff --git a/extensions/expression-language/impl/expression/Expression.h b/extensions/expression-language/impl/expression/Expression.h new file mode 100644 index 0000000..492bc0b --- /dev/null +++ b/extensions/expression-language/impl/expression/Expression.h @@ -0,0 +1,151 @@ +/** + * 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. + */ + +#ifndef NIFI_MINIFI_CPP_EXPRESSION_H +#define NIFI_MINIFI_CPP_EXPRESSION_H + +#include <core/FlowFile.h> + +#include <string> +#include <memory> +#include <functional> + +namespace org { +namespace apache { +namespace nifi { +namespace minifi { +namespace expression { + +typedef struct { + std::weak_ptr<core::FlowFile> flow_file; +} Parameters; + +static const std::function<std::string(const Parameters ¶ms)> NOOP_FN; + +/** + * A NiFi expression langauge expression which can be composed with other + * expressions (via the + operator. + */ +class Expression { + public: + + explicit Expression(std::string val = "", + std::function<std::string(const Parameters &)> val_fn = NOOP_FN) + : val_(std::move(val)), + val_fn_(std::move(val_fn)), + fn_args_() { + } + + /** + * Whether or not this expression is dynamic. If it is not dynamic, then + * the expression can be computed at compile time when composed with other + * expressions. + * + * @return true if expression is dynamic + */ + bool isDynamic() const; + + /** + * Combine this expression with another expression. Intermediate results + * are computed when non-dynamic expressions are composed. + * + * @param other_expr + * @return combined expression + */ + Expression operator+(const Expression &other_expr) const; + + /** + * Evaluate the result of the expression as a function of the given parameters. + * + * @param params + * @return dynamically-computed result of expression + */ + std::string operator()(const Parameters ¶ms) const; + + /** + * If this expression is incomplete (i.e. a function with incomplete arguments), then this + * will return a completed expression based on the provided expression. + */ + std::function<Expression(Expression)> complete = [](Expression expr) -> Expression { + throw std::runtime_error("Attempted to complete already complete expression."); + }; + + protected: + std::string val_; + std::function<std::string(const Parameters ¶ms)> val_fn_; + std::vector<Expression> fn_args_; +}; + +/** + * Compiles an expression from a string in the NiFi expression language syntax. + * + * @param expr_str + * @return + */ +Expression compile(const std::string &expr_str); + +/** + * Creates a string expression that is not dynamic. + * + * @param val + * @return + */ +Expression make_static(std::string val); + +/** + * Creates an arbitrary dynamic expression which evaluates to the given value function. + * + * @param val_fn + * @return + */ +Expression make_dynamic(std::function<std::string(const Parameters ¶ms)> val_fn); + +/** + * Creates a dynamic expression which evaluates the given flow file attribute. + * + * @param attribute_id + * @return + */ +Expression make_dynamic_attr(const std::string &attribute_id); + +/** + * Creates a dynamic expression which evaluates the given function as defined + * in NiFi expression language. + * + * @param function_name + * @param args + * @return + */ +Expression make_dynamic_function(const std::string &function_name, const std::vector<Expression> &args); + +/** + * Creates a dynamic expression which feeds the subject as the first argument into + * the provided function expression. This enables expressions like attr:toLower() + * + * @param subject + * @param fn + * @return + */ +Expression make_dynamic_function_postfix(const Expression &subject, const Expression &fn); + +} /* namespace expression */ +} /* namespace minifi */ +} /* namespace nifi */ +} /* namespace apache */ +} /* namespace org */ + +#endif //NIFI_MINIFI_CPP_EXPRESSION_H http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/ff105036/extensions/expression-language/noop/CMakeLists.txt ---------------------------------------------------------------------- diff --git a/extensions/expression-language/noop/CMakeLists.txt b/extensions/expression-language/noop/CMakeLists.txt new file mode 100644 index 0000000..9cf450b --- /dev/null +++ b/extensions/expression-language/noop/CMakeLists.txt @@ -0,0 +1,23 @@ +# 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. +# + +message(STATUS "Expression language is disabled; using NoOp implementation") +file(GLOB SOURCES "*.cpp") +include_directories(../../../libminifi/include ../../../libminifi/include/core ../../../thirdparty/spdlog-20170710/include ../../../thirdparty/concurrentqueue ../../../thirdparty/yaml-cpp-yaml-cpp-0.5.3/include ../../../thirdparty/civetweb-1.9.1/include ../../../thirdparty/jsoncpp/include ../../../thirdparty/) +add_library(minifi-expression-language-extensions STATIC ${SOURCES}) \ No newline at end of file http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/ff105036/extensions/expression-language/noop/ProcessContextExprNoOp.cpp ---------------------------------------------------------------------- diff --git a/extensions/expression-language/noop/ProcessContextExprNoOp.cpp b/extensions/expression-language/noop/ProcessContextExprNoOp.cpp new file mode 100644 index 0000000..fb34b15 --- /dev/null +++ b/extensions/expression-language/noop/ProcessContextExprNoOp.cpp @@ -0,0 +1,35 @@ +/** + * 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 <core/ProcessContext.h> + +namespace org { +namespace apache { +namespace nifi { +namespace minifi { +namespace core { + +bool ProcessContext::getProperty(const std::string &name, std::string &value, + const std::shared_ptr<FlowFile> &flow_file) { + return getProperty(name, value); +} + +} /* namespace core */ +} /* namespace minifi */ +} /* namespace nifi */ +} /* namespace apache */ +} /* namespace org */ \ No newline at end of file http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/ff105036/extensions/expression-language/noop/expression/Expression.h ---------------------------------------------------------------------- diff --git a/extensions/expression-language/noop/expression/Expression.h b/extensions/expression-language/noop/expression/Expression.h new file mode 100644 index 0000000..08c4ae4 --- /dev/null +++ b/extensions/expression-language/noop/expression/Expression.h @@ -0,0 +1,48 @@ +/** + * 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. + */ + +#ifndef NIFI_MINIFI_CPP_EXPRESSION_H +#define NIFI_MINIFI_CPP_EXPRESSION_H + +#include <core/FlowFile.h> + +namespace org { +namespace apache { +namespace nifi { +namespace minifi { +namespace expression { + +typedef struct { + std::weak_ptr<core::FlowFile> flow_file; +} Parameters; + +/** + * A minimal definition of an Expression with a NoOp implementation. + */ +class Expression { + public: + + explicit Expression(std::string, std::function<std::string(const Parameters &)>); +}; + +} /* namespace expression */ +} /* namespace minifi */ +} /* namespace nifi */ +} /* namespace apache */ +} /* namespace org */ + +#endif //NIFI_MINIFI_CPP_EXPRESSION_H http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/ff105036/extensions/http-curl/processors/InvokeHTTP.cpp ---------------------------------------------------------------------- diff --git a/extensions/http-curl/processors/InvokeHTTP.cpp b/extensions/http-curl/processors/InvokeHTTP.cpp index e33b651..ef13be4 100644 --- a/extensions/http-curl/processors/InvokeHTTP.cpp +++ b/extensions/http-curl/processors/InvokeHTTP.cpp @@ -241,10 +241,10 @@ bool InvokeHTTP::emitFlowFile(const std::string &method) { } void InvokeHTTP::onTrigger(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSession> &session) { - logger_->log_info("onTrigger InvokeHTTP with %s to %s", method_, url_); - std::shared_ptr<FlowFileRecord> flowFile = std::static_pointer_cast<FlowFileRecord>(session->get()); + std::string url = url_; + if (flowFile == nullptr) { if (!emitFlowFile(method_)) { logger_->log_info("InvokeHTTP -- create flow file with %s", method_.c_str()); @@ -254,12 +254,16 @@ void InvokeHTTP::onTrigger(const std::shared_ptr<core::ProcessContext> &context, return; } } else { - logger_->log_info("InvokeHTTP -- Received flowfile "); + context->getProperty(URL.getName(), url, flowFile); + logger_->log_info("InvokeHTTP -- Received flowfile"); } + + logger_->log_info("onTrigger InvokeHTTP with %s to %s", method_, url); + // create a transaction id std::string tx_id = generateId(); - utils::HTTPClient client(url_, ssl_context_service_); + utils::HTTPClient client(url, ssl_context_service_); client.initialize(method_); client.setConnectionTimeout(connect_timeout_); @@ -319,7 +323,7 @@ void InvokeHTTP::onTrigger(const std::shared_ptr<core::ProcessContext> &context, flowFile->addAttribute(STATUS_CODE, std::to_string(http_code)); if (response_headers.size() > 0) flowFile->addAttribute(STATUS_MESSAGE, response_headers.at(0)); - flowFile->addAttribute(REQUEST_URL, url_); + flowFile->addAttribute(REQUEST_URL, url); flowFile->addAttribute(TRANSACTION_ID, tx_id); bool isSuccess = ((int32_t) (http_code / 100)) == 2; @@ -342,7 +346,7 @@ void InvokeHTTP::onTrigger(const std::shared_ptr<core::ProcessContext> &context, response_flow->addAttribute(STATUS_CODE, std::to_string(http_code)); if (response_headers.size() > 0) flowFile->addAttribute(STATUS_MESSAGE, response_headers.at(0)); - response_flow->addAttribute(REQUEST_URL, url_); + response_flow->addAttribute(REQUEST_URL, url); response_flow->addAttribute(TRANSACTION_ID, tx_id); io::DataStream stream((const uint8_t*) response_body.data(), response_body.size()); // need an import from the data stream. @@ -355,8 +359,12 @@ void InvokeHTTP::onTrigger(const std::shared_ptr<core::ProcessContext> &context, } } -void InvokeHTTP::route(std::shared_ptr<FlowFileRecord> &request, std::shared_ptr<FlowFileRecord> &response, const std::shared_ptr<core::ProcessSession> &session, - const std::shared_ptr<core::ProcessContext> &context, bool isSuccess, int statusCode) { +void InvokeHTTP::route(std::shared_ptr<FlowFileRecord> &request, + std::shared_ptr<FlowFileRecord> &response, + const std::shared_ptr<core::ProcessSession> &session, + const std::shared_ptr<core::ProcessContext> &context, + bool isSuccess, + int statusCode) { // check if we should yield the processor if (!isSuccess && request == nullptr) { context->yield(); http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/ff105036/libminifi/CMakeLists.txt ---------------------------------------------------------------------- diff --git a/libminifi/CMakeLists.txt b/libminifi/CMakeLists.txt index b184337..0829072 100644 --- a/libminifi/CMakeLists.txt +++ b/libminifi/CMakeLists.txt @@ -74,6 +74,7 @@ target_link_libraries(core-minifi ${UUID_LIBRARIES} ${JSONCPP_LIB} yaml-cpp ) find_package(ZLIB REQUIRED) include_directories(${ZLIB_INCLUDE_DIRS}) +target_link_libraries(core-minifi minifi-expression-language-extensions) target_link_libraries (core-minifi ${ZLIB_LIBRARIES}) # Include OpenSSL http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/ff105036/libminifi/include/core/ProcessContext.h ---------------------------------------------------------------------- diff --git a/libminifi/include/core/ProcessContext.h b/libminifi/include/core/ProcessContext.h index 1158c71..a9e0013 100644 --- a/libminifi/include/core/ProcessContext.h +++ b/libminifi/include/core/ProcessContext.h @@ -26,6 +26,7 @@ #include <atomic> #include <algorithm> #include <memory> +#include <expression/Expression.h> #include "Property.h" #include "core/ContentRepository.h" #include "core/repository/FileSystemRepository.h" @@ -34,6 +35,7 @@ #include "core/logging/LoggerConfiguration.h" #include "ProcessorNode.h" #include "core/Repository.h" +#include "core/FlowFile.h" namespace org { namespace apache { @@ -67,6 +69,7 @@ class ProcessContext : public controller::ControllerServiceLookup { bool getProperty(const std::string &name, std::string &value) { return processor_node_->getProperty(name, value); } + bool getProperty(const std::string &name, std::string &value, const std::shared_ptr<FlowFile> &flow_file); // Sets the property value using the property's string name bool setProperty(const std::string &name, std::string value) { return processor_node_->setProperty(name, value); @@ -168,6 +171,9 @@ class ProcessContext : public controller::ControllerServiceLookup { std::shared_ptr<core::ContentRepository> content_repo_; // Processor std::shared_ptr<ProcessorNode> processor_node_; + + std::map<std::string, org::apache::nifi::minifi::expression::Expression> expressions_; + // Logger std::shared_ptr<logging::Logger> logger_; http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/ff105036/libminifi/include/processors/PutFile.h ---------------------------------------------------------------------- diff --git a/libminifi/include/processors/PutFile.h b/libminifi/include/processors/PutFile.h index 46a2f57..c2f4d88 100644 --- a/libminifi/include/processors/PutFile.h +++ b/libminifi/include/processors/PutFile.h @@ -92,6 +92,7 @@ class PutFile : public core::Processor { bool write_succeeded_ = false; std::string tmp_file_; std::string dest_file_; + std::string dest_dir_; bool try_mkdirs_; }; @@ -101,13 +102,12 @@ class PutFile : public core::Processor { * @param filename from which to generate temporary write file path * @return */ - std::string tmpWritePath(const std::string &filename) const; + std::string tmpWritePath(const std::string &filename, const std::string &directory) const; protected: private: - std::string directory_; std::string conflict_resolution_; bool try_mkdirs_ = true; int64_t max_dest_files_ = -1; @@ -115,7 +115,8 @@ class PutFile : public core::Processor { bool putFile(core::ProcessSession *session, std::shared_ptr<FlowFileRecord> flowFile, const std::string &tmpFile, - const std::string &destFile); + const std::string &destFile, + const std::string &destDir); std::shared_ptr<logging::Logger> logger_; static std::shared_ptr<utils::IdGenerator> id_generator_; }; http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/ff105036/libminifi/src/processors/PutFile.cpp ---------------------------------------------------------------------- diff --git a/libminifi/src/processors/PutFile.cpp b/libminifi/src/processors/PutFile.cpp index a36f6d7..b0e4f15 100644 --- a/libminifi/src/processors/PutFile.cpp +++ b/libminifi/src/processors/PutFile.cpp @@ -28,17 +28,8 @@ #include <cstdio> #include <iostream> #include <memory> -#include <set> -#include <algorithm> #include <string> - -#include "core/logging/Logger.h" -#include "core/ProcessContext.h" -#include "core/Property.h" -#include "core/Relationship.h" -#include "io/BaseStream.h" -#include "io/DataStream.h" -#include "io/validation.h" +#include <set> namespace org { namespace apache { @@ -89,10 +80,6 @@ void PutFile::initialize() { } void PutFile::onSchedule(core::ProcessContext *context, core::ProcessSessionFactory *sessionFactory) { - if (!context->getProperty(Directory.getName(), directory_)) { - logger_->log_error("Directory attribute is missing or invalid"); - } - if (!context->getProperty(ConflictResolution.getName(), conflict_resolution_)) { logger_->log_error("Conflict Resolution Strategy attribute is missing or invalid"); } @@ -107,7 +94,8 @@ void PutFile::onSchedule(core::ProcessContext *context, core::ProcessSessionFact } void PutFile::onTrigger(core::ProcessContext *context, core::ProcessSession *session) { - if (IsNullOrEmpty(directory_) || IsNullOrEmpty(conflict_resolution_)) { + if (IsNullOrEmpty(conflict_resolution_)) { + logger_->log_error("Conflict resolution value is invalid"); context->yield(); return; } @@ -119,29 +107,41 @@ void PutFile::onTrigger(core::ProcessContext *context, core::ProcessSession *ses return; } + std::string directory; + + if (!context->getProperty(Directory.getName(), directory, flowFile)) { + logger_->log_error("Directory attribute is missing or invalid"); + } + + if (IsNullOrEmpty(directory)) { + logger_->log_error("Directory attribute evaluated to invalid value"); + session->transfer(flowFile, Failure); + return; + } + std::string filename; flowFile->getKeyedAttribute(FILENAME, filename); - std::string tmpFile = tmpWritePath(filename); + std::string tmpFile = tmpWritePath(filename, directory); logger_->log_info("PutFile using temporary file %s", tmpFile.c_str()); // Determine dest full file paths std::stringstream destFileSs; - destFileSs << directory_ << "/" << filename; + destFileSs << directory << "/" << filename; std::string destFile = destFileSs.str(); - logger_->log_info("PutFile writing file %s into directory %s", filename.c_str(), directory_.c_str()); + logger_->log_info("PutFile writing file %s into directory %s", filename.c_str(), directory.c_str()); // If file exists, apply conflict resolution strategy struct stat statResult; - if ((max_dest_files_ != -1) && (stat(directory_.c_str(), &statResult) == 0)) { + if ((max_dest_files_ != -1) && (stat(directory.c_str(), &statResult) == 0)) { // something exists at directory path if (S_ISDIR(statResult.st_mode)) { // it's a directory, count the files - DIR *myDir = opendir(directory_.c_str()); + DIR *myDir = opendir(directory.c_str()); if (!myDir) { - logger_->log_warn("Could not open %s", directory_.c_str()); + logger_->log_warn("Could not open %s", directory.c_str()); session->transfer(flowFile, Failure); return; } @@ -153,7 +153,7 @@ void PutFile::onTrigger(core::ProcessContext *context, core::ProcessSession *ses ct++; if (ct >= max_dest_files_) { logger_->log_warn("Routing to failure because the output directory %s has at least %u files, which exceeds the " - "configured max number of files", directory_.c_str(), max_dest_files_); + "configured max number of files", directory.c_str(), max_dest_files_); session->transfer(flowFile, Failure); closedir(myDir); return; @@ -170,24 +170,24 @@ void PutFile::onTrigger(core::ProcessContext *context, core::ProcessSession *ses conflict_resolution_.c_str()); if (conflict_resolution_ == CONFLICT_RESOLUTION_STRATEGY_REPLACE) { - putFile(session, flowFile, tmpFile, destFile); + putFile(session, flowFile, tmpFile, destFile, directory); } else if (conflict_resolution_ == CONFLICT_RESOLUTION_STRATEGY_IGNORE) { session->transfer(flowFile, Success); } else { session->transfer(flowFile, Failure); } } else { - putFile(session, flowFile, tmpFile, destFile); + putFile(session, flowFile, tmpFile, destFile, directory); } } -std::string PutFile::tmpWritePath(const std::string &filename) const { +std::string PutFile::tmpWritePath(const std::string &filename, const std::string &directory) const { char tmpFileUuidStr[37]; uuid_t tmpFileUuid; id_generator_->generate(tmpFileUuid); uuid_unparse_lower(tmpFileUuid, tmpFileUuidStr); std::stringstream tmpFileSs; - tmpFileSs << directory_; + tmpFileSs << directory; auto lastSeparatorPos = filename.find_last_of("/"); if (lastSeparatorPos == std::string::npos) { @@ -207,7 +207,37 @@ std::string PutFile::tmpWritePath(const std::string &filename) const { bool PutFile::putFile(core::ProcessSession *session, std::shared_ptr<FlowFileRecord> flowFile, const std::string &tmpFile, - const std::string &destFile) { + const std::string &destFile, + const std::string &destDir) { + struct stat dir_stat; + + if (stat(destDir.c_str(), &dir_stat) && try_mkdirs_) { + // Attempt to create directories in file's path + std::stringstream dir_path_stream; + + logger_->log_warn("Destination directory does not exist; will attempt to create: ", destDir); + size_t i = 0; + auto pos = destFile.find('/'); + + while (pos != std::string::npos) { + auto dir_path_component = destFile.substr(i, pos - i); + dir_path_stream << dir_path_component; + auto dir_path = dir_path_stream.str(); + + if (!dir_path_component.empty()) { + logger_->log_info("Attempting to create directory if it does not already exist: %s", dir_path); + mkdir(dir_path.c_str(), 0700); + dir_path_stream << '/'; + } else if (pos == 0) { + // Support absolute paths + dir_path_stream << '/'; + } + + i = pos + 1; + pos = destFile.find('/', pos + 1); + } + } + ReadCallback cb(tmpFile, destFile, try_mkdirs_); session->read(flowFile, &cb); @@ -234,71 +264,30 @@ PutFile::ReadCallback::ReadCallback(const std::string &tmp_file, int64_t PutFile::ReadCallback::process(std::shared_ptr<io::BaseStream> stream) { // Copy file contents into tmp file write_succeeded_ = false; - bool try_mkdirs = false; size_t size = 0; uint8_t buffer[1024]; - // Attempt writing file. After one failure, try to create parent directories if they don't already exist. - // This is done so that a stat syscall of the directory is not required on multiple file writes to a good dir, - // which is assumed to be a very common case. - while (!write_succeeded_) { - std::ofstream tmp_file_os(tmp_file_); + std::ofstream tmp_file_os(tmp_file_); - // Attempt to create directories in file's path - std::stringstream dir_path_stream; - - if (try_mkdirs) { - size_t i = 0; - auto pos = tmp_file_.find('/'); - while (pos != std::string::npos) { - auto dir_path_component = tmp_file_.substr(i, pos - i); - dir_path_stream << dir_path_component; - auto dir_path = dir_path_stream.str(); - - if (!dir_path_component.empty()) { - logger_->log_info("Attempting to create directory if it does not already exist: %s", dir_path); - mkdir(dir_path.c_str(), 0700); - dir_path_stream << '/'; - } + do { + int read = stream->read(buffer, 1024); - i = pos + 1; - pos = tmp_file_.find('/', pos + 1); - } + if (read < 0) { + return -1; } - do { - int read = stream->read(buffer, 1024); - - if (read < 0) { - return -1; - } - - if (read == 0) { - break; - } + if (read == 0) { + break; + } - tmp_file_os.write(reinterpret_cast<char *>(buffer), read); - size += read; - } while (size < stream->getSize()); + tmp_file_os.write(reinterpret_cast<char *>(buffer), read); + size += read; + } while (size < stream->getSize()); - tmp_file_os.close(); + tmp_file_os.close(); - if (tmp_file_os) { - write_succeeded_ = true; - } else { - if (try_mkdirs) { - // We already tried to create dirs, so give up - break; - } else { - if (try_mkdirs_) { - // This write failed; try creating the dir on another attempt - try_mkdirs = true; - } else { - // We've been instructed to not attempt to create dirs, so give up - break; - } - } - } + if (tmp_file_os) { + write_succeeded_ = true; } return size; http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/ff105036/libminifi/test/expression-language-tests/CMakeLists.txt ---------------------------------------------------------------------- diff --git a/libminifi/test/expression-language-tests/CMakeLists.txt b/libminifi/test/expression-language-tests/CMakeLists.txt new file mode 100644 index 0000000..d11b310 --- /dev/null +++ b/libminifi/test/expression-language-tests/CMakeLists.txt @@ -0,0 +1,40 @@ +# +# 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. +# + + +file(GLOB EXPRESSION_LANGUAGE_TESTS "*.cpp") + +SET(EXTENSIONS_TEST_COUNT 0) + +FOREACH(testfile ${EXPRESSION_LANGUAGE_TESTS}) + get_filename_component(testfilename "${testfile}" NAME_WE) + add_executable("${testfilename}" "${testfile}") + target_include_directories(${testfilename} BEFORE PRIVATE "${CMAKE_SOURCE_DIR}/extensions/expression-language") + createTests("${testfilename}") + target_link_libraries(${testfilename} ${CATCH_MAIN_LIB}) + if (APPLE) + target_link_libraries ("${testfilename}" -Wl,-all_load minifi-expression-language-extensions ) + else () + target_link_libraries ("${testfilename}" -Wl,--whole-archive minifi-expression-language-extensions -Wl,--no-whole-archive) + endif() + MATH(EXPR EXTENSIONS_TEST_COUNT "${EXTENSIONS_TEST_COUNT}+1") + add_test(NAME "${testfilename}" COMMAND "${testfilename}" WORKING_DIRECTORY ${TEST_DIR}) +ENDFOREACH() + +message("-- Finished building ${EXTENSIONS_TEST_COUNT} expression language related test file(s)...") http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/ff105036/libminifi/test/expression-language-tests/ExpressionLanguageTests.cpp ---------------------------------------------------------------------- diff --git a/libminifi/test/expression-language-tests/ExpressionLanguageTests.cpp b/libminifi/test/expression-language-tests/ExpressionLanguageTests.cpp new file mode 100644 index 0000000..c0dd7f2 --- /dev/null +++ b/libminifi/test/expression-language-tests/ExpressionLanguageTests.cpp @@ -0,0 +1,237 @@ +/** + * + * 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 <memory> +#include <string> + +#include "../TestBase.h" +#include <ExtractText.h> +#include <GetFile.h> +#include <PutFile.h> +#include <LogAttribute.h> + +namespace expression = org::apache::nifi::minifi::expression; + +class MockFlowFile : public core::FlowFile { + void releaseClaim(const std::shared_ptr<minifi::ResourceClaim> claim) override {} +}; + +TEST_CASE("Trivial static expression", "[expressionLanguageTestTrivialStaticExpr]") { // NOLINT + REQUIRE("a" == expression::make_static("a")({})); +} + +TEST_CASE("Text expression", "[expressionLanguageTestTextExpression]") { // NOLINT + auto expr = expression::compile("text"); + REQUIRE("text" == expr({})); +} + +TEST_CASE("Text expression with escaped dollar", "[expressionLanguageTestEscapedDollar]") { // NOLINT + auto expr = expression::compile("te$$xt"); + REQUIRE("te$xt" == expr({})); +} + +TEST_CASE("Attribute expression", "[expressionLanguageTestAttributeExpression]") { // NOLINT + auto flow_file = std::make_shared<MockFlowFile>(); + flow_file->addAttribute("attr_a", "__attr_value_a__"); + auto expr = expression::compile("text_before${attr_a}text_after"); + REQUIRE("text_before__attr_value_a__text_after" == expr({flow_file})); +} + +TEST_CASE("Multi-attribute expression", "[expressionLanguageTestMultiAttributeExpression]") { // NOLINT + auto flow_file = std::make_shared<MockFlowFile>(); + flow_file->addAttribute("attr_a", "__attr_value_a__"); + flow_file->addAttribute("attr_b", "__attr_value_b__"); + auto expr = expression::compile("text_before${attr_a}text_between${attr_b}text_after"); + REQUIRE("text_before__attr_value_a__text_between__attr_value_b__text_after" == expr({flow_file})); +} + +TEST_CASE("Multi-flowfile attribute expression", + "[expressionLanguageTestMultiFlowfileAttributeExpression]") { // NOLINT + auto expr = expression::compile("text_before${attr_a}text_after"); + + auto flow_file_a = std::make_shared<MockFlowFile>(); + flow_file_a->addAttribute("attr_a", "__flow_a_attr_value_a__"); + REQUIRE("text_before__flow_a_attr_value_a__text_after" == expr({flow_file_a})); + + auto flow_file_b = std::make_shared<MockFlowFile>(); + flow_file_b->addAttribute("attr_a", "__flow_b_attr_value_a__"); + REQUIRE("text_before__flow_b_attr_value_a__text_after" == expr({flow_file_b})); +} + +TEST_CASE("Attribute expression with whitespace", "[expressionLanguageTestAttributeExpressionWhitespace]") { // NOLINT + auto flow_file = std::make_shared<MockFlowFile>(); + flow_file->addAttribute("attr_a", "__attr_value_a__"); + auto expr = expression::compile("text_before${\n\tattr_a \r}text_after"); + REQUIRE("text_before__attr_value_a__text_after" == expr({flow_file})); +} + +TEST_CASE("Special characters expression", "[expressionLanguageTestSpecialCharactersExpression]") { // NOLINT + auto expr = expression::compile("text_before|{}()[],:;\\/*#'\" \t\r\n${attr_a}}()text_after"); + + auto flow_file_a = std::make_shared<MockFlowFile>(); + flow_file_a->addAttribute("attr_a", "__flow_a_attr_value_a__"); + REQUIRE("text_before|{}()[],:;\\/*#'\" \t\r\n__flow_a_attr_value_a__}()text_after" == expr({flow_file_a})); +} + +TEST_CASE("UTF-8 characters expression", "[expressionLanguageTestUTF8Expression]") { // NOLINT + auto expr = expression::compile("text_before¥£â¬Â¢â¡â¢â£â¤â¥â¦â§â¨â©âªâ«ââ®â¯â¹${attr_a}text_after"); + + auto flow_file_a = std::make_shared<MockFlowFile>(); + flow_file_a->addAttribute("attr_a", "__flow_a_attr_value_a__"); + REQUIRE("text_before¥£â¬Â¢â¡â¢â£â¤â¥â¦â§â¨â©âªâ«ââ®â¯â¹__flow_a_attr_value_a__text_after" == expr({flow_file_a})); +} + +TEST_CASE("UTF-8 characters attribute", "[expressionLanguageTestUTF8Attribute]") { // NOLINT + auto expr = expression::compile("text_before${attr_a}text_after"); + + auto flow_file_a = std::make_shared<MockFlowFile>(); + flow_file_a->addAttribute("attr_a", "__¥£â¬Â¢â¡â¢â£â¤â¥â¦â§â¨â©âªâ«ââ®â¯â¹__"); + REQUIRE("text_before__¥£â¬Â¢â¡â¢â£â¤â¥â¦â§â¨â©âªâ«ââ®â¯â¹__text_after" == expr({flow_file_a})); +} + +TEST_CASE("Single quoted attribute expression", "[expressionLanguageTestSingleQuotedAttributeExpression]") { // NOLINT + auto expr = expression::compile("text_before${'|{}()[],:;\\/*# \t\r\n$'}text_after"); + + auto flow_file_a = std::make_shared<MockFlowFile>(); + flow_file_a->addAttribute("|{}()[],:;\\/*# \t\r\n$", "__flow_a_attr_value_a__"); + REQUIRE("text_before__flow_a_attr_value_a__text_after" == expr({flow_file_a})); +} + +TEST_CASE("Double quoted attribute expression", "[expressionLanguageTestDoubleQuotedAttributeExpression]") { // NOLINT + auto expr = expression::compile("text_before${\"|{}()[],:;\\/*# \t\r\n$\"}text_after"); + + auto flow_file_a = std::make_shared<MockFlowFile>(); + flow_file_a->addAttribute("|{}()[],:;\\/*# \t\r\n$", "__flow_a_attr_value_a__"); + REQUIRE("text_before__flow_a_attr_value_a__text_after" == expr({flow_file_a})); +} + +TEST_CASE("Hostname function", "[expressionLanguageTestHostnameFunction]") { // NOLINT + auto expr = expression::compile("text_before${\n\t hostname ()\n\t }text_after"); + + char hostname[1024]; + hostname[1023] = '\0'; + gethostname(hostname, 1023); + std::string expected("text_before"); + expected.append(hostname); + expected.append("text_after"); + + auto flow_file_a = std::make_shared<MockFlowFile>(); + REQUIRE(expected == expr({flow_file_a})); +} + +TEST_CASE("ToUpper function", "[expressionLanguageTestToUpperFunction]") { // NOLINT + auto expr = expression::compile(R"(text_before${ + attr_a : toUpper() + }text_after)"); + auto flow_file_a = std::make_shared<MockFlowFile>(); + flow_file_a->addAttribute("attr_a", "__flow_a_attr_value_a__"); + REQUIRE("text_before__FLOW_A_ATTR_VALUE_A__text_after" == expr({flow_file_a})); +} + +TEST_CASE("GetFile PutFile dynamic attribute", "[expressionLanguageTestGetFilePutFileDynamicAttribute]") { // NOLINT + TestController testController; + + LogTestController::getInstance().setTrace<TestPlan>(); + LogTestController::getInstance().setTrace<processors::PutFile>(); + LogTestController::getInstance().setTrace<processors::ExtractText>(); + LogTestController::getInstance().setTrace<processors::GetFile>(); + LogTestController::getInstance().setTrace<processors::PutFile>(); + LogTestController::getInstance().setTrace<processors::LogAttribute>(); + + auto plan = testController.createPlan(); + auto repo = std::make_shared<TestRepository>(); + + std::string in_dir("/tmp/gt.XXXXXX"); + REQUIRE(testController.createTempDirectory(&in_dir[0]) != nullptr); + + std::string in_file(in_dir); + in_file.append("/file"); + + std::string out_dir("/tmp/gt.XXXXXX"); + REQUIRE(testController.createTempDirectory(&out_dir[0]) != nullptr); + + std::string out_file(out_dir); + out_file.append("/extracted_attr/file"); + + // Build MiNiFi processing graph + auto get_file = plan->addProcessor( + "GetFile", + "GetFile"); + plan->setProperty( + get_file, + processors::GetFile::Directory.getName(), in_dir); + plan->setProperty( + get_file, + processors::GetFile::KeepSourceFile.getName(), + "false"); + plan->addProcessor( + "LogAttribute", + "LogAttribute", + core::Relationship("success", "description"), + true); + auto extract_text = plan->addProcessor( + "ExtractText", + "ExtractText", + core::Relationship("success", "description"), + true); + plan->setProperty( + extract_text, + processors::ExtractText::Attribute.getName(), "extracted_attr_name"); + plan->addProcessor( + "LogAttribute", + "LogAttribute", + core::Relationship("success", "description"), + true); + auto put_file = plan->addProcessor( + "PutFile", + "PutFile", + core::Relationship("success", "description"), + true); + plan->setProperty( + put_file, + processors::PutFile::Directory.getName(), + out_dir + "/${extracted_attr_name}"); + plan->setProperty( + put_file, + processors::PutFile::ConflictResolution.getName(), + processors::PutFile::CONFLICT_RESOLUTION_STRATEGY_REPLACE); + plan->setProperty( + put_file, + processors::PutFile::CreateDirs.getName(), + "true"); + + // Write test input + { + std::ofstream in_file_stream(in_file); + in_file_stream << "extracted_attr"; + } + + plan->runNextProcessor(); // GetFile + plan->runNextProcessor(); // Log + plan->runNextProcessor(); // ExtractText + plan->runNextProcessor(); // Log + plan->runNextProcessor(); // PutFile + + // Verify output + { + std::stringstream output_str; + std::ifstream out_file_stream(out_file); + output_str << out_file_stream.rdbuf(); + REQUIRE("extracted_attr" == output_str.str()); + } +} http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/ff105036/libminifi/test/unit/PutFileTests.cpp ---------------------------------------------------------------------- diff --git a/libminifi/test/unit/PutFileTests.cpp b/libminifi/test/unit/PutFileTests.cpp index fa3d185..711c9c8 100644 --- a/libminifi/test/unit/PutFileTests.cpp +++ b/libminifi/test/unit/PutFileTests.cpp @@ -315,6 +315,6 @@ TEST_CASE("PutFileTestFileExistsReplace", "[getfileputpfile]") { TEST_CASE("Test generation of temporary write path", "[putfileTmpWritePath]") { auto processor = std::make_shared<org::apache::nifi::minifi::processors::PutFile>("processorname"); - REQUIRE(processor->tmpWritePath("a/b/c").substr(1, strlen("a/b/.c")) == "a/b/.c"); + REQUIRE(processor->tmpWritePath("a/b/c", "").substr(1, strlen("a/b/.c")) == "a/b/.c"); }
