Repository: nifi-minifi-cpp Updated Branches: refs/heads/master 7546b1daa -> 627d8fe2c
MINIFICPP-682 - C API: provide functions to create custom processors MINIFICPP-682 - add doxygen style comments to C API MINIFICPP-682 - fix leaking ExecutionPlans MINIFICPP-682 - Remove usage of deprecated functions MINIFICPP-682 - review comment fixes This closes #448. Signed-off-by: Marc Parisi <[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/627d8fe2 Tree: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/tree/627d8fe2 Diff: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/diff/627d8fe2 Branch: refs/heads/master Commit: 627d8fe2c9f6658fa3fd1ef6ccfda22998248455 Parents: 7546b1d Author: Arpad Boda <[email protected]> Authored: Mon Nov 26 08:50:28 2018 +0100 Committer: Marc Parisi <[email protected]> Committed: Thu Dec 6 10:11:05 2018 -0500 ---------------------------------------------------------------------- nanofi/examples/terminate_handler.c | 4 +- nanofi/include/api/nanofi.h | 271 +++++++++++++++++++++++++--- nanofi/include/core/cstructs.h | 14 ++ nanofi/include/core/cxxstructs.h | 4 + nanofi/include/cxx/CallbackProcessor.h | 17 +- nanofi/include/cxx/Plan.h | 15 +- nanofi/src/api/nanofi.cpp | 96 +++++++--- nanofi/src/cxx/CallbackProcessor.cpp | 16 +- nanofi/src/cxx/Plan.cpp | 61 ++++++- nanofi/tests/CAPITests.cpp | 82 +++++++-- 10 files changed, 495 insertions(+), 85 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/627d8fe2/nanofi/examples/terminate_handler.c ---------------------------------------------------------------------- diff --git a/nanofi/examples/terminate_handler.c b/nanofi/examples/terminate_handler.c index 1d5150d..8b8989b 100644 --- a/nanofi/examples/terminate_handler.c +++ b/nanofi/examples/terminate_handler.c @@ -40,7 +40,9 @@ int main(int argc, char **argv) { nifi_instance *instance = create_instance("random instance", &port); - flow *new_flow = create_flow(instance, "GenerateFlowFile"); + flow *new_flow = create_new_flow(instance); + + processor *generate_proc = add_processor(new_flow, "GenerateFlowFile"); processor *put_proc = add_processor(new_flow, "PutFile"); http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/627d8fe2/nanofi/include/api/nanofi.h ---------------------------------------------------------------------- diff --git a/nanofi/include/api/nanofi.h b/nanofi/include/api/nanofi.h index e25a3a0..99eadf8 100644 --- a/nanofi/include/api/nanofi.h +++ b/nanofi/include/api/nanofi.h @@ -37,8 +37,18 @@ extern "C" { */ #define API_VERSION "0.02" +#define SUCCESS_RELATIONSHIP "success" +#define FAILURE_RELATIONSHIP "failure" + +/** + * Enables logging (disabled by default) + **/ void enable_logging(); +/** + * Sets terminate callback. The callback is executed upon termination (undhandled exception in C++ backend) + * @param terminate_callback the callback to execute + **/ void set_terminate_callback(void (*terminate_callback)()); /**** @@ -47,11 +57,27 @@ void set_terminate_callback(void (*terminate_callback)()); * ################################################################## */ +/** + * Creates a new MiNiFi instance + * @param url remote URL the instance connects to + * @param port remote port the instance connects to + * @return pointer to the new instance + **/ nifi_instance *create_instance(const char *url, nifi_port *port); -void initialize_instance(nifi_instance *); +/** + * Initialize remote connection of instance for transfers + * @param instance + **/ +void initialize_instance(nifi_instance * instance); -void free_instance(nifi_instance*); +/** + * Frees instance + * @attention Any action on flows that belong to the freed instance are undefined after this is done! + * It's recommended to free all flows before freeing the instance. + * @param instance instance to be freed + **/ +void free_instance(nifi_instance * instance); /**** * ################################################################## @@ -68,60 +94,171 @@ typedef int c2_start_callback(char *); void enable_async_c2(nifi_instance *, C2_Server *, c2_stop_callback *, c2_start_callback *, c2_update_callback *); +/** + * Creates a new, empty flow + * @param instance the instance new flow will belong to + * @return a pointer to the created flow + **/ +flow *create_new_flow(nifi_instance * instance); -uint8_t run_processor(const processor *processor); - -flow *create_new_flow(nifi_instance *); -flow *create_flow(nifi_instance *, const char *); +/** + * Creates new flow and adds the first processor in case a valid name is provided + * @deprecated as there is no proper indication of processor adding errors, + * usage of "create_new_flow" and "add_processor is recommended instead + * @param instance the instance new flow will belong to + * @param first_processor name of the first processor to be instanciated + * @attention in case first processor is empty or doesn't name any existing processor, an empty flow is returned. + * @return a pointer to the created flow + **/ +DEPRECATED flow *create_flow(nifi_instance * instance, const char * first_processor); +/** + * Add a getfile processor to "parent" flow. + * Creates new flow in instance in case "parent" is nullptr + * @param instance the instance the flow belongs to + * @param parent the flow to be extended with a new getfile processor + * @param c configuration of the new processor + * @return parent in case it wasn't null, otherwise a pointer to a new flow + */ flow *create_getfile(nifi_instance *instance, flow *parent, GetFileConfig *c); -processor *add_processor(flow *, const char *); +/** + * Extend a flow with a new processor + * @param flow the flow to be extended with the new processor + * @param name name of the new processor + * @return pointer to the new processor or nullptr in case it cannot be instantiated (wrong name?) + **/ +processor *add_processor(flow * flow, const char * name); processor *add_python_processor(flow *, void (*ontrigger_callback)(processor_session *session)); -standalone_processor *create_processor(const char *); +/** + * Create a standalone instance of the given processor. + * Standalone instances can be invoked without having an instance/flow that contains them. + * @param name the name of the processor to instanciate + * @return pointer to the new processor or nullptr in case it cannot be instantiated (wrong name?) + **/ +standalone_processor *create_processor(const char * name); -void free_standalone_processor(standalone_processor*); +/** + * Free a standalone processor + * @param processor the processor to be freed + */ +void free_standalone_processor(standalone_processor* processor); /** -* Register your callback to received flow files that the flow failed to process -* The flow file ownership is transferred to the caller! -* The first callback should be registered before the flow is used. Can be changed later during runtime. -*/ + * Register your callback to received flow files that the flow failed to process + * @attention The flow file ownership is transferred to the callback! + * @attention The first callback should be registered before the flow is used. Can be changed later during runtime. + * @param flow flow the callback belongs to + * @param onerror_callback callback to execute in case of failure + * @return 0 in case of success, -1 otherwise (flow is already in use) + **/ int add_failure_callback(flow *flow, void (*onerror_callback)(flow_file_record*)); - /** -* Set failure strategy. Please use the enum defined in cstructs.h -* Return values: 0 (success), -1 (strategy cannot be set - no failure callback added?) -* Can be changed runtime. -* The defailt strategy is AS IS. -*/ + * Set failure strategy. Please use the enum defined in cstructs.h + * Can be changed runtime. + * The default strategy is AS IS. + * @param flow the flow to set strategy for + * @param strategy the strategy to be set + * @return 0 (success), -1 (strategy cannot be set - no failure callback added?) + **/ int set_failure_strategy(flow *flow, FailureStrategy strategy); -int set_property(processor *, const char *, const char *); +/** + * Set property for a processor + * @param processor the processor the property is set for + * @param name name of the property + * @param value value of the property + * @return 0 in case of success, -1 otherwise (the processor doesn't support such property) + **/ +int set_property(processor * processor, const char * name, const char * value); -int set_standalone_property(standalone_processor*, const char*, const char *); +/** + * Set property for a standalone processor + * @param processor the processor the property is set for + * @param name name of the property + * @param value value of the property + * @return 0 in case of success, -1 otherwise (the processor doesn't support such property) + **/ +int set_standalone_property(standalone_processor * processor, const char * name, const char * value); -int set_instance_property(nifi_instance *instance, const char*, const char *); +/** + * Set property for an instance + * @param instance the instance the property is set for + * @param name name of the property + * @param value value of the property + * @return 0 in case of success, -1 otherwise. Always succeeds unless instance or name is nullptr/emtpy. + **/ +int set_instance_property(nifi_instance *instance, const char * name, const char * value); -int free_flow(flow *); +/** + * Get a property. Should be used in custom processor logic callbacks. + * @attention The returned value transfers ownership, it's the callers responsibility to free it! + * @param context the current processor context + * @param name name of the property + * @return null-terminated char* in case of success, nullptr otherwise + **/ +char * get_property(const processor_context * context, const char * name); +/** + * Free a flow + * @param flow the flow to free + * @attention All the processor in the flow are freed, too! Actions performed on freed processors are undefined! + * @return 0 in case of success, -1 otherwise. Always succeeds unless flow is nullptr. + **/ +int free_flow(flow * flow); + +/** + * Get the next flow file of the given flow + * @param instance the instance the flow belongs to + * @param flow the flow to get flowfile from + * @return a flow file record or nullptr in case no flowfile was generated by the flow + **/ flow_file_record *get_next_flow_file(nifi_instance *, flow *); -size_t get_flow_files(nifi_instance *, flow *, flow_file_record **, size_t); +/** + * Get all flow files of the given flow + * @param instance the instance the flow belongs to + * @param flow the flow to get flowfiles from + * @param flowfiles target area to copy the flowfiles to + * @param size the maximum number of flowfiles to copy to target (size of target) + * @return the number of flow files copies to target. Less or equal to size. + **/ +size_t get_flow_files(nifi_instance * instance, flow * flow, flow_file_record ** flowfiles, size_t size); flow_file_record *get(nifi_instance *,flow *, processor_session *); +/** + * Invoke a standalone processor without input data. + * The processor is expected to generate flow file. + * @return a flow file record or nullptr in case no flowfile was generated + **/ flow_file_record *invoke(standalone_processor* proc); +/** + * Invoke a standalone processor with input flow file + * @param input_ff input flow file, which can belong be the output of another processor or flow + * @return a flow file record or nullptr in case no flowfile was generated + **/ flow_file_record *invoke_ff(standalone_processor* proc, const flow_file_record *input_ff); +/** + * Invoke a standalone processor with file input + * @param path specifies the file system path of the input file + * @return a flow file record or nullptr in case no flowfile was generated + **/ flow_file_record *invoke_file(standalone_processor* proc, const char* path); -flow_file_record *invoke_chunck(standalone_processor* proc, uint8_t* buf, uint64_t); +/** + * Invoke a standalone processor with some in-memory data + * @param buf specifies the beginning of the input buffer + * @param size specifies the size of the buffer + * @return a flow file record or nullptr in case no flowfile was generated + **/ +flow_file_record *invoke_chunk(standalone_processor *proc, uint8_t *buf, uint64_t size); int transfer(processor_session* session, flow *flow, const char *rel); @@ -135,16 +272,62 @@ flow_file_record* create_ff_object(const char *file, const size_t len, const uin flow_file_record* create_ff_object_na(const char *file, const size_t len, const uint64_t size); -void free_flowfile(flow_file_record*); +/** + * Get incoming flow file. To be used in processor logic callbacks. + * @param session current processor session + * @param context current processor context + * @return a flow file record or nullptr in case there is none in the session + **/ +flow_file_record* get_flowfile(processor_session* session, processor_context* context); + + +/** + * Free flow file + * @param ff flow file + **/ +void free_flowfile(flow_file_record* ff); +/** + * Adds an attribute, fails in case there is already an attribute with the given key. + * @param ff flow file + * @param key name of attribute + * @param value location of value + * @size size size of the data pointed by "value" + * @return 0 in case of success, -1 otherwise (already existed) + **/ uint8_t add_attribute(flow_file_record*, const char *key, void *value, size_t size); -void update_attribute(flow_file_record*, const char *key, void *value, size_t size); +/** + * Updates an attribute (adds if it hasn't existed before) + * @param ff flow file + * @param key name of attribute + * @param value location of value + * @size size size of the data pointed by "value" + **/ +void update_attribute(flow_file_record* ff, const char *key, void *value, size_t size); -uint8_t get_attribute(flow_file_record *ff, attribute *caller_attribute); +/** + * Get the value of an attribute. Value and value size are written to parameter "caller_attribute" + * @param ff flow file + * @param caller_attribute attribute structure to provide name and get value, size + * @return 0 in case of success, -1 otherwise (no such attribute) + **/ +uint8_t get_attribute(const flow_file_record *ff, attribute *caller_attribute); + +/** + * Get the quantity of attributes + * @param ff flow file + * @return the number of attributes + **/ +int get_attribute_quantity(const flow_file_record *ff); -int get_attribute_qty(const flow_file_record* ff); +/** + * Copies all attributes of the flowfile that fits target. + * @param ff flow file + * @param target attribute set to copy to. target->size determines the maximum number of attributes copied + * @return the number of attributes copied, which is the minimum of attribute quantity and target size + **/ int get_all_attributes(const flow_file_record* ff, attribute_set *target); /** @@ -155,6 +338,11 @@ int get_all_attributes(const flow_file_record* ff, attribute_set *target); **/ int get_content(const flow_file_record* ff, uint8_t* target, int size); +/** + * Removes an attribute + * @param name name of the attribute + * @return 0 on success, -1 otherwise (doesn't exist) + **/ uint8_t remove_attribute(flow_file_record*, char *key); /**** @@ -165,6 +353,31 @@ uint8_t remove_attribute(flow_file_record*, char *key); int transmit_flowfile(flow_file_record *, nifi_instance *); +/** + * 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 + * @return 0 on success, -1 otherwise (name already in use for eg.) + **/ +int add_custom_processor(const char * name, processor_logic* logic); + +/** + * Removes a custom processor + * @param name name of the processor + * @return 0 on success, -1 otherwise (didn't exist) + **/ +int delete_custom_processor(const char * name); + +/** + * Transfers a flowfile to the given relationship + * This function is only to be used within processor logic callback + * @param ffr flow file to be transfered + * @param ps processor session the transfer happens within + * @param relationship name of the relationship ("success" and "failure" are supported currently) + * @return 0 on success, -1 otherwise (didn't exist) + **/ +int transfer_to_relationship(flow_file_record * ffr, processor_session * ps, const char * relationship); + /**** * ################################################################## * Persistence Operations http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/627d8fe2/nanofi/include/core/cstructs.h ---------------------------------------------------------------------- diff --git a/nanofi/include/core/cstructs.h b/nanofi/include/core/cstructs.h index e493166..d85c856 100644 --- a/nanofi/include/core/cstructs.h +++ b/nanofi/include/core/cstructs.h @@ -22,6 +22,14 @@ #include <stddef.h> #include <stdint.h> +#ifdef _MSC_VER +#define DEPRECATED __declspec(deprecated) +#elif defined(__GNUC__) | defined(__clang__) +#define DEPRECATED __attribute__((__deprecated__)) +#else +#define DEPRECATED +#endif + /** * NiFi Port struct */ @@ -71,6 +79,8 @@ typedef struct standalone_processor standalone_processor; typedef struct processor_session processor_session; +typedef struct processor_context processor_context; + /**** * ################################################################## * FLOWFILE OPERATIONS @@ -105,6 +115,8 @@ typedef struct { void *ffp; + uint8_t keepContent; + } flow_file_record; typedef struct flow flow; @@ -114,4 +126,6 @@ typedef enum FS { ROLLBACK } FailureStrategy; +typedef void (processor_logic)(processor_session*, processor_context *); + #endif /* LIBMINIFI_SRC_CAPI_CSTRUCTS_H_ */ http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/627d8fe2/nanofi/include/core/cxxstructs.h ---------------------------------------------------------------------- diff --git a/nanofi/include/core/cxxstructs.h b/nanofi/include/core/cxxstructs.h index dfa327c..d32b720 100644 --- a/nanofi/include/core/cxxstructs.h +++ b/nanofi/include/core/cxxstructs.h @@ -38,4 +38,8 @@ struct processor_session : public core::ProcessSession { using core::ProcessSession::ProcessSession; }; +struct processor_context : public core::ProcessContext { + using core::ProcessContext::ProcessContext; +}; + #endif //NIFI_MINIFI_CPP_CXXSTRUCTS_H http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/627d8fe2/nanofi/include/cxx/CallbackProcessor.h ---------------------------------------------------------------------- diff --git a/nanofi/include/cxx/CallbackProcessor.h b/nanofi/include/cxx/CallbackProcessor.h index 7cfcaf2..81c5546 100644 --- a/nanofi/include/cxx/CallbackProcessor.h +++ b/nanofi/include/cxx/CallbackProcessor.h @@ -46,6 +46,8 @@ namespace processors { // CallbackProcessor Class class CallbackProcessor : public core::Processor { public: + static core::Relationship Success; + static core::Relationship Failure; // Constructor /*! * Create a new processor @@ -65,24 +67,23 @@ class CallbackProcessor : public core::Processor { public: - void setCallback(void *obj,std::function<void(core::ProcessSession*)> ontrigger_callback) { + void setCallback(void *obj,std::function<void(core::ProcessSession*, core::ProcessContext *context)> ontrigger_callback) { objref_ = obj; callback_ = ontrigger_callback; } // OnTrigger method, implemented by NiFi CallbackProcessor - virtual void onTrigger(core::ProcessContext *context, core::ProcessSession *session); + virtual void onTrigger(core::ProcessContext *context, core::ProcessSession *session); // override; // Initialize, over write by NiFi CallbackProcessor - virtual void initialize() { - std::set<core::Relationship> relationships; - core::Relationship Success("success", "description"); - relationships.insert(Success); - setSupportedRelationships(relationships); + virtual void initialize(); // override; + + virtual bool supportsDynamicProperties() /*override*/ { + return true; } protected: void *objref_; - std::function<void(core::ProcessSession*)> callback_; + std::function<void(core::ProcessSession*, core::ProcessContext *context)> callback_; private: // Logger std::shared_ptr<logging::Logger> logger_; http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/627d8fe2/nanofi/include/cxx/Plan.h ---------------------------------------------------------------------- diff --git a/nanofi/include/cxx/Plan.h b/nanofi/include/cxx/Plan.h index e2cb827..96f858f 100644 --- a/nanofi/include/cxx/Plan.h +++ b/nanofi/include/cxx/Plan.h @@ -46,6 +46,8 @@ #include "core/reporting/SiteToSiteProvenanceReportingTask.h" #include "api/nanofi.h" +static const std::string CallbackProcessorName = "CallbackProcessor"; + using failure_callback_type = std::function<void(flow_file_record*)>; using content_repo_sptr = std::shared_ptr<core::ContentRepository>; @@ -70,7 +72,7 @@ namespace { auto path = claim->getContentFullPath(); auto ffr = create_ff_object_na(path.c_str(), path.length(), ff->getSize()); ffr->attributes = ff->getAttributesPtr(); - ffr->ffp = ff.get(); + ffr->ffp = static_cast<void*>(new std::shared_ptr<minifi::core::FlowFile>(ff)); auto content_repo_ptr = static_cast<std::shared_ptr<minifi::core::ContentRepository>*>(ffr->crp); *content_repo_ptr = cr_ptr; user_callback(ffr); @@ -92,7 +94,9 @@ class ExecutionPlan { explicit ExecutionPlan(std::shared_ptr<core::ContentRepository> content_repo, std::shared_ptr<core::Repository> flow_repo, std::shared_ptr<core::Repository> prov_repo); - std::shared_ptr<core::Processor> addCallback(void *, std::function<void(core::ProcessSession*)>); + 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> addProcessor(const std::shared_ptr<core::Processor> &processor, const std::string &name, core::Relationship relationship = core::Relationship("success", "description"), @@ -143,6 +147,8 @@ 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<ExecutionPlan> getPlan(const std::string& uuid) { auto it = proc_plan_map_.find(uuid); return it != proc_plan_map_.end() ? it->second : nullptr; @@ -160,6 +166,10 @@ class ExecutionPlan { return proc_plan_map_.size(); } + static bool addCustomProcessor(const char * name, processor_logic* logic); + + static int deleteCustomProcessor(const char * name); + protected: class FailureHandler { public: @@ -224,6 +234,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; }; #endif /* LIBMINIFI_CAPI_PLAN_H_ */ http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/627d8fe2/nanofi/src/api/nanofi.cpp ---------------------------------------------------------------------- diff --git a/nanofi/src/api/nanofi.cpp b/nanofi/src/api/nanofi.cpp index 290a0ec..78d1c09 100644 --- a/nanofi/src/api/nanofi.cpp +++ b/nanofi/src/api/nanofi.cpp @@ -27,6 +27,7 @@ #include "core/expect.h" #include "cxx/Instance.h" #include "cxx/Plan.h" +#include "cxx/CallbackProcessor.h" #include "ResourceClaim.h" #include "processors/GetFile.h" #include "core/logging/LoggerConfiguration.h" @@ -95,7 +96,6 @@ nifi_instance *create_instance(const char *url, nifi_port *port) { } standalone_processor *create_processor(const char *name) { - static int proc_counter = 0; auto ptr = ExecutionPlan::createProcessor(name, name); if (!ptr) { return nullptr; @@ -106,8 +106,7 @@ standalone_processor *create_processor(const char *name) { port.port_id = portnum; standalone_instance = create_instance("internal_standalone", &port); } - std::string flow_name = std::to_string(proc_counter++); - auto flow = create_flow(standalone_instance, flow_name.c_str()); + auto flow = create_new_flow(standalone_instance); std::shared_ptr<ExecutionPlan> plan(flow); plan->addProcessor(ptr, name); ExecutionPlan::addProcessorWithPlan(ptr->getUUIDStr(), plan); @@ -187,6 +186,7 @@ flow_file_record* create_flowfile(const char *file, const size_t len) { std::ifstream in(file, std::ifstream::ate | std::ifstream::binary); // set the size of the flow file. new_ff->size = in.tellg(); + new_ff->keepContent = 0; return new_ff; } @@ -212,6 +212,7 @@ flow_file_record* create_ff_object_na(const char *file, const size_t len, const // set the size of the flow file. new_ff->size = size; new_ff->crp = static_cast<void*>(new std::shared_ptr<minifi::core::ContentRepository>); + new_ff->keepContent = 0; return new_ff; } /** @@ -223,13 +224,16 @@ void free_flowfile(flow_file_record *ff) { return; } auto content_repo_ptr = static_cast<std::shared_ptr<minifi::core::ContentRepository>*>(ff->crp); - if (content_repo_ptr->get()) { + if (content_repo_ptr->get() && (ff->keepContent == 0)) { std::shared_ptr<minifi::ResourceClaim> claim = std::make_shared<minifi::ResourceClaim>(ff->contentLocation, *content_repo_ptr); (*content_repo_ptr)->remove(claim); } if (ff->ffp == nullptr) { auto map = static_cast<string_map*>(ff->attributes); delete map; + } else { + auto ff_sptr = reinterpret_cast<std::shared_ptr<core::FlowFile>*>(ff->ffp); + delete ff_sptr; } free(ff->contentLocation); free(ff); @@ -269,7 +273,7 @@ void update_attribute(flow_file_record *ff, const char *key, void *value, size_t * @param caller_attribute caller supplied object in which we will copy the data ptr * @return 0 if successful, -1 if the key does not exist */ -uint8_t get_attribute(flow_file_record * ff, attribute * caller_attribute) { +uint8_t get_attribute(const flow_file_record * ff, attribute * caller_attribute) { if (ff == nullptr) { return -1; } @@ -286,7 +290,7 @@ uint8_t get_attribute(flow_file_record * ff, attribute * caller_attribute) { return -1; } -int get_attribute_qty(const flow_file_record* ff) { +int get_attribute_quantity(const flow_file_record *ff) { if (ff == nullptr) { return 0; } @@ -370,15 +374,6 @@ int transmit_flowfile(flow_file_record *ff, nifi_instance *instance) { } flow * create_new_flow(nifi_instance * instance) { - auto minifi_instance_ref = static_cast<minifi::Instance*>(instance->instance_ptr); - flow * area = static_cast<flow*>(malloc(1*sizeof(flow))); - if(area == nullptr) { - return nullptr; - } - return new(area) flow(minifi_instance_ref->getContentRepository(), minifi_instance_ref->getNoOpRepository(), minifi_instance_ref->getNoOpRepository()); -} - -flow *create_flow(nifi_instance *instance, const char *first_processor) { if (nullptr == instance || nullptr == instance->instance_ptr) { return nullptr; } @@ -387,9 +382,15 @@ flow *create_flow(nifi_instance *instance, const char *first_processor) { if(area == nullptr) { return nullptr; } + flow *new_flow = new(area) flow(minifi_instance_ref->getContentRepository(), minifi_instance_ref->getNoOpRepository(), minifi_instance_ref->getNoOpRepository()); + return new_flow; +} - if (first_processor != nullptr && strlen(first_processor) > 0) { +flow *create_flow(nifi_instance *instance, const char *first_processor) { + auto new_flow = create_new_flow(instance); + + if(new_flow != nullptr && first_processor != nullptr && strlen(first_processor) > 0) { // automatically adds it with success new_flow->addProcessor(first_processor, first_processor); } @@ -403,13 +404,13 @@ processor *add_python_processor(flow *flow, void (*ontrigger_callback)(processor auto lambda = [ontrigger_callback](core::ProcessSession *ps) { ontrigger_callback(static_cast<processor_session*>(ps)); //Meh, sorry for this }; - auto proc = flow->addCallback(nullptr, lambda); + auto proc = flow->addSimpleCallback(nullptr, lambda); return static_cast<processor*>(proc.get()); } flow * create_getfile(nifi_instance * instance, flow * parent_flow, GetFileConfig * c) { static const std::string first_processor = "GetFile"; - flow *new_flow = parent_flow == nullptr ? create_flow(instance, nullptr) : parent_flow; + flow *new_flow = parent_flow == nullptr ? create_new_flow(instance) : parent_flow; // automatically adds it with success auto getFile = new_flow->addProcessor(first_processor, first_processor); @@ -460,17 +461,27 @@ int set_standalone_property(standalone_processor *proc, const char *name, const return -1; } +char * get_property(const processor_context * context, const char * name) { + std::string value; + if(!context->getDynamicProperty(name, value)) { + return nullptr; + } + size_t len = value.length(); + char * ret_val = (char*)malloc((len +1) * sizeof(char)); + strncpy(ret_val, value.data(), len); + ret_val[len] = '\0'; + return ret_val; +} + int free_flow(flow *flow) { if (flow == nullptr) return -1; + flow->~flow(); free(flow); return 0; } -flow_file_record* flowfile_to_record(std::shared_ptr<core::FlowFile> ff, ExecutionPlan* plan) { - if (ff == nullptr) { - return nullptr; - } +flow_file_record* flowfile_to_record(std::shared_ptr<core::FlowFile> ff, const std::shared_ptr<minifi::core::ContentRepository>& crp) { auto claim = ff->getResourceClaim(); if(claim == nullptr) { return nullptr; @@ -480,10 +491,30 @@ flow_file_record* flowfile_to_record(std::shared_ptr<core::FlowFile> ff, Executi claim->increaseFlowFileRecordOwnedCount(); auto path = claim->getContentFullPath(); auto ffr = create_ff_object_na(path.c_str(), path.length(), ff->getSize()); - ffr->ffp = ff.get(); + ffr->ffp = static_cast<void*>(new std::shared_ptr<core::FlowFile>(ff)); ffr->attributes = ff->getAttributesPtr(); auto content_repo_ptr = static_cast<std::shared_ptr<minifi::core::ContentRepository>*>(ffr->crp); - *content_repo_ptr = plan->getContentRepo(); + *content_repo_ptr = crp; + return ffr; +} + +flow_file_record* flowfile_to_record(std::shared_ptr<core::FlowFile> ff, ExecutionPlan* plan) { + if (ff == nullptr) { + return nullptr; + } + + return flowfile_to_record(ff, plan->getContentRepo()); +} + +flow_file_record* get_flowfile(processor_session* session, processor_context* context) { + auto ff = session->get(); + if(!ff) { + return nullptr; + } + + auto ffr = flowfile_to_record(ff, context->getContentRepository()); + // The content of the flow file must be kept in a processor logic + ffr->keepContent = 1; return ffr; } @@ -610,3 +641,20 @@ int transfer(processor_session* session, flow *flow, const char *rel) { session->transfer(ff, relationship); return 0; } + +int add_custom_processor(const char * name, processor_logic* logic) { + return ExecutionPlan::addCustomProcessor(name, logic) ? 0 : -1; +} + +int delete_custom_processor(const char * name) { + return ExecutionPlan::deleteCustomProcessor(name) - 1; +} + +int transfer_to_relationship(flow_file_record * ffr, processor_session * ps, const char * relationship) { + if(ffr == nullptr || ffr->ffp == nullptr || ps == nullptr || relationship == nullptr || strlen(relationship) == 0) { + return -1; + } + auto ff_sptr = reinterpret_cast<std::shared_ptr<core::FlowFile>*>(ffr->ffp); + ps->transfer(*ff_sptr, core::Relationship(relationship, "desc")); + return 0; +} http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/627d8fe2/nanofi/src/cxx/CallbackProcessor.cpp ---------------------------------------------------------------------- diff --git a/nanofi/src/cxx/CallbackProcessor.cpp b/nanofi/src/cxx/CallbackProcessor.cpp index 013ec47..c527f86 100644 --- a/nanofi/src/cxx/CallbackProcessor.cpp +++ b/nanofi/src/cxx/CallbackProcessor.cpp @@ -23,10 +23,20 @@ namespace nifi { namespace minifi { namespace processors { +core::Relationship CallbackProcessor::Success("success", "All files are routed to success"); +core::Relationship CallbackProcessor::Failure("failure", "Failed files (based on callback logic) are transferred to failure"); + +void CallbackProcessor::initialize() { + std::set<core::Relationship> relationships; + relationships.insert(Success); + relationships.insert(Failure); + setSupportedRelationships(relationships); +} + void CallbackProcessor::onTrigger(core::ProcessContext *context, core::ProcessSession *session) { - if (callback_ != nullptr) { - callback_(session); - } + if (callback_ != nullptr) { + callback_(session, context); + } } } /* namespace processors */ http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/627d8fe2/nanofi/src/cxx/Plan.cpp ---------------------------------------------------------------------- diff --git a/nanofi/src/cxx/Plan.cpp b/nanofi/src/cxx/Plan.cpp index b2b4690..2ab464c 100644 --- a/nanofi/src/cxx/Plan.cpp +++ b/nanofi/src/cxx/Plan.cpp @@ -25,6 +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 = {}; 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), @@ -41,19 +42,27 @@ ExecutionPlan::ExecutionPlan(std::shared_ptr<core::ContentRepository> content_re * Add a callback to obtain and pass processor session to a generated processor * */ -std::shared_ptr<core::Processor> ExecutionPlan::addCallback(void *obj, std::function<void(core::ProcessSession*)> fp) { +std::shared_ptr<core::Processor> ExecutionPlan::addSimpleCallback(void *obj, std::function<void(core::ProcessSession*)> fp) { if (finalized) { return nullptr; } - auto ptr = createProcessor("CallbackProcessor", "CallbackProcessor"); - if (!ptr) + auto simple_func_wrapper = [fp](core::ProcessSession *session, core::ProcessContext *context)->void { fp(session); }; + + return addCallback(obj, simple_func_wrapper); +} + +std::shared_ptr<core::Processor> ExecutionPlan::addCallback(void *obj, std::function<void(core::ProcessSession*, core::ProcessContext *)> fp) { + if (finalized) { return nullptr; + } - std::shared_ptr<processors::CallbackProcessor> processor = std::static_pointer_cast<processors::CallbackProcessor>(ptr); - processor->setCallback(obj, fp); + auto proc = createCallback(obj, fp); - return addProcessor(processor, "CallbackProcessor", core::Relationship("success", "description"), true); + if (!proc) + return nullptr; + + return addProcessor(proc, CallbackProcessorName, core::Relationship("success", "description"), true); } bool ExecutionPlan::setProperty(const std::shared_ptr<core::Processor> proc, const std::string &prop, const std::string &value) { @@ -198,7 +207,7 @@ std::shared_ptr<minifi::Connection> ExecutionPlan::buildFinalConnection(std::sha void ExecutionPlan::finalize() { if (failure_handler_) { - auto failure_proc = createProcessor("CallbackProcessor", "CallbackProcessor"); + auto failure_proc = createProcessor(CallbackProcessorName, CallbackProcessorName); std::shared_ptr<processors::CallbackProcessor> callback_proc = std::static_pointer_cast<processors::CallbackProcessor>(failure_proc); callback_proc->setCallback(nullptr, std::bind(&FailureHandler::operator(), failure_handler_, std::placeholders::_1)); @@ -237,6 +246,17 @@ std::shared_ptr<core::Processor> ExecutionPlan::createProcessor(const std::strin utils::Identifier uuid; id_generator_->generate(uuid); + 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)); + }; + return createCallback(nullptr, wrapper_func); + } + + auto ptr = core::ClassLoader::getDefaultClassLoader().instantiate(processor_name, uuid); if (nullptr == ptr) { return nullptr; @@ -247,6 +267,17 @@ 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) { + 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); + + return ptr; +} + std::shared_ptr<minifi::Connection> ExecutionPlan::connectProcessors(std::shared_ptr<core::Processor> src_proc, std::shared_ptr<core::Processor> dst_proc, core::Relationship relationship, bool set_dst) { std::stringstream connection_name; @@ -292,3 +323,19 @@ bool ExecutionPlan::setFailureStrategy(FailureStrategy start) { return true; } +bool ExecutionPlan::addCustomProcessor(const char * name, processor_logic* logic) { + if(CallbackProcessorName == name) { + return false; // This name cannot be registered + } + + if (custom_processors.count(name) > 0 ) { + return false; // Already exists + } + custom_processors[name] = logic; + return true; +} + +int ExecutionPlan::deleteCustomProcessor(const char * name) { + return custom_processors.erase(name); +} + http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/627d8fe2/nanofi/tests/CAPITests.cpp ---------------------------------------------------------------------- diff --git a/nanofi/tests/CAPITests.cpp b/nanofi/tests/CAPITests.cpp index 1c94b87..721a0cd 100644 --- a/nanofi/tests/CAPITests.cpp +++ b/nanofi/tests/CAPITests.cpp @@ -45,7 +45,7 @@ static int failure_count = 0; void failure_counter(flow_file_record * fr) { failure_count++; - REQUIRE(get_attribute_qty(fr) > 0); + REQUIRE(get_attribute_quantity(fr) > 0); free_flowfile(fr); } @@ -54,6 +54,36 @@ void big_failure_counter(flow_file_record * fr) { free_flowfile(fr); } +void custom_processor_logic(processor_session * ps, processor_context * ctx) { + flow_file_record * ffr = get_flowfile(ps, ctx); + REQUIRE(ffr != nullptr); + uint8_t * buffer = (uint8_t*)malloc(ffr->size* sizeof(uint8_t)); + get_content(ffr, buffer, ffr->size); + REQUIRE(strncmp(reinterpret_cast<const char *>(buffer), test_file_content.c_str(), test_file_content.size()) == 0); + + attribute attr; + attr.key = "filename"; + attr.value_size = 0; + REQUIRE(get_attribute(ffr, &attr) == 0); + REQUIRE(attr.value_size > 0); + + const char * custom_value = "custom value"; + + REQUIRE(add_attribute(ffr, "custom attribute", (void*)custom_value, strlen(custom_value)) == 0); + + char * prop_value = get_property(ctx, "Some test propery"); + + REQUIRE(prop_value != nullptr); + REQUIRE(strncmp("test value", prop_value, strlen(prop_value)) == 0); + + free(prop_value); + + transfer_to_relationship(ffr, ps, SUCCESS_RELATIONSHIP); + + free_flowfile(ffr); + free(buffer); +} + std::string create_testfile_for_getfile(const char* sourcedir, const std::string& filename = test_file_name) { std::fstream file; std::stringstream ss; @@ -67,7 +97,7 @@ std::string create_testfile_for_getfile(const char* sourcedir, const std::string TEST_CASE("Test Creation of instance, one processor", "[createInstanceAndFlow]") { auto instance = create_instance_obj(); REQUIRE(instance != nullptr); - flow *test_flow = create_flow(instance, nullptr); + flow *test_flow = create_new_flow(instance); REQUIRE(test_flow != nullptr); processor *test_proc = add_processor(test_flow, "GenerateFlowFile"); REQUIRE(test_proc != nullptr); @@ -78,7 +108,7 @@ TEST_CASE("Test Creation of instance, one processor", "[createInstanceAndFlow]") TEST_CASE("Invalid processor returns null", "[addInvalidProcessor]") { auto instance = create_instance_obj(); REQUIRE(instance != nullptr); - flow *test_flow = create_flow(instance, nullptr); + flow *test_flow = create_new_flow(instance); processor *test_proc = add_processor(test_flow, "NeverExisted"); REQUIRE(test_proc == nullptr); processor *no_proc = add_processor(test_flow, ""); @@ -90,7 +120,7 @@ TEST_CASE("Invalid processor returns null", "[addInvalidProcessor]") { TEST_CASE("Set valid and invalid properties", "[setProcesssorProperties]") { auto instance = create_instance_obj(); REQUIRE(instance != nullptr); - flow *test_flow = create_flow(instance, nullptr); + flow *test_flow = create_new_flow(instance); REQUIRE(test_flow != nullptr); processor *test_proc = add_processor(test_flow, "GenerateFlowFile"); REQUIRE(test_proc != nullptr); @@ -114,7 +144,7 @@ TEST_CASE("get file and put file", "[getAndPutFile]") { const char *putfiledir = testController.createTempDirectory(put_format); auto instance = create_instance_obj(); REQUIRE(instance != nullptr); - flow *test_flow = create_flow(instance, nullptr); + flow *test_flow = create_new_flow(instance); REQUIRE(test_flow != nullptr); processor *get_proc = add_processor(test_flow, "GetFile"); REQUIRE(get_proc != nullptr); @@ -164,7 +194,7 @@ TEST_CASE("Test manipulation of attributes", "[testAttributes]") { auto instance = create_instance_obj(); REQUIRE(instance != nullptr); - flow *test_flow = create_flow(instance, nullptr); + flow *test_flow = create_new_flow(instance); REQUIRE(test_flow != nullptr); processor *get_proc = add_processor(test_flow, "GetFile"); @@ -201,7 +231,7 @@ TEST_CASE("Test manipulation of attributes", "[testAttributes]") { // Update overwrites values update_attribute(record, test_attr.key, (void*) new_testattr_value, strlen(new_testattr_value)); // NOLINT - int attr_size = get_attribute_qty(record); + int attr_size = get_attribute_quantity(record); REQUIRE(attr_size > 0); attribute_set attr_set; @@ -237,7 +267,7 @@ TEST_CASE("Test error handling callback", "[errorHandling]") { auto instance = create_instance_obj(); REQUIRE(instance != nullptr); - flow *test_flow = create_flow(instance, nullptr); + flow *test_flow = create_new_flow(instance); REQUIRE(test_flow != nullptr); // Failure strategy cannot be set before a valid callback is added @@ -289,7 +319,7 @@ TEST_CASE("Test standalone processors", "[testStandalone]") { flow_file_record* ffr = invoke(getfile_proc); REQUIRE(ffr != nullptr); - REQUIRE(get_attribute_qty(ffr) > 0); + REQUIRE(get_attribute_quantity(ffr) > 0); standalone_processor* extract_test = create_processor("ExtractText"); REQUIRE(extract_test != nullptr); @@ -301,7 +331,7 @@ TEST_CASE("Test standalone processors", "[testStandalone]") { // Verify the transfer of attributes REQUIRE(ffr2 != nullptr); - REQUIRE(get_attribute_qty(ffr2) > 0); + REQUIRE(get_attribute_quantity(ffr2) > 0); char filename_key[] = "filename"; attribute attr; @@ -334,7 +364,7 @@ TEST_CASE("Test interaction of flow and standlone processors", "[testStandaloneW auto instance = create_instance_obj(); REQUIRE(instance != nullptr); - flow *test_flow = create_flow(instance, nullptr); + flow *test_flow = create_new_flow(instance); REQUIRE(test_flow != nullptr); processor *get_proc = add_processor(test_flow, "GetFile"); @@ -392,3 +422,33 @@ TEST_CASE("Test standalone processors with file input", "[testStandaloneWithFile free_flowfile(ffr); free_standalone_processor(extract_test); } + +TEST_CASE("Test custom processor", "[TestCutomProcessor]") { + TestController testController; + + char src_format[] = "/tmp/gt.XXXXXX"; + const char *sourcedir = testController.createTempDirectory(src_format); + + create_testfile_for_getfile(sourcedir); + + add_custom_processor("myproc", custom_processor_logic); + + auto instance = create_instance_obj(); + REQUIRE(instance != nullptr); + flow *test_flow = create_new_flow(instance); + REQUIRE(test_flow != nullptr); + + processor *get_proc = add_processor(test_flow, "GetFile"); + REQUIRE(get_proc != nullptr); + + REQUIRE(set_property(get_proc, "Input Directory", sourcedir) == 0); + + processor *my_proc = add_processor(test_flow, "myproc"); + REQUIRE(my_proc != nullptr); + + REQUIRE(set_property(my_proc, "Some test propery", "test value") == 0); + + flow_file_record *record = get_next_flow_file(instance, test_flow); + + REQUIRE(record != nullptr); +} \ No newline at end of file
