This is an automated email from the ASF dual-hosted git repository.
phrocker pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
The following commit(s) were added to refs/heads/master by this push:
new 81d552b MINIFICPP-694 - Ensure that processor consist of proper phases
81d552b is described below
commit 81d552b0fb1ff02f2cb5e81f2b9b114561175277
Author: Arpad Boda <[email protected]>
AuthorDate: Tue Apr 16 15:46:53 2019 +0200
MINIFICPP-694 - Ensure that processor consist of proper phases
This closes #536.
Signed-off-by: Marc Parisi <[email protected]>
---
nanofi/include/api/nanofi.h | 22 ++++++++++++++++--
nanofi/include/core/cstructs.h | 5 +++-
nanofi/include/cxx/CallbackProcessor.h | 15 ++++++++----
nanofi/include/cxx/Plan.h | 12 ++++++----
nanofi/src/api/nanofi.cpp | 4 ++--
nanofi/src/cxx/CallbackProcessor.cpp | 10 ++++++--
nanofi/src/cxx/Plan.cpp | 42 +++++++++++++++++++++++-----------
nanofi/tests/CAPITests.cpp | 12 ++++++++--
8 files changed, 91 insertions(+), 31 deletions(-)
diff --git a/nanofi/include/api/nanofi.h b/nanofi/include/api/nanofi.h
index 7b2cb18..c014030 100644
--- a/nanofi/include/api/nanofi.h
+++ b/nanofi/include/api/nanofi.h
@@ -379,13 +379,31 @@ int8_t remove_attribute(flow_file_record*, const char *
key);
int transmit_flowfile(flow_file_record *, nifi_instance *);
+
+/****
+ * ##################################################################
+ * API functions for user-defined processor
+ * ##################################################################
+ */
+
+typedef struct {
+ const char * name;
+ ontrigger_callback * ontr_cb;
+ onschedule_callback * onsc_cb;
+} custom_processor_args;
+
+
/**
* Adds a custom processor for later instantiation
* @param name name of the processor
- * @param logic the callback to be invoked when the processor is triggered
+ * @param in the name and the callbacks used for the processor
+ * @attention it's recommended to use this function via the variadic arg
macro: the caller doesn't need to create the
+ * parameter struct and callback arguments can be optional.
* @return 0 on success, -1 otherwise (name already in use for eg.)
**/
-int add_custom_processor(const char * name, processor_logic* logic);
+int var_add_custom_processor(custom_processor_args in);
+
+#define add_custom_processor(...)
var_add_custom_processor((custom_processor_args){__VA_ARGS__});
/**
* Removes a custom processor
diff --git a/nanofi/include/core/cstructs.h b/nanofi/include/core/cstructs.h
index c137cf8..7d258f7 100644
--- a/nanofi/include/core/cstructs.h
+++ b/nanofi/include/core/cstructs.h
@@ -126,7 +126,10 @@ typedef enum FS {
ROLLBACK
} FailureStrategy;
-typedef void (processor_logic)(processor_session*, processor_context *);
+typedef void (ontrigger_callback)(processor_session*, processor_context *);
+typedef void (onschedule_callback)(processor_context *);
+
+typedef ontrigger_callback processor_logic;
typedef struct file_buffer {
uint8_t * buffer;
diff --git a/nanofi/include/cxx/CallbackProcessor.h
b/nanofi/include/cxx/CallbackProcessor.h
index 21e8200..62eaa68 100644
--- a/nanofi/include/cxx/CallbackProcessor.h
+++ b/nanofi/include/cxx/CallbackProcessor.h
@@ -54,7 +54,8 @@ class CallbackProcessor : public core::Processor {
*/
CallbackProcessor(std::string name, utils::Identifier uuid =
utils::Identifier())
: Processor(name, uuid),
- callback_(nullptr),
+ ontrigger_callback_({}),
+ onschedule_callback_({}),
objref_(nullptr),
logger_(logging::LoggerFactory<CallbackProcessor>::getLogger()) {
}
@@ -67,12 +68,15 @@ class CallbackProcessor : public core::Processor {
public:
- void setCallback(void *obj,std::function<void(core::ProcessSession*,
core::ProcessContext *context)> ontrigger_callback) {
+ void setCallback(void *obj, std::function<void(core::ProcessSession*,
core::ProcessContext *context)> ontrigger_callback,
+ std::function<void(core::ProcessContext *context)>
onschedule_callback = {}) {
objref_ = obj;
- callback_ = ontrigger_callback;
+ ontrigger_callback_ = ontrigger_callback;
+ onschedule_callback_ = onschedule_callback;
}
- // OnTrigger method, implemented by NiFi CallbackProcessor
+ virtual void onSchedule(core::ProcessContext *context,
core::ProcessSessionFactory *sessionFactory);
+ // OnTrigger method, implemented by MiNiFi CallbackProcessor
virtual void onTrigger(core::ProcessContext *context, core::ProcessSession
*session); // override;
// Initialize, over write by NiFi CallbackProcessor
virtual void initialize(); // override;
@@ -83,7 +87,8 @@ class CallbackProcessor : public core::Processor {
protected:
void *objref_;
- std::function<void(core::ProcessSession*, core::ProcessContext *context)>
callback_;
+ std::function<void(core::ProcessSession*, core::ProcessContext *context)>
ontrigger_callback_;
+ std::function<void(core::ProcessContext *context)> onschedule_callback_;
private:
// Logger
std::shared_ptr<logging::Logger> logger_;
diff --git a/nanofi/include/cxx/Plan.h b/nanofi/include/cxx/Plan.h
index 87e9642..d988810 100644
--- a/nanofi/include/cxx/Plan.h
+++ b/nanofi/include/cxx/Plan.h
@@ -96,7 +96,9 @@ class ExecutionPlan {
std::shared_ptr<core::Processor> addSimpleCallback(void *,
std::function<void(core::ProcessSession*)>);
- std::shared_ptr<core::Processor> addCallback(void *obj,
std::function<void(core::ProcessSession*, core::ProcessContext *)> fp);
+ std::shared_ptr<core::Processor> addCallback(void *obj,
+ std::function<void(core::ProcessSession*, core::ProcessContext *)>
ontrigger_callback,
+ std::function<void(core::ProcessContext *)> onschedule_callback = {});
std::shared_ptr<core::Processor> addProcessor(const
std::shared_ptr<core::Processor> &processor, const std::string &name,
core::Relationship
relationship = core::Relationship("success", "description"),
@@ -147,7 +149,9 @@ class ExecutionPlan {
static std::shared_ptr<core::Processor> createProcessor(const std::string
&processor_name, const std::string &name);
- static std::shared_ptr<core::Processor> createCallback(void *obj,
std::function<void(core::ProcessSession*, core::ProcessContext *)> fp);
+ static std::shared_ptr<core::Processor> createCallback(void *obj,
+ std::function<void(core::ProcessSession*, core::ProcessContext *)>
ontrigger_callback,
+ std::function<void(core::ProcessContext *)> onschedule_callback = {});
static std::shared_ptr<ExecutionPlan> getPlan(const std::string& uuid) {
auto it = proc_plan_map_.find(uuid);
@@ -166,7 +170,7 @@ class ExecutionPlan {
return proc_plan_map_.size();
}
- static bool addCustomProcessor(const char * name, processor_logic* logic);
+ static bool addCustomProcessor(custom_processor_args);
static int deleteCustomProcessor(const char * name);
@@ -234,7 +238,7 @@ class ExecutionPlan {
std::shared_ptr<logging::Logger> logger_;
std::shared_ptr<FailureHandler> failure_handler_;
static std::unordered_map<std::string, std::shared_ptr<ExecutionPlan>>
proc_plan_map_;
- static std::map<std::string, processor_logic*> custom_processors;
+ static std::map<std::string, custom_processor_args> custom_processors;
};
#endif /* LIBMINIFI_CAPI_PLAN_H_ */
diff --git a/nanofi/src/api/nanofi.cpp b/nanofi/src/api/nanofi.cpp
index a7a37df..71bb50b 100644
--- a/nanofi/src/api/nanofi.cpp
+++ b/nanofi/src/api/nanofi.cpp
@@ -686,8 +686,8 @@ int transfer(processor_session* session, flow *flow, const
char *rel) {
return 0;
}
-int add_custom_processor(const char * name, processor_logic* logic) {
- return ExecutionPlan::addCustomProcessor(name, logic) ? 0 : -1;
+int var_add_custom_processor(custom_processor_args in) {
+ return ExecutionPlan::addCustomProcessor(in) ? 0 : -1;
}
int delete_custom_processor(const char * name) {
diff --git a/nanofi/src/cxx/CallbackProcessor.cpp
b/nanofi/src/cxx/CallbackProcessor.cpp
index c527f86..fbe48d5 100644
--- a/nanofi/src/cxx/CallbackProcessor.cpp
+++ b/nanofi/src/cxx/CallbackProcessor.cpp
@@ -33,9 +33,15 @@ void CallbackProcessor::initialize() {
setSupportedRelationships(relationships);
}
+void CallbackProcessor::onSchedule(core::ProcessContext *context,
core::ProcessSessionFactory *sessionFactory){
+ if(onschedule_callback_ != nullptr) {
+ onschedule_callback_(context);
+ }
+}
+
void CallbackProcessor::onTrigger(core::ProcessContext *context,
core::ProcessSession *session) {
- if (callback_ != nullptr) {
- callback_(session, context);
+ if (ontrigger_callback_ != nullptr) {
+ ontrigger_callback_(session, context);
}
}
diff --git a/nanofi/src/cxx/Plan.cpp b/nanofi/src/cxx/Plan.cpp
index 242b03b..0be7eea 100644
--- a/nanofi/src/cxx/Plan.cpp
+++ b/nanofi/src/cxx/Plan.cpp
@@ -25,7 +25,7 @@
std::shared_ptr<utils::IdGenerator> ExecutionPlan::id_generator_ =
utils::IdGenerator::getIdGenerator();
std::unordered_map<std::string, std::shared_ptr<ExecutionPlan>>
ExecutionPlan::proc_plan_map_ = {};
-std::map<std::string, processor_logic*> ExecutionPlan::custom_processors = {};
+std::map<std::string, custom_processor_args> ExecutionPlan::custom_processors
= {};
ExecutionPlan::ExecutionPlan(std::shared_ptr<core::ContentRepository>
content_repo, std::shared_ptr<core::Repository> flow_repo,
std::shared_ptr<core::Repository> prov_repo)
: content_repo_(content_repo),
@@ -52,12 +52,15 @@ std::shared_ptr<core::Processor>
ExecutionPlan::addSimpleCallback(void *obj, std
return addCallback(obj, simple_func_wrapper);
}
-std::shared_ptr<core::Processor> ExecutionPlan::addCallback(void *obj,
std::function<void(core::ProcessSession*, core::ProcessContext *)> fp) {
+std::shared_ptr<core::Processor> ExecutionPlan::addCallback(void *obj,
+ std::function<void(core::ProcessSession*, core::ProcessContext *)>
ontrigger_callback,
+ std::function<void(core::ProcessContext *)> onschedule_callback) {
+
if (finalized) {
return nullptr;
}
- auto proc = createCallback(obj, fp);
+ auto proc = createCallback(obj, ontrigger_callback, onschedule_callback);
if (!proc)
return nullptr;
@@ -249,11 +252,21 @@ std::shared_ptr<core::Processor>
ExecutionPlan::createProcessor(const std::strin
auto custom_proc = custom_processors.find(processor_name);
if(custom_proc != custom_processors.end()) {
- auto c_func = custom_proc->second;
- auto wrapper_func = [c_func](core::ProcessSession * session,
core::ProcessContext * context) {
- return c_func(reinterpret_cast<processor_session*>(session),
reinterpret_cast<processor_context*>(context));
+ auto ontrigger_c_func = custom_proc->second.ontr_cb;
+ auto onschedule_c_func = custom_proc->second.onsc_cb;
+ auto ontrigger_wrapper_func = [ontrigger_c_func](core::ProcessSession *
session, core::ProcessContext * context) {
+ if(ontrigger_c_func) {
+ ontrigger_c_func(reinterpret_cast<processor_session *>(session),
+ reinterpret_cast<processor_context *>(context));
+ }
+ };
+ auto onschedule_wrapper_func = [onschedule_c_func](core::ProcessContext *
context) {
+ if (onschedule_c_func) {
+ onschedule_c_func(reinterpret_cast<processor_context*>(context));
+ }
};
- return createCallback(nullptr, wrapper_func);
+
+ return createCallback(nullptr, ontrigger_wrapper_func,
onschedule_wrapper_func);
}
@@ -267,13 +280,16 @@ std::shared_ptr<core::Processor>
ExecutionPlan::createProcessor(const std::strin
return processor;
}
-std::shared_ptr<core::Processor> ExecutionPlan::createCallback(void *obj,
std::function<void(core::ProcessSession*, core::ProcessContext *)> fp) {
+std::shared_ptr<core::Processor> ExecutionPlan::createCallback(void *obj,
+ std::function<void(core::ProcessSession*, core::ProcessContext *)>
ontrigger_callback,
+ std::function<void(core::ProcessContext *)> onschedule_callback) {
+
auto ptr = createProcessor(CallbackProcessorName, CallbackProcessorName);
if (!ptr)
return nullptr;
std::shared_ptr<processors::CallbackProcessor> processor =
std::static_pointer_cast<processors::CallbackProcessor>(ptr);
- processor->setCallback(obj, fp);
+ processor->setCallback(obj, ontrigger_callback, onschedule_callback);
return ptr;
}
@@ -323,15 +339,15 @@ bool ExecutionPlan::setFailureStrategy(FailureStrategy
start) {
return true;
}
-bool ExecutionPlan::addCustomProcessor(const char * name, processor_logic*
logic) {
- if(CallbackProcessorName == name) {
+bool ExecutionPlan::addCustomProcessor(custom_processor_args in) {
+ if(CallbackProcessorName == in.name) {
return false; // This name cannot be registered
}
- if (custom_processors.count(name) > 0 ) {
+ if (custom_processors.count(in.name) > 0 ) {
return false; // Already exists
}
- custom_processors[name] = logic;
+ custom_processors[in.name] = in;
return true;
}
diff --git a/nanofi/tests/CAPITests.cpp b/nanofi/tests/CAPITests.cpp
index 52a793c..ec2f2d5 100644
--- a/nanofi/tests/CAPITests.cpp
+++ b/nanofi/tests/CAPITests.cpp
@@ -43,6 +43,8 @@ static nifi_instance *create_instance_obj(const char *name =
"random_instance")
static int failure_count = 0;
+static int custom_onschedule_count = 0;
+
void failure_counter(flow_file_record * fr) {
failure_count++;
REQUIRE(get_attribute_quantity(fr) > 0);
@@ -54,7 +56,11 @@ void big_failure_counter(flow_file_record * fr) {
free_flowfile(fr);
}
-void custom_processor_logic(processor_session * ps, processor_context * ctx) {
+void custom_onschedule_logic(processor_context* ctx) {
+ custom_onschedule_count++;
+}
+
+void custom_ontrigger_logic(processor_session *ps, processor_context *ctx) {
flow_file_record * ffr = get(ps, ctx);
REQUIRE(ffr != nullptr);
uint8_t * buffer = (uint8_t*)malloc(ffr->size* sizeof(uint8_t));
@@ -430,7 +436,7 @@ TEST_CASE("Test custom processor", "[TestCutomProcessor]") {
create_testfile_for_getfile(sourcedir);
- add_custom_processor("myproc", custom_processor_logic);
+ add_custom_processor("myproc", custom_ontrigger_logic,
custom_onschedule_logic);
auto instance = create_instance_obj();
REQUIRE(instance != nullptr);
@@ -449,6 +455,8 @@ TEST_CASE("Test custom processor", "[TestCutomProcessor]") {
flow_file_record *record = get_next_flow_file(instance, test_flow);
+ REQUIRE(custom_onschedule_count > 0);
+
REQUIRE(record != nullptr);
}
