MINIFI-6: Add ProcessGroup/GenerateFlowFile, etc
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/9280e814 Tree: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/tree/9280e814 Diff: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/diff/9280e814 Branch: refs/heads/MINIFI-6 Commit: 9280e814d6633cb4a91e574c66e4b801a1ccad6c Parents: 62394f7 Author: Bin Qiu <[email protected]> Authored: Tue May 3 14:29:56 2016 -0700 Committer: Bin Qiu <[email protected]> Committed: Tue May 3 14:29:56 2016 -0700 ---------------------------------------------------------------------- Makefile | 2 +- inc/FlowFileRecord.h | 12 ++- inc/GenerateFlowFile.h | 87 +++++++++++++++++++++ inc/ProcessGroup.h | 127 ++++++++++++++++++++++++++++++ inc/ProcessSession.h | 6 +- inc/Property.h | 95 +++++++++++++++++++++++ src/GenerateFlowFile.cpp | 133 ++++++++++++++++++++++++++++++++ src/ProcessGroup.cpp | 174 ++++++++++++++++++++++++++++++++++++++++++ src/ProcessSession.cpp | 12 +-- 9 files changed, 636 insertions(+), 12 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/9280e814/Makefile ---------------------------------------------------------------------- diff --git a/Makefile b/Makefile index 7529a33..bf93910 100644 --- a/Makefile +++ b/Makefile @@ -5,7 +5,7 @@ TARGET_LIB=libminifi.a TARGET_EXE=minifi CFLAGS=-O0 -fexceptions -fpermissive -Wno-write-strings -std=c++11 -fPIC -Wall -g -Wno-unused-private-field INCLUDES=-I./inc -I./src -I./test -I/usr/include/libxml2 -I/usr/local/opt/leveldb/include/ -LDDIRECTORY=-L/usr/local/opt/leveldb/out-static/ -L./build +LDDIRECTORY=-L/usr/local/opt/leveldb/lib/ -L./build LDFLAGS=-lminifi -lxml2 -lleveldb -pthread -luuid UNAME_S := $(shell uname -s) http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/9280e814/inc/FlowFileRecord.h ---------------------------------------------------------------------- diff --git a/inc/FlowFileRecord.h b/inc/FlowFileRecord.h index bceee9f..31838e5 100644 --- a/inc/FlowFileRecord.h +++ b/inc/FlowFileRecord.h @@ -86,8 +86,16 @@ inline const char *FlowAttributeKey(FlowAttribute attribute) //! FlowFile IO Callback functions for input and output //! throw exception for error -typedef void (*InputStreamCallback) (std::ifstream *stream); -typedef void (*OutputStreamCallback) (std::ofstream *stream); +class InputStreamCallback +{ +public: + virtual void process(std::ifstream *stream) = 0; +}; +class OutputStreamCallback +{ +public: + virtual void process(std::ofstream *stream) = 0; +}; //! FlowFile Record Class http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/9280e814/inc/GenerateFlowFile.h ---------------------------------------------------------------------- diff --git a/inc/GenerateFlowFile.h b/inc/GenerateFlowFile.h new file mode 100644 index 0000000..27aa43b --- /dev/null +++ b/inc/GenerateFlowFile.h @@ -0,0 +1,87 @@ +/** + * @file GenerateFlowFile.h + * GenerateFlowFile class declaration + * + * 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 __GENERATE_FLOW_FILE_H__ +#define __GENERATE_FLOW_FILE_H__ + +#include "FlowFileRecord.h" +#include "Processor.h" +#include "ProcessSession.h" + +//! GenerateFlowFile Class +class GenerateFlowFile : public Processor +{ +public: + //! Constructor + /*! + * Create a new processor + */ + GenerateFlowFile(std::string name, uuid_t uuid = NULL) + : Processor(name, uuid) + { + _data = NULL; + _dataSize = 0; + } + //! Destructor + virtual ~GenerateFlowFile() + { + if (_data) + delete[] _data; + } + //! Processor Name + static const std::string ProcessorName; + //! Supported Properties + static Property FileSize; + static Property BatchSize; + static Property DataFormat; + static Property UniqueFlowFiles; + static const char *DATA_FORMAT_BINARY; + static const char *DATA_FORMAT_TEXT; + //! Supported Relationships + static Relationship Success; + //! Nest Callback Class for write stream + class WriteCallback : public OutputStreamCallback + { + public: + WriteCallback(char *data, uint64_t size) + : _data(data), _dataSize(size) {} + char *_data; + uint64_t _dataSize; + void process(std::ofstream *stream) { + if (_data && _dataSize > 0) + stream->write(_data, _dataSize); + } + }; + +public: + //! OnTrigger method, implemented by NiFi GenerateFlowFile + virtual void onTrigger(ProcessContext *context, ProcessSession *session); + //! Initialize, over write by NiFi GenerateFlowFile + virtual void initialize(void); + +protected: + +private: + //! Generated data + char * _data; + //! Size of the generate data + uint64_t _dataSize; +}; + +#endif http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/9280e814/inc/ProcessGroup.h ---------------------------------------------------------------------- diff --git a/inc/ProcessGroup.h b/inc/ProcessGroup.h new file mode 100644 index 0000000..a95ae8d --- /dev/null +++ b/inc/ProcessGroup.h @@ -0,0 +1,127 @@ +/** + * @file ProcessGroup.h + * ProcessGroup class declaration + * + * 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 __PROCESS_GROUP_H__ +#define __PROCESS_GROUP_H__ + +#include <uuid/uuid.h> +#include <vector> +#include <queue> +#include <map> +#include <mutex> +#include <atomic> +#include <algorithm> +#include <set> + +#include "Logger.h" +#include "Processor.h" +#include "Exception.h" + +//! Process Group Type +enum ProcessGroupType +{ + ROOT_PROCESS_GROUP = 0, + REMOTE_PROCESS_GROUP, + MAX_PROCESS_GROUP_TYPE +}; + +//! ProcessGroup Class +class ProcessGroup +{ +public: + //! Constructor + /*! + * Create a new process group + */ + ProcessGroup(ProcessGroupType type, std::string name, uuid_t uuid = NULL, ProcessGroup *parent = NULL); + //! Destructor + virtual ~ProcessGroup(); + //! Set Processor Name + void setName(std::string name) { + _name = name; + } + //! Get Process Name + std::string getName(void) { + return (_name); + } + //! Set UUID + void setUUID(uuid_t uuid) { + uuid_copy(_uuid, uuid); + } + //! Get UUID + bool getUUID(uuid_t uuid) { + if (uuid) + { + uuid_copy(uuid, _uuid); + return true; + } + else + return false; + } + //! Start Processing + void startProcessing(); + //! Stop Processing + void stopProcessing(); + //! Whether it is root process group + bool isRootProcessGroup(); + //! set parent process group + void setParent(ProcessGroup *parent) { + std::lock_guard<std::mutex> lock(_mtx); + _parentProcessGroup = parent; + } + //! get parent process group + ProcessGroup *getParent(void) { + std::lock_guard<std::mutex> lock(_mtx); + return _parentProcessGroup; + } + //! Add processor + void addProcessor(Processor *processor); + //! Remove processor + void removeProcessor(Processor *processor); + //! Add child processor group + void addProcessGroup(ProcessGroup *child); + //! Remove child processor group + void removeProcessGroup(ProcessGroup *child); + +protected: + //! A global unique identifier + uuid_t _uuid; + //! Processor Group Name + std::string _name; + //! Process Group Type + ProcessGroupType _type; + //! Processors (ProcessNode) inside this process group which include Input/Output Port, Remote Process Group input/Output port + std::set<Processor *> _processors; + std::set<ProcessGroup *> _childProcessGroups; + //! Parent Process Group + ProcessGroup* _parentProcessGroup; + +private: + + //! Mutex for protection + std::mutex _mtx; + //! Logger + Logger *_logger; + // Prevent default copy constructor and assignment operation + // Only support pass by reference or pointer + ProcessGroup(const ProcessGroup &parent); + ProcessGroup &operator=(const ProcessGroup &parent); +}; + +#endif http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/9280e814/inc/ProcessSession.h ---------------------------------------------------------------------- diff --git a/inc/ProcessSession.h b/inc/ProcessSession.h index 2ee1cf0..efd882e 100644 --- a/inc/ProcessSession.h +++ b/inc/ProcessSession.h @@ -75,11 +75,11 @@ public: //! Remove Flow File void remove(FlowFileRecord *flow); //! Execute the given read callback against the content - void read(FlowFileRecord *flow, InputStreamCallback callback); + void read(FlowFileRecord *flow, InputStreamCallback *callback); //! Execute the given write callback against the content - void write(FlowFileRecord *flow, OutputStreamCallback callback); + void write(FlowFileRecord *flow, OutputStreamCallback *callback); //! Execute the given write/append callback against the content - void append(FlowFileRecord *flow, OutputStreamCallback callback); + void append(FlowFileRecord *flow, OutputStreamCallback *callback); //! Penalize the flow void penalize(FlowFileRecord *flow); http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/9280e814/inc/Property.h ---------------------------------------------------------------------- diff --git a/inc/Property.h b/inc/Property.h index e5a6e2a..f3d58f1 100644 --- a/inc/Property.h +++ b/inc/Property.h @@ -27,6 +27,9 @@ #include <mutex> #include <atomic> #include <set> +#include <stdlib.h> +#include <math.h> + //! Property Class class Property { @@ -62,6 +65,98 @@ public: bool operator < (const Property & right) const { return _name < right._name; } + //! Convert String to Integer + static bool StringToInt(std::string input, int64_t &output) + { + if (input.size() == 0) { + return false; + } + + const char *cvalue = input.c_str(); + char *pEnd; + long int ival = strtol(cvalue, &pEnd, 0); + + if (pEnd[0] == '\0') + { + output = ival; + return true; + } + + char end0 = toupper(pEnd[0]); + if ( (end0 == 'K') || (end0 == 'M') || (end0 == 'G') || (end0 == 'T') || (end0 == 'P') ) + { + if (pEnd[1] == '\0') + { + unsigned long int multiplier = 1000; + + if ( (end0 != 'K') ) { + multiplier *= 1000; + if (end0 != 'M') { + multiplier *= 1000; + if (end0 != 'G') { + multiplier *= 1000; + if (end0 != 'T') { + multiplier *= 1000; + } + } + } + } + output = ival * multiplier; + return true; + + } else if ((pEnd[1] == 'b' || pEnd[1] == 'B') && (pEnd[2] == '\0')) { + + unsigned long int multiplier = 1024; + + if ( (end0 != 'K') ) { + multiplier *= 1024; + if (end0 != 'M') { + multiplier *= 1024; + if (end0 != 'G') { + multiplier *= 1024; + if (end0 != 'T') { + multiplier *= 1024; + } + } + } + } + output = ival * multiplier; + return true; + } + } + + return false; + } + //! Convert String to Float + static bool StringToFloat(std::string input, float &output) + { + const char *cvalue = input.c_str(); + char *pEnd; + float val = strtof(cvalue, &pEnd); + + if (pEnd[0] == '\0') + { + output = val; + return true; + } + else + return false; + } + //! Convert String to Bool + static bool StringToBool(std::string input, bool &output) + { + if (input == "true" || input == "True" || input == "TRUE") + { + output = true; + return true; + } + if (input == "false" || input == "False" || input == "FALSE") + { + output = false; + return true; + } + return false; + } protected: //! Name http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/9280e814/src/GenerateFlowFile.cpp ---------------------------------------------------------------------- diff --git a/src/GenerateFlowFile.cpp b/src/GenerateFlowFile.cpp new file mode 100644 index 0000000..16181de --- /dev/null +++ b/src/GenerateFlowFile.cpp @@ -0,0 +1,133 @@ +/** + * @file GenerateFlowFile.cpp + * GenerateFlowFile class implementation + * + * 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 <vector> +#include <queue> +#include <map> +#include <set> +#include <sys/time.h> +#include <time.h> +#include <chrono> +#include <thread> +#include <random> + +#include "GenerateFlowFile.h" +#include "ProcessContext.h" +#include "ProcessSession.h" + +const char *GenerateFlowFile::DATA_FORMAT_BINARY = "Binary"; +const char *GenerateFlowFile::DATA_FORMAT_TEXT = "Text"; +const std::string GenerateFlowFile::ProcessorName("GenerateFlowFile"); +Property GenerateFlowFile::FileSize("File Size", "The size of the file that will be used", "1 kB"); +Property GenerateFlowFile::BatchSize("Batch Size", "The number of FlowFiles to be transferred in each invocation", "1"); +Property GenerateFlowFile::DataFormat("Data Format", "Specifies whether the data should be Text or Binary", GenerateFlowFile::DATA_FORMAT_BINARY); +Property GenerateFlowFile::UniqueFlowFiles("Unique FlowFiles", + "If true, each FlowFile that is generated will be unique. If false, a random value will be generated and all FlowFiles", "true"); +Relationship GenerateFlowFile::Success("success", "success operational on the flow record"); + +void GenerateFlowFile::initialize() +{ + //! Set the supported properties + std::set<Property> properties; + properties.insert(FileSize); + properties.insert(BatchSize); + properties.insert(DataFormat); + properties.insert(UniqueFlowFiles); + setSupportedProperties(properties); + //! Set the supported relationships + std::set<Relationship> relationships; + relationships.insert(Success); + setSupportedRelationships(relationships); +} + +void GenerateFlowFile::onTrigger(ProcessContext *context, ProcessSession *session) +{ + int64_t batchSize = 1; + bool uniqueFlowFile = true; + int64_t fileSize = 1024; + + std::string value; + if (context->getProperty(FileSize.getName(), value)) + { + Property::StringToInt(value, fileSize); + } + if (context->getProperty(BatchSize.getName(), value)) + { + Property::StringToInt(value, batchSize); + } + if (context->getProperty(UniqueFlowFiles.getName(), value)) + { + Property::StringToBool(value, uniqueFlowFile); + } + + if (!uniqueFlowFile) + { + char *data; + data = new char[fileSize]; + if (!data) + return; + uint64_t dataSize = fileSize; + GenerateFlowFile::WriteCallback callback(data, dataSize); + char *current = data; + for (int i = 0; i < fileSize; i+= sizeof(int)) + { + int randValue = random(); + *((int *) current) = randValue; + current += sizeof(int); + } + for (int i = 0; i < batchSize; i++) + { + // For each batch + FlowFileRecord *flowFile = session->create(); + if (!flowFile) + return; + if (fileSize > 0) + session->write(flowFile, &callback); + session->transfer(flowFile, Success); + } + delete[] data; + } + else + { + if (!_data) + { + // We have not create the unique data yet + _data = new char[fileSize]; + _dataSize = fileSize; + char *current = _data; + for (int i = 0; i < fileSize; i+= sizeof(int)) + { + int randValue = random(); + *((int *) current) = randValue; + current += sizeof(int); + } + } + GenerateFlowFile::WriteCallback callback(_data, _dataSize); + for (int i = 0; i < batchSize; i++) + { + // For each batch + FlowFileRecord *flowFile = session->create(); + if (!flowFile) + return; + if (fileSize > 0) + session->write(flowFile, &callback); + session->transfer(flowFile, Success); + } + } +} http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/9280e814/src/ProcessGroup.cpp ---------------------------------------------------------------------- diff --git a/src/ProcessGroup.cpp b/src/ProcessGroup.cpp new file mode 100644 index 0000000..c214c23 --- /dev/null +++ b/src/ProcessGroup.cpp @@ -0,0 +1,174 @@ +/** + * @file ProcessGroup.cpp + * ProcessGroup class implementation + * + * 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 <vector> +#include <queue> +#include <map> +#include <set> +#include <sys/time.h> +#include <time.h> +#include <chrono> +#include <thread> + +#include "ProcessGroup.h" +#include "Processor.h" + +ProcessGroup::ProcessGroup(ProcessGroupType type, std::string name, uuid_t uuid, ProcessGroup *parent) +: _name(name), + _type(type), + _parentProcessGroup(parent) +{ + if (!uuid) + // Generate the global UUID for the flow record + uuid_generate(_uuid); + else + uuid_copy(_uuid, uuid); + + _logger = Logger::getLogger(); + _logger->log_info("ProcessGroup %s created", _name.c_str()); +} + +ProcessGroup::~ProcessGroup() +{ + +} + +bool ProcessGroup::isRootProcessGroup() +{ + std::lock_guard<std::mutex> lock(_mtx); + return (_type == ROOT_PROCESS_GROUP); +} + +void ProcessGroup::addProcessor(Processor *processor) +{ + std::lock_guard<std::mutex> lock(_mtx); + + if (_processors.find(processor) == _processors.end()) + { + // We do not have the same processor in this process group yet + _processors.insert(processor); + _logger->log_info("Add processor %s into process group %s", + processor->getName().c_str(), _name.c_str()); + } +} + +void ProcessGroup::removeProcessor(Processor *processor) +{ + std::lock_guard<std::mutex> lock(_mtx); + + if (_processors.find(processor) != _processors.end()) + { + // We do have the same processor in this process group yet + _processors.erase(processor); + _logger->log_info("Remove processor %s from process group %s", + processor->getName().c_str(), _name.c_str()); + } +} + +void ProcessGroup::addProcessGroup(ProcessGroup *child) +{ + std::lock_guard<std::mutex> lock(_mtx); + + if (_childProcessGroups.find(child) == _childProcessGroups.end()) + { + // We do not have the same child process group in this process group yet + _childProcessGroups.insert(child); + _logger->log_info("Add child process group %s into process group %s", + child->getName().c_str(), _name.c_str()); + } +} + +void ProcessGroup::removeProcessGroup(ProcessGroup *child) +{ + std::lock_guard<std::mutex> lock(_mtx); + + if (_childProcessGroups.find(child) != _childProcessGroups.end()) + { + // We do have the same child process group in this process group yet + _childProcessGroups.erase(child); + _logger->log_info("Remove child process group %s from process group %s", + child->getName().c_str(), _name.c_str()); + } +} + +void ProcessGroup::startProcessing() +{ + std::lock_guard<std::mutex> lock(_mtx); + + try + { + // Start all the processor node, input and output ports + for (std::set<Processor *>::iterator it = _processors.begin(); it != _processors.end(); ++it) + { + Processor *processor(*it); + if (!processor->isRunning() && processor->getScheduledState() != DISABLED) + // TODO: call scheduler to start processor + ; + } + + for (std::set<ProcessGroup *>::iterator it = _childProcessGroups.begin(); it != _childProcessGroups.end(); ++it) + { + ProcessGroup *processGroup(*it); + processGroup->startProcessing(); + } + } + catch (std::exception &exception) + { + _logger->log_debug("Caught Exception %s", exception.what()); + throw; + } + catch (...) + { + _logger->log_debug("Caught Exception during process group start processing"); + throw; + } +} + +void ProcessGroup::stopProcessing() +{ + std::lock_guard<std::mutex> lock(_mtx); + + try + { + // Stop all the processor node, input and output ports + for (std::set<Processor *>::iterator it = _processors.begin(); it != _processors.end(); ++it) + { + Processor *processor(*it); + if (!processor->isRunning()) + // TODO: call scheduler to stop processor + ; + } + + for (std::set<ProcessGroup *>::iterator it = _childProcessGroups.begin(); it != _childProcessGroups.end(); ++it) + { + ProcessGroup *processGroup(*it); + processGroup->stopProcessing(); + } + } + catch (std::exception &exception) + { + _logger->log_debug("Caught Exception %s", exception.what()); + throw; + } + catch (...) + { + _logger->log_debug("Caught Exception during process group stop processing"); + throw; + } +} http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/9280e814/src/ProcessSession.cpp ---------------------------------------------------------------------- diff --git a/src/ProcessSession.cpp b/src/ProcessSession.cpp index 3cd7ba5..11c2e94 100644 --- a/src/ProcessSession.cpp +++ b/src/ProcessSession.cpp @@ -141,7 +141,7 @@ void ProcessSession::transfer(FlowFileRecord *flow, Relationship relationship) _transferRelationship[flow->getUUIDStr()] = relationship; } -void ProcessSession::write(FlowFileRecord *flow, OutputStreamCallback callback) +void ProcessSession::write(FlowFileRecord *flow, OutputStreamCallback *callback) { ResourceClaim *claim = NULL; @@ -154,7 +154,7 @@ void ProcessSession::write(FlowFileRecord *flow, OutputStreamCallback callback) if (fs.is_open()) { // Call the callback to write the content - callback(&fs); + callback->process(&fs); if (fs.good() && fs.tellp() >= 0) { flow->_size = fs.tellp(); @@ -197,7 +197,7 @@ void ProcessSession::write(FlowFileRecord *flow, OutputStreamCallback callback) } } -void ProcessSession::append(FlowFileRecord *flow, OutputStreamCallback callback) +void ProcessSession::append(FlowFileRecord *flow, OutputStreamCallback *callback) { ResourceClaim *claim = NULL; @@ -217,7 +217,7 @@ void ProcessSession::append(FlowFileRecord *flow, OutputStreamCallback callback) { // Call the callback to write the content std::streampos oldPos = fs.tellp(); - callback(&fs); + callback->process(&fs); if (fs.good() && fs.tellp() >= 0) { uint64_t appendSize = fs.tellp() - oldPos; @@ -253,7 +253,7 @@ void ProcessSession::append(FlowFileRecord *flow, OutputStreamCallback callback) } } -void ProcessSession::read(FlowFileRecord *flow, InputStreamCallback callback) +void ProcessSession::read(FlowFileRecord *flow, InputStreamCallback *callback) { try { @@ -273,7 +273,7 @@ void ProcessSession::read(FlowFileRecord *flow, InputStreamCallback callback) if (fs.good()) { - callback(&fs); + callback->process(&fs); _logger->log_debug("Read offset %d size %d content %s for FlowFile UUID %s", flow->_offset, flow->_size, claim->getContentFullPath().c_str(), flow->getUUIDStr().c_str()); fs.close();
