You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by al...@apache.org on 2017/04/05 14:38:34 UTC
nifi-minifi-cpp git commit: MINIFI-236: Make GetFile, PutFile,
TailFile, and ExecuteProcess thread safe
Repository: nifi-minifi-cpp
Updated Branches:
refs/heads/master a8de19653 -> 3676468fc
MINIFI-236: Make GetFile, PutFile, TailFile, and ExecuteProcess thread safe
This closes #71.
Signed-off-by: Aldrin Piri <al...@apache.org>
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/3676468f
Tree: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/tree/3676468f
Diff: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/diff/3676468f
Branch: refs/heads/master
Commit: 3676468fc0746ca9a31c92fa980fb412f3c010b2
Parents: a8de196
Author: Marc Parisi <ph...@apache.org>
Authored: Tue Mar 28 15:50:48 2017 -0400
Committer: Aldrin Piri <al...@apache.org>
Committed: Wed Apr 5 16:38:03 2017 +0200
----------------------------------------------------------------------
.travis.yml | 2 +-
CMakeLists.txt | 15 +
libminifi/include/core/ConfigurableComponent.h | 16 +-
libminifi/include/core/FlowConfiguration.h | 2 +-
libminifi/include/core/ProcessSession.h | 17 +-
libminifi/include/core/Processor.h | 2 -
libminifi/include/processors/GetFile.h | 77 +++---
libminifi/include/processors/PutFile.h | 22 +-
libminifi/include/processors/TailFile.h | 19 +-
libminifi/include/utils/ThreadPool.h | 287 ++++++++++++++++++++
libminifi/src/core/ConfigurableComponent.cpp | 10 +-
libminifi/src/core/FlowConfiguration.cpp | 6 +-
libminifi/src/core/ProcessSession.cpp | 94 ++++++-
libminifi/src/processors/GetFile.cpp | 100 +++----
libminifi/src/processors/PutFile.cpp | 53 ++--
libminifi/src/processors/TailFile.cpp | 32 ++-
libminifi/test/TestExecuteProcess.cpp | 131 +++++++++
libminifi/test/unit/ProcessorTests.cpp | 144 ++++++++--
18 files changed, 852 insertions(+), 177 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/3676468f/.travis.yml
----------------------------------------------------------------------
diff --git a/.travis.yml b/.travis.yml
index 31cb731..faa291b 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -47,4 +47,4 @@ matrix:
- package='openssl'; [[ $(brew ls --versions ${package}) ]] && { brew outdated ${package} || brew upgrade ${package}; } || brew install ${package}
script:
- - mkdir ./build && cd ./build && cmake .. && make && ./tests
+ - mkdir ./build && cd ./build && cmake .. && make && make test
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/3676468f/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 5d7875e..b84706d 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -136,6 +136,21 @@ enable_testing(test)
target_include_directories(tests PRIVATE BEFORE "libminifi/include/provenance")
target_link_libraries(tests ${CMAKE_THREAD_LIBS_INIT} ${UUID_LIBRARIES} ${LEVELDB_LIBRARIES} ${OPENSSL_LIBRARIES} minifi yaml-cpp c-library civetweb-cpp)
add_test(NAME LibMinifiTests COMMAND tests)
+
+ file(GLOB LIBMINIFI_TEST_EXECUTE_PROCESS "libminifi/test/TestExecuteProcess.cpp")
+ add_executable(testExecuteProcess ${LIBMINIFI_TEST_EXECUTE_PROCESS} ${SPD_SOURCES})
+ target_include_directories(testExecuteProcess PRIVATE BEFORE "thirdparty/yaml-cpp-yaml-cpp-0.5.3/include")
+ target_include_directories(testExecuteProcess PRIVATE BEFORE ${LEVELDB_INCLUDE_DIRS})
+ target_include_directories(testExecuteProcess PRIVATE BEFORE "include")
+ target_include_directories(testExecuteProcess PRIVATE BEFORE "libminifi/include/")
+ target_include_directories(testExecuteProcess PRIVATE BEFORE "libminifi/include/core")
+ target_include_directories(testExecuteProcess PRIVATE BEFORE "libminifi/include/core/repository")
+ target_include_directories(testExecuteProcess PRIVATE BEFORE "libminifi/include/io")
+ target_include_directories(testExecuteProcess PRIVATE BEFORE "libminifi/include/utils")
+ target_include_directories(testExecuteProcess PRIVATE BEFORE "libminifi/include/processors")
+ target_include_directories(testExecuteProcess PRIVATE BEFORE "libminifi/include/provenance")
+ target_link_libraries(testExecuteProcess ${CMAKE_THREAD_LIBS_INIT} ${UUID_LIBRARIES} ${LEVELDB_LIBRARIES} ${OPENSSL_LIBRARIES} minifi yaml-cpp c-library civetweb-cpp)
+ add_test(NAME ExecuteProcess COMMAND testExecuteProcess)
# Create a custom build target called "docker" that will invoke DockerBuild.sh and create the NiFi-MiNiFi-CPP Docker image
add_custom_target(
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/3676468f/libminifi/include/core/ConfigurableComponent.h
----------------------------------------------------------------------
diff --git a/libminifi/include/core/ConfigurableComponent.h b/libminifi/include/core/ConfigurableComponent.h
index c0cc623..d46216b 100644
--- a/libminifi/include/core/ConfigurableComponent.h
+++ b/libminifi/include/core/ConfigurableComponent.h
@@ -19,8 +19,14 @@
#ifndef LIBMINIFI_INCLUDE_CORE_CONFIGURABLECOMPONENT_H_
#define LIBMINIFI_INCLUDE_CORE_CONFIGURABLECOMPONENT_H_
+#include <mutex>
+#include <iostream>
+#include <map>
+#include <memory>
+#include <set>
+
+#include "logging/Logger.h"
#include "Property.h"
-#include "core/logging/Logger.h"
namespace org {
namespace apache {
@@ -35,10 +41,8 @@ namespace core {
class ConfigurableComponent {
public:
-
ConfigurableComponent() = delete;
-
explicit ConfigurableComponent(std::shared_ptr<logging::Logger> logger);
explicit ConfigurableComponent(const ConfigurableComponent &&other);
@@ -81,7 +85,6 @@ class ConfigurableComponent {
protected:
-
/**
* Returns true if the instance can be edited.
* @return true/false
@@ -89,10 +92,13 @@ class ConfigurableComponent {
virtual bool canEdit()= 0;
std::mutex configuration_mutex_;
- std::shared_ptr<logging::Logger> logger_;
+
// Supported properties
std::map<std::string, Property> properties_;
+ private:
+ std::shared_ptr<logging::Logger> my_logger_;
+
};
} /* namespace core */
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/3676468f/libminifi/include/core/FlowConfiguration.h
----------------------------------------------------------------------
diff --git a/libminifi/include/core/FlowConfiguration.h b/libminifi/include/core/FlowConfiguration.h
index c7eedd2..e95e684 100644
--- a/libminifi/include/core/FlowConfiguration.h
+++ b/libminifi/include/core/FlowConfiguration.h
@@ -25,10 +25,10 @@
#include "processors/GetFile.h"
#include "processors/PutFile.h"
#include "processors/TailFile.h"
+#include "processors/ListenHTTP.h"
#include "processors/ListenSyslog.h"
#include "processors/GenerateFlowFile.h"
#include "processors/RealTimeDataCollector.h"
-#include "processors/ListenHTTP.h"
#include "processors/LogAttribute.h"
#include "processors/ExecuteProcess.h"
#include "processors/AppendHostInfo.h"
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/3676468f/libminifi/include/core/ProcessSession.h
----------------------------------------------------------------------
diff --git a/libminifi/include/core/ProcessSession.h b/libminifi/include/core/ProcessSession.h
index b516817..a80769e 100644
--- a/libminifi/include/core/ProcessSession.h
+++ b/libminifi/include/core/ProcessSession.h
@@ -77,7 +77,13 @@ class ProcessSession {
// Create a new UUID FlowFile with no content resource claim and without parent
std::shared_ptr<core::FlowFile> create();
// Create a new UUID FlowFile with no content resource claim and inherit all attributes from parent
- std::shared_ptr<core::FlowFile> create(std::shared_ptr<core::FlowFile> &parent);
+ std::shared_ptr<core::FlowFile> create(
+ std::shared_ptr<core::FlowFile> &&parent);
+
+ std::shared_ptr<core::FlowFile> create(
+ std::shared_ptr<core::FlowFile> &parent){
+ return create(parent);
+ }
// Clone a new UUID FlowFile from parent both for content resource claim and attributes
std::shared_ptr<core::FlowFile> clone(
std::shared_ptr<core::FlowFile> &parent);
@@ -121,7 +127,14 @@ class ProcessSession {
// Penalize the flow
void penalize(std::shared_ptr<core::FlowFile> &flow);
void penalize(std::shared_ptr<core::FlowFile> &&flow);
-// Import the existed file into the flow
+
+ /**
+ * Imports a file from the data stream
+ * @param stream incoming data stream that contains the data to store into a file
+ * @param flow flow file
+ */
+ void importFrom(io::DataStream &stream, std::shared_ptr<core::FlowFile> &&flow);
+ // import from the data source.
void import(std::string source, std::shared_ptr<core::FlowFile> &flow,
bool keepSource = true, uint64_t offset = 0);
void import(std::string source, std::shared_ptr<core::FlowFile> &&flow,
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/3676468f/libminifi/include/core/Processor.h
----------------------------------------------------------------------
diff --git a/libminifi/include/core/Processor.h b/libminifi/include/core/Processor.h
index fd0411f..4a71816 100644
--- a/libminifi/include/core/Processor.h
+++ b/libminifi/include/core/Processor.h
@@ -251,8 +251,6 @@ private:
// Check all incoming connections for work
bool isWorkAvailable();
- // Logger
- std::shared_ptr<logging::Logger> logger_;
// Prevent default copy constructor and assignment operation
// Only support pass by reference or pointer
Processor(const Processor &parent);
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/3676468f/libminifi/include/processors/GetFile.h
----------------------------------------------------------------------
diff --git a/libminifi/include/processors/GetFile.h b/libminifi/include/processors/GetFile.h
index f1f0694..cc3beaa 100644
--- a/libminifi/include/processors/GetFile.h
+++ b/libminifi/include/processors/GetFile.h
@@ -1,6 +1,4 @@
/**
- * @file GetFile.h
- * GetFile class declaration
*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
@@ -20,6 +18,7 @@
#ifndef __GET_FILE_H__
#define __GET_FILE_H__
+#include <atomic>
#include "FlowFileRecord.h"
#include "core/Processor.h"
#include "core/ProcessSession.h"
@@ -31,6 +30,20 @@ namespace nifi {
namespace minifi {
namespace processors {
+ struct GetFileRequest{
+ std::string directory = ".";
+ bool recursive = true;
+ bool keepSourceFile = false;
+ int64_t minAge = 0;
+ int64_t maxAge = 0;
+ int64_t minSize = 0;
+ int64_t maxSize = 0;
+ bool ignoreHiddenFile = true;
+ int64_t pollInterval = 0;
+ int64_t batchSize = 10;
+ std::string fileFilter= "[^\\.].*";
+ };
+
// GetFile Class
class GetFile : public core::Processor {
public:
@@ -38,21 +51,9 @@ class GetFile : public core::Processor {
/*!
* Create a new processor
*/
- GetFile(std::string name, uuid_t uuid = NULL)
+ explicit GetFile(std::string name, uuid_t uuid = NULL)
: Processor(name, uuid) {
- logger_ = logging::Logger::getLogger();
- _directory = ".";
- _recursive = true;
- _keepSourceFile = false;
- _minAge = 0;
- _maxAge = 0;
- _minSize = 0;
- _maxSize = 0;
- _ignoreHiddenFile = true;
- _pollInterval = 0;
- _batchSize = 10;
- _lastDirectoryListingTime = getTimeMillis();
- _fileFilter = "[^\\.].*";
+
}
// Destructor
virtual ~GetFile() {
@@ -79,16 +80,28 @@ class GetFile : public core::Processor {
virtual void onTrigger(
core::ProcessContext *context,
core::ProcessSession *session);
+ /**
+ * Function that's executed when the processor is scheduled.
+ * @param context process context.
+ * @param sessionFactory process session factory that is used when creating
+ * ProcessSession objects.
+ */
+ void onSchedule(
+ core::ProcessContext *context,
+ core::ProcessSessionFactory *sessionFactory);
// Initialize, over write by NiFi GetFile
virtual void initialize(void);
- // perform directory listing
- void performListing(std::string dir);
+ /**
+ * performs a listeing on the directory.
+ * @param dir directory to list
+ * @param request get file request.
+ */
+ void performListing(std::string dir,const GetFileRequest &request);
protected:
private:
- // Logger
- std::shared_ptr<logging::Logger> logger_;
+
// Queue for store directory list
std::queue<std::string> _dirList;
// Get Listing size
@@ -101,23 +114,19 @@ class GetFile : public core::Processor {
// Put full path file name into directory listing
void putListing(std::string fileName);
// Poll directory listing for files
- void pollListing(std::queue<std::string> &list, int maxSize);
+ void pollListing(std::queue<std::string> &list,const GetFileRequest &request);
// Check whether file can be added to the directory listing
- bool acceptFile(std::string fullName, std::string name);
+ bool acceptFile(std::string fullName, std::string name, const GetFileRequest &request);
+ // Get file request object.
+ GetFileRequest request_;
// Mutex for protection of the directory listing
+
std::mutex mutex_;
- std::string _directory;
- bool _recursive;
- bool _keepSourceFile;
- int64_t _minAge;
- int64_t _maxAge;
- int64_t _minSize;
- int64_t _maxSize;
- bool _ignoreHiddenFile;
- int64_t _pollInterval;
- int64_t _batchSize;
- uint64_t _lastDirectoryListingTime;
- std::string _fileFilter;
+
+ // last listing time for root directory ( if recursive, we will consider the root
+ // as the top level time.
+ std::atomic<uint64_t> last_listing_time_;
+
};
} /* namespace processors */
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/3676468f/libminifi/include/processors/PutFile.h
----------------------------------------------------------------------
diff --git a/libminifi/include/processors/PutFile.h b/libminifi/include/processors/PutFile.h
index 7653fac..c0effaf 100644
--- a/libminifi/include/processors/PutFile.h
+++ b/libminifi/include/processors/PutFile.h
@@ -45,7 +45,6 @@ class PutFile : public core::Processor {
*/
PutFile(std::string name, uuid_t uuid = NULL)
: core::Processor(name, uuid) {
- logger_ = logging::Logger::getLogger();
}
// Destructor
virtual ~PutFile() {
@@ -59,10 +58,18 @@ class PutFile : public core::Processor {
static core::Relationship Success;
static core::Relationship Failure;
+ /**
+ * Function that's executed when the processor is scheduled.
+ * @param context process context.
+ * @param sessionFactory process session factory that is used when creating
+ * ProcessSession objects.
+ */
+ void onSchedule(core::ProcessContext *context,
+ core::ProcessSessionFactory *sessionFactory);
+
// OnTrigger method, implemented by NiFi PutFile
- virtual void onTrigger(
- core::ProcessContext *context,
- core::ProcessSession *session);
+ virtual void onTrigger(core::ProcessContext *context,
+ core::ProcessSession *session);
// Initialize, over write by NiFi PutFile
virtual void initialize(void);
@@ -84,8 +91,11 @@ class PutFile : public core::Processor {
protected:
private:
- // Logger
- std::shared_ptr<logging::Logger> logger_;
+
+ // directory
+ std::string directory_;
+ // conflict resolution type.
+ std::string conflict_resolution_;
bool putFile(core::ProcessSession *session,
std::shared_ptr<FlowFileRecord> flowFile,
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/3676468f/libminifi/include/processors/TailFile.h
----------------------------------------------------------------------
diff --git a/libminifi/include/processors/TailFile.h b/libminifi/include/processors/TailFile.h
index 5be76e4..c7b7b46 100644
--- a/libminifi/include/processors/TailFile.h
+++ b/libminifi/include/processors/TailFile.h
@@ -38,9 +38,8 @@ class TailFile : public core::Processor {
/*!
* Create a new processor
*/
- TailFile(std::string name, uuid_t uuid = NULL)
+ explicit TailFile(std::string name, uuid_t uuid = NULL)
: core::Processor(name, uuid) {
- logger_ = logging::Logger::getLogger();
_stateRecovered = false;
}
// Destructor
@@ -57,9 +56,8 @@ class TailFile : public core::Processor {
public:
// OnTrigger method, implemented by NiFi TailFile
- virtual void onTrigger(
- core::ProcessContext *context,
- core::ProcessSession *session);
+ virtual void onTrigger(core::ProcessContext *context,
+ core::ProcessSession *session);
// Initialize, over write by NiFi TailFile
virtual void initialize(void);
// recoverState
@@ -70,11 +68,7 @@ class TailFile : public core::Processor {
protected:
private:
- // Logger
- std::shared_ptr<logging::Logger> logger_;
- std::string _fileLocation;
- // Property Specified Tailed File Name
- std::string _fileName;
+ std::mutex tail_file_mutex_;
// File to save state
std::string _stateFile;
// State related to the tailed file
@@ -86,7 +80,10 @@ class TailFile : public core::Processor {
std::string trimLeft(const std::string& s);
std::string trimRight(const std::string& s);
void parseStateFileLine(char *buf);
- void checkRollOver();
+ /**
+ * Check roll over for the provided file.
+ */
+ void checkRollOver(const std::string &, const std::string&);
};
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/3676468f/libminifi/include/utils/ThreadPool.h
----------------------------------------------------------------------
diff --git a/libminifi/include/utils/ThreadPool.h b/libminifi/include/utils/ThreadPool.h
new file mode 100644
index 0000000..7508900
--- /dev/null
+++ b/libminifi/include/utils/ThreadPool.h
@@ -0,0 +1,287 @@
+/**
+ * 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 LIBMINIFI_INCLUDE_THREAD_POOL_H
+#define LIBMINIFI_INCLUDE_THREAD_POOL_H
+
+#include <iostream>
+#include <atomic>
+#include <mutex>
+#include <vector>
+#include <queue>
+#include <future>
+#include <thread>
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace utils {
+
+/**
+ * Worker task
+ * purpose: Provides a wrapper for the functor
+ * and returns a future based on the template argument.
+ */
+template< typename T>
+class Worker{
+public:
+ explicit Worker(std::function<T()> &task) : task(task)
+ {
+ promise = std::make_shared<std::promise<T>>();
+ }
+
+ /**
+ * Move constructor for worker tasks
+ */
+ Worker(Worker &&other) : task (std::move(other.task)),
+ promise(other.promise)
+ {
+ }
+
+
+ /**
+ * Runs the task and takes the output from the funtor
+ * setting the result into the promise
+ */
+ void run()
+ {
+ T result = task();
+ promise->set_value(result);
+ }
+
+ Worker<T>(const Worker<T>&) = delete;
+ Worker<T>& operator = (const Worker<T>&) = delete;
+
+ Worker<T>& operator = (Worker<T>&&) ;
+
+ std::shared_ptr<std::promise<T>> getPromise();
+
+private:
+ std::function<T()> task;
+ std::shared_ptr<std::promise<T>> promise;
+};
+
+template< typename T>
+Worker<T>& Worker<T>::operator = (Worker<T>&& other)
+{
+ task = std::move(other.task);
+ promise = other.promise;
+ return *this;
+}
+
+
+template<typename T>
+std::shared_ptr<std::promise<T>> Worker<T>::getPromise(){
+ return promise;
+ }
+
+/**
+ * Thread pool
+ * Purpose: Provides a thread pool with basic functionality similar to
+ * ThreadPoolExecutor
+ * Design: Locked control over a manager thread that controls the worker threads
+ */
+template<typename T>
+class ThreadPool
+ {
+ public:
+ ThreadPool(int max_worker_threads, bool daemon_threads=false) : max_worker_threads_(max_worker_threads)
+ ,daemon_threads_(daemon_threads), running_(false){
+ current_workers_ = 0;
+ }
+ virtual ~ThreadPool(){
+ shutdown();
+ }
+
+ /**
+ * Execute accepts a worker task and returns
+ * a future
+ * @param task this thread pool will subsume ownership of
+ * the worker task
+ * @return future with the impending result.
+ */
+ std::future<T> execute(Worker<T> &&task);
+ /**
+ * Starts the Thread Pool
+ */
+ void start();
+ /**
+ * Shutdown the thread pool and clear any
+ * currently running activities
+ */
+ void shutdown();
+ /**
+ * Set the max concurrent tasks. When this is done
+ * we must start and restart the thread pool if
+ * the number of tasks is less than the currently configured number
+ */
+ void setMaxConcurrentTasks(uint16_t max)
+ {
+ std::lock_guard<std::recursive_mutex> lock(manager_mutex_);
+ if (running_)
+ {
+ shutdown();
+ }
+ max_worker_threads_= max;
+ if (!running_)
+ start();
+ }
+
+ protected:
+
+ /**
+ * Drain will notify tasks to stop following notification
+ */
+ void drain()
+ {
+ while(current_workers_ > 0)
+ {
+ tasks_available_.notify_one();
+ }
+ }
+ // determines if threads are detached
+ bool daemon_threads_;
+ // max worker threads
+ int max_worker_threads_;
+ // current worker tasks.
+ std::atomic<int> current_workers_;
+ // thread queue
+ std::vector<std::thread> thread_queue_;
+ // manager thread
+ std::thread manager_thread_;
+ // atomic running boolean
+ std::atomic<bool> running_;
+ // worker queue of worker objects
+ std::queue<Worker<T>> worker_queue_;
+ // notification for available work
+ std::condition_variable tasks_available_;
+ // manager mutex
+ std::recursive_mutex manager_mutex_;
+ // work queue mutex
+ std::mutex worker_queue_mutex_;
+
+ /**
+ * Call for the manager to start worker threads
+ */
+ void startWorkers();
+
+ /**
+ * Runs worker tasks
+ */
+ void run_tasks();
+ };
+
+template<typename T>
+std::future<T> ThreadPool<T>::execute(Worker<T> &&task){
+
+ std::unique_lock<std::mutex> lock(worker_queue_mutex_);
+ bool wasEmpty = worker_queue_.empty();
+ std::future<T> future = task.getPromise()->get_future();
+ worker_queue_.push(std::move(task));
+ if (wasEmpty)
+ {
+ tasks_available_.notify_one();
+ }
+ return future;
+}
+
+template< typename T>
+void ThreadPool<T>::startWorkers(){
+ for (int i = 0; i < max_worker_threads_; i++)
+ {
+ thread_queue_.push_back( std::thread(&ThreadPool::run_tasks, this));
+ current_workers_++;
+ }
+
+ if (daemon_threads_)
+ {
+ for (auto &thread : thread_queue_){
+ thread.detach();
+ }
+ }
+ for (auto &thread : thread_queue_)
+ {
+ if (thread.joinable())
+ thread.join();
+ }
+}
+template< typename T>
+void ThreadPool<T>::run_tasks()
+{
+ while (running_.load())
+ {
+ std::unique_lock<std::mutex> lock(worker_queue_mutex_);
+ if (worker_queue_.empty())
+ {
+
+ tasks_available_.wait(lock);
+ }
+
+ if (!running_.load())
+ break;
+
+ if (worker_queue_.empty())
+ continue;
+ Worker<T> task = std::move(worker_queue_.front());
+ worker_queue_.pop();
+ task.run();
+ }
+ current_workers_--;
+
+}
+template< typename T>
+ void ThreadPool<T>::start()
+{
+ std::lock_guard<std::recursive_mutex> lock(manager_mutex_);
+ if (!running_)
+ {
+ running_ = true;
+ manager_thread_ = std::thread(&ThreadPool::startWorkers, this);
+
+ }
+}
+
+template< typename T>
+void ThreadPool<T>::shutdown(){
+
+ std::lock_guard<std::recursive_mutex> lock(manager_mutex_);
+ if (running_.load())
+ {
+
+ running_.store(false);
+
+ drain();
+ if (manager_thread_.joinable())
+ manager_thread_.join();
+ {
+ std::unique_lock<std::mutex> lock(worker_queue_mutex_);
+ thread_queue_.clear();
+ current_workers_ = 0;
+ while(!worker_queue_.empty())
+ worker_queue_.pop();
+ }
+ }
+}
+
+
+} /* namespace utils */
+} /* namespace minifi */
+} /* namespace nifi */
+} /* namespace apache */
+} /* namespace org */
+
+
+#endif
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/3676468f/libminifi/src/core/ConfigurableComponent.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/core/ConfigurableComponent.cpp b/libminifi/src/core/ConfigurableComponent.cpp
index e5703d1..67a43dd 100644
--- a/libminifi/src/core/ConfigurableComponent.cpp
+++ b/libminifi/src/core/ConfigurableComponent.cpp
@@ -28,14 +28,14 @@ namespace minifi {
namespace core {
ConfigurableComponent::ConfigurableComponent(std::shared_ptr<logging::Logger> logger)
- : logger_(logger) {
+ : my_logger_(logger) {
}
ConfigurableComponent::ConfigurableComponent(
const ConfigurableComponent &&other)
: properties_(std::move(other.properties_)),
- logger_(std::move(other.logger_)) {
+ my_logger_(std::move(other.my_logger_)) {
}
ConfigurableComponent::~ConfigurableComponent() {
@@ -57,7 +57,7 @@ bool ConfigurableComponent::getProperty(const std::string name,
if (it != properties_.end()) {
Property item = it->second;
value = item.getValue();
- logger_->log_info("Processor %s property name %s value %s", name.c_str(),
+ my_logger_->log_info("Processor %s property name %s value %s", name.c_str(),
item.getName().c_str(), value.c_str());
return true;
} else {
@@ -79,7 +79,7 @@ bool ConfigurableComponent::setProperty(const std::string name,
Property item = it->second;
item.setValue(value);
properties_[item.getName()] = item;
- logger_->log_info("Component %s property name %s value %s", name.c_str(),
+ my_logger_->log_info("Component %s property name %s value %s", name.c_str(),
item.getName().c_str(), value.c_str());
return true;
} else {
@@ -101,7 +101,7 @@ bool ConfigurableComponent::setProperty(Property &prop, std::string value) {
Property item = it->second;
item.setValue(value);
properties_[item.getName()] = item;
- logger_->log_info("property name %s value %s", prop.getName().c_str(),
+ my_logger_->log_info("property name %s value %s", prop.getName().c_str(),
item.getName().c_str(), value.c_str());
return true;
} else {
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/3676468f/libminifi/src/core/FlowConfiguration.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/core/FlowConfiguration.cpp b/libminifi/src/core/FlowConfiguration.cpp
index c6472cc..f2dda0d 100644
--- a/libminifi/src/core/FlowConfiguration.cpp
+++ b/libminifi/src/core/FlowConfiguration.cpp
@@ -64,9 +64,9 @@ std::shared_ptr<core::Processor> FlowConfiguration::createProcessor(
processor = std::make_shared<
org::apache::nifi::minifi::processors::ListenSyslog>(name, uuid);
} else if (name
- == org::apache::nifi::minifi::processors::ListenHTTP::ProcessorName) {
- processor = std::make_shared<
- org::apache::nifi::minifi::processors::ListenHTTP>(name, uuid);
+ == org::apache::nifi::minifi::processors::ListenHTTP::ProcessorName) {
+ processor = std::make_shared<
+ org::apache::nifi::minifi::processors::ListenHTTP>(name, uuid);
} else if (name
== org::apache::nifi::minifi::processors::ExecuteProcess::ProcessorName) {
processor = std::make_shared<
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/3676468f/libminifi/src/core/ProcessSession.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/core/ProcessSession.cpp b/libminifi/src/core/ProcessSession.cpp
index e6fa7c4..09c3fa3 100644
--- a/libminifi/src/core/ProcessSession.cpp
+++ b/libminifi/src/core/ProcessSession.cpp
@@ -50,7 +50,7 @@ std::shared_ptr<core::FlowFile> ProcessSession::create() {
return record;
}
-std::shared_ptr<core::FlowFile> ProcessSession::create(std::shared_ptr<core::FlowFile> &parent) {
+std::shared_ptr<core::FlowFile> ProcessSession::create(std::shared_ptr<core::FlowFile> &&parent) {
std::map<std::string, std::string> empty;
std::shared_ptr<core::FlowFile> record = std::make_shared<FlowFileRecord>(process_context_->getProvenanceRepository(),
empty);
@@ -530,6 +530,97 @@ void ProcessSession::read(std::shared_ptr<core::FlowFile> &&flow,
}
}
+/**
+ * Imports a file from the data stream
+ * @param stream incoming data stream that contains the data to store into a file
+ * @param flow flow file
+ *
+ */
+void ProcessSession::importFrom(io::DataStream &stream,
+ std::shared_ptr<core::FlowFile> &&flow) {
+ std::shared_ptr<ResourceClaim> claim = std::make_shared<ResourceClaim>();
+
+ int max_read = getpagesize();
+ std::vector<uint8_t> charBuffer;
+ charBuffer.resize(max_read);
+
+ try {
+ std::ofstream fs;
+ uint64_t startTime = getTimeMillis();
+ fs.open(claim->getContentFullPath().c_str(),
+ std::fstream::out | std::fstream::binary | std::fstream::trunc);
+
+
+ if (fs.is_open() ) {
+
+ size_t position = 0;
+ const size_t max_size = stream.getSize();
+ size_t read_size = max_read;
+ while(position < max_size)
+ {
+ if ((max_size - position) > max_read)
+ {
+ read_size = max_read;
+ }
+ else
+ {
+ read_size = max_size - position;
+ }
+ charBuffer.clear();
+ stream.readData(charBuffer,read_size);
+
+ fs.write((const char*)charBuffer.data(),read_size);
+ position+=read_size;
+ }
+ // Open the source file and stream to the flow file
+
+ if (fs.good() && fs.tellp() >= 0) {
+ flow->setSize(fs.tellp());
+ flow->setOffset(0);
+ if (flow->getResourceClaim() != nullptr) {
+ // Remove the old claim
+ flow->getResourceClaim()->decreaseFlowFileRecordOwnedCount();
+ flow->clearResourceClaim();
+ }
+ flow->setResourceClaim(claim);
+ claim->increaseFlowFileRecordOwnedCount();
+
+ logger_->log_debug(
+ "Import offset %d length %d into content %s for FlowFile UUID %s",
+ flow->getOffset(), flow->getSize(),
+ flow->getResourceClaim()->getContentFullPath().c_str(),
+ flow->getUUIDStr().c_str());
+
+ fs.close();
+ std::string details = process_context_->getProcessorNode().getName()
+ + " modify flow record content " + flow->getUUIDStr();
+ uint64_t endTime = getTimeMillis();
+ provenance_report_->modifyContent(flow, details, endTime - startTime);
+ } else {
+ fs.close();
+ throw Exception(FILE_OPERATION_EXCEPTION, "File Import Error");
+ }
+ } else {
+ throw Exception(FILE_OPERATION_EXCEPTION, "File Import Error");
+ }
+
+ } catch (std::exception &exception) {
+ if (flow && flow->getResourceClaim() == claim) {
+ flow->getResourceClaim()->decreaseFlowFileRecordOwnedCount();
+ flow->clearResourceClaim();
+ }
+ logger_->log_debug("Caught Exception %s", exception.what());
+ throw;
+ } catch (...) {
+ if (flow && flow->getResourceClaim() == claim) {
+ flow->getResourceClaim()->decreaseFlowFileRecordOwnedCount();
+ flow->clearResourceClaim();
+ }
+ logger_->log_debug("Caught Exception during process session write");
+ throw;
+ }
+}
+
void ProcessSession::import(std::string source,
std::shared_ptr<core::FlowFile> &flow,
bool keepSource, uint64_t offset) {
@@ -639,6 +730,7 @@ void ProcessSession::import(std::string source,
fs.write(buf, input.gcount());
}
+
if (fs.good() && fs.tellp() >= 0) {
flow->setSize(fs.tellp());
flow->setOffset(0);
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/3676468f/libminifi/src/processors/GetFile.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/processors/GetFile.cpp b/libminifi/src/processors/GetFile.cpp
index cf05657..652caf7 100644
--- a/libminifi/src/processors/GetFile.cpp
+++ b/libminifi/src/processors/GetFile.cpp
@@ -1,6 +1,4 @@
/**
- * @file GetFile.cpp
- * GetFile class implementation
*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
@@ -87,8 +85,8 @@ core::Property GetFile::FileFilter(
"File Filter",
"Only files whose names match the given regular expression will be picked up",
"[^\\.].*");
-core::Relationship GetFile::Success(
- "success", "All files are routed to success");
+core::Relationship GetFile::Success("success",
+ "All files are routed to success");
void GetFile::initialize() {
// Set the supported properties
@@ -111,77 +109,78 @@ void GetFile::initialize() {
setSupportedRelationships(relationships);
}
-void GetFile::onTrigger(
- core::ProcessContext *context,
- core::ProcessSession *session) {
+void GetFile::onSchedule(core::ProcessContext *context,
+ core::ProcessSessionFactory *sessionFactory) {
std::string value;
logger_->log_info("onTrigger GetFile");
if (context->getProperty(Directory.getName(), value)) {
- _directory = value;
+ request_.directory = value;
}
if (context->getProperty(BatchSize.getName(), value)) {
- core::Property::StringToInt(value, _batchSize);
+ core::Property::StringToInt(value, request_.batchSize);
}
if (context->getProperty(IgnoreHiddenFile.getName(), value)) {
org::apache::nifi::minifi::utils::StringUtils::StringToBool(
- value, _ignoreHiddenFile);
+ value, request_.ignoreHiddenFile);
}
if (context->getProperty(KeepSourceFile.getName(), value)) {
org::apache::nifi::minifi::utils::StringUtils::StringToBool(
- value, _keepSourceFile);
+ value, request_.keepSourceFile);
}
logger_->log_info("onTrigger GetFile");
if (context->getProperty(MaxAge.getName(), value)) {
core::TimeUnit unit;
- if (core::Property::StringToTime(value, _maxAge,
- unit)
- && core::Property::ConvertTimeUnitToMS(
- _maxAge, unit, _maxAge)) {
+ if (core::Property::StringToTime(value, request_.maxAge, unit)
+ && core::Property::ConvertTimeUnitToMS(request_.maxAge, unit,
+ request_.maxAge)) {
}
}
if (context->getProperty(MinAge.getName(), value)) {
core::TimeUnit unit;
- if (core::Property::StringToTime(value, _minAge,
- unit)
- && core::Property::ConvertTimeUnitToMS(
- _minAge, unit, _minAge)) {
+ if (core::Property::StringToTime(value, request_.minAge, unit)
+ && core::Property::ConvertTimeUnitToMS(request_.minAge, unit,
+ request_.minAge)) {
}
}
if (context->getProperty(MaxSize.getName(), value)) {
- core::Property::StringToInt(value, _maxSize);
+ core::Property::StringToInt(value, request_.maxSize);
}
if (context->getProperty(MinSize.getName(), value)) {
- core::Property::StringToInt(value, _minSize);
+ core::Property::StringToInt(value, request_.minSize);
}
if (context->getProperty(PollInterval.getName(), value)) {
core::TimeUnit unit;
- if (core::Property::StringToTime(value,
- _pollInterval,
- unit)
- && core::Property::ConvertTimeUnitToMS(
- _pollInterval, unit, _pollInterval)) {
+ if (core::Property::StringToTime(value, request_.pollInterval, unit)
+ && core::Property::ConvertTimeUnitToMS(request_.pollInterval, unit,
+ request_.pollInterval)) {
}
}
if (context->getProperty(Recurse.getName(), value)) {
- org::apache::nifi::minifi::utils::StringUtils::StringToBool(value,
- _recursive);
+ org::apache::nifi::minifi::utils::StringUtils::StringToBool(
+ value, request_.recursive);
}
if (context->getProperty(FileFilter.getName(), value)) {
- _fileFilter = value;
+ request_.fileFilter = value;
}
+}
+
+void GetFile::onTrigger(core::ProcessContext *context,
+ core::ProcessSession *session) {
// Perform directory list
logger_->log_info("Is listing empty %i", isListingEmpty());
if (isListingEmpty()) {
- if (_pollInterval == 0
- || (getTimeMillis() - _lastDirectoryListingTime) > _pollInterval) {
- performListing(_directory);
+
+ if (request_.pollInterval == 0
+ || (getTimeMillis() - last_listing_time_) > request_.pollInterval) {
+ performListing(request_.directory, request_);
+ last_listing_time_.store(getTimeMillis());
}
}
logger_->log_info("Is listing empty %i", isListingEmpty());
@@ -189,7 +188,7 @@ void GetFile::onTrigger(
if (!isListingEmpty()) {
try {
std::queue<std::string> list;
- pollListing(list, _batchSize);
+ pollListing(list, request_);
while (!list.empty()) {
std::string fileName = list.front();
@@ -205,7 +204,7 @@ void GetFile::onTrigger(
flowFile->updateKeyedAttribute(FILENAME, name);
flowFile->updateKeyedAttribute(PATH, path);
flowFile->addKeyedAttribute(ABSOLUTE_PATH, fileName);
- session->import(fileName, flowFile, _keepSourceFile);
+ session->import(fileName, flowFile, request_.keepSourceFile);
session->transfer(flowFile, Success);
}
} catch (std::exception &exception) {
@@ -230,10 +229,12 @@ void GetFile::putListing(std::string fileName) {
_dirList.push(fileName);
}
-void GetFile::pollListing(std::queue<std::string> &list, int maxSize) {
+void GetFile::pollListing(std::queue<std::string> &list,
+ const GetFileRequest &request) {
std::lock_guard<std::mutex> lock(mutex_);
- while (!_dirList.empty() && (maxSize == 0 || list.size() < maxSize)) {
+ while (!_dirList.empty()
+ && (request.maxSize == 0 || list.size() < request.maxSize)) {
std::string fileName = _dirList.front();
_dirList.pop();
list.push(fileName);
@@ -242,37 +243,38 @@ void GetFile::pollListing(std::queue<std::string> &list, int maxSize) {
return;
}
-bool GetFile::acceptFile(std::string fullName, std::string name) {
+bool GetFile::acceptFile(std::string fullName, std::string name,
+ const GetFileRequest &request) {
struct stat statbuf;
if (stat(fullName.c_str(), &statbuf) == 0) {
- if (_minSize > 0 && statbuf.st_size < _minSize)
+ if (request.minSize > 0 && statbuf.st_size < request.minSize)
return false;
- if (_maxSize > 0 && statbuf.st_size > _maxSize)
+ if (request.maxSize > 0 && statbuf.st_size > request.maxSize)
return false;
uint64_t modifiedTime = ((uint64_t) (statbuf.st_mtime) * 1000);
uint64_t fileAge = getTimeMillis() - modifiedTime;
- if (_minAge > 0 && fileAge < _minAge)
+ if (request.minAge > 0 && fileAge < request.minAge)
return false;
- if (_maxAge > 0 && fileAge > _maxAge)
+ if (request.maxAge > 0 && fileAge > request.maxAge)
return false;
- if (_ignoreHiddenFile && fullName.c_str()[0] == '.')
+ if (request.ignoreHiddenFile && fullName.c_str()[0] == '.')
return false;
if (access(fullName.c_str(), R_OK) != 0)
return false;
- if (_keepSourceFile == false && access(fullName.c_str(), W_OK) != 0)
+ if (request.keepSourceFile == false && access(fullName.c_str(), W_OK) != 0)
return false;
#ifdef __GNUC__
#if (__GNUC__ >= 4)
#if (__GNUC_MINOR__ < 9)
regex_t regex;
- int ret = regcomp(®ex, _fileFilter.c_str(), 0);
+ int ret = regcomp(®ex, request.fileFilter.c_str(), 0);
if (ret)
return false;
ret = regexec(®ex, name.c_str(), (size_t) 0, NULL, 0);
@@ -281,7 +283,7 @@ bool GetFile::acceptFile(std::string fullName, std::string name) {
return false;
#else
try {
- std::regex re(_fileFilter);
+ std::regex re(fileFilter);
if (!std::regex_match(name, re)) {
return false;
@@ -301,7 +303,7 @@ bool GetFile::acceptFile(std::string fullName, std::string name) {
return false;
}
-void GetFile::performListing(std::string dir) {
+void GetFile::performListing(std::string dir, const GetFileRequest &request) {
logger_->log_info("Performing file listing against %s", dir.c_str());
DIR *d;
d = opendir(dir.c_str());
@@ -317,14 +319,14 @@ void GetFile::performListing(std::string dir) {
std::string d_name = entry->d_name;
if ((entry->d_type & DT_DIR)) {
// if this is a directory
- if (_recursive && strcmp(d_name.c_str(), "..") != 0
+ if (request.recursive && strcmp(d_name.c_str(), "..") != 0
&& strcmp(d_name.c_str(), ".") != 0) {
std::string path = dir + "/" + d_name;
- performListing(path);
+ performListing(path, request);
}
} else {
std::string fileName = dir + "/" + d_name;
- if (acceptFile(fileName, d_name)) {
+ if (acceptFile(fileName, d_name, request)) {
// check whether we can take this file
putListing(fileName);
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/3676468f/libminifi/src/processors/PutFile.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/processors/PutFile.cpp b/libminifi/src/processors/PutFile.cpp
index 85cf09b..51fbb6f 100644
--- a/libminifi/src/processors/PutFile.cpp
+++ b/libminifi/src/processors/PutFile.cpp
@@ -24,13 +24,13 @@
#include <fstream>
#include <uuid/uuid.h>
+#include "io/validation.h"
#include "utils/StringUtils.h"
#include "utils/TimeUtil.h"
#include "processors/PutFile.h"
#include "core/ProcessContext.h"
#include "core/ProcessSession.h"
-
namespace org {
namespace apache {
namespace nifi {
@@ -43,15 +43,16 @@ const std::string PutFile::CONFLICT_RESOLUTION_STRATEGY_FAIL("fail");
const std::string PutFile::ProcessorName("PutFile");
-core::Property PutFile::Directory(
- "Output Directory", "The output directory to which to put files", ".");
+core::Property PutFile::Directory("Output Directory",
+ "The output directory to which to put files",
+ ".");
core::Property PutFile::ConflictResolution(
"Conflict Resolution Strategy",
"Indicates what should happen when a file with the same name already exists in the output directory",
CONFLICT_RESOLUTION_STRATEGY_FAIL);
-core::Relationship PutFile::Success(
- "success", "All files are routed to success");
+core::Relationship PutFile::Success("success",
+ "All files are routed to success");
core::Relationship PutFile::Failure(
"failure",
"Failed files (conflict, write failure, etc.) are transferred to failure");
@@ -69,25 +70,30 @@ void PutFile::initialize() {
setSupportedRelationships(relationships);
}
-void PutFile::onTrigger(
- core::ProcessContext *context,
- core::ProcessSession *session) {
- std::string directory;
+void PutFile::onSchedule(core::ProcessContext *context,
+ core::ProcessSessionFactory *sessionFactory) {
- if (!context->getProperty(Directory.getName(), directory)) {
+ if (!context->getProperty(Directory.getName(), directory_)) {
logger_->log_error("Directory attribute is missing or invalid");
- return;
}
- std::string conflictResolution;
-
- if (!context->getProperty(ConflictResolution.getName(), conflictResolution)) {
+ if (!context->getProperty(ConflictResolution.getName(),
+ conflict_resolution_)) {
logger_->log_error(
"Conflict Resolution Strategy attribute is missing or invalid");
+ }
+
+}
+void PutFile::onTrigger(core::ProcessContext *context,
+ core::ProcessSession *session) {
+
+ if (IsNullOrEmpty(directory_) || IsNullOrEmpty(conflict_resolution_)) {
+ context->yield();
return;
}
- std::shared_ptr<FlowFileRecord> flowFile = std::static_pointer_cast<FlowFileRecord>(session->get());
+ std::shared_ptr<FlowFileRecord> flowFile = std::static_pointer_cast<
+ FlowFileRecord>(session->get());
// Do nothing if there are no incoming files
if (!flowFile) {
@@ -103,17 +109,17 @@ void PutFile::onTrigger(
uuid_generate(tmpFileUuid);
uuid_unparse_lower(tmpFileUuid, tmpFileUuidStr);
std::stringstream tmpFileSs;
- tmpFileSs << directory << "/." << filename << "." << tmpFileUuidStr;
+ tmpFileSs << directory_ << "/." << filename << "." << tmpFileUuidStr;
std::string tmpFile = tmpFileSs.str();
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());
+ filename.c_str(), directory_.c_str());
// If file exists, apply conflict resolution strategy
struct stat statResult;
@@ -121,11 +127,11 @@ void PutFile::onTrigger(
if (stat(destFile.c_str(), &statResult) == 0) {
logger_->log_info(
"Destination file %s exists; applying Conflict Resolution Strategy: %s",
- destFile.c_str(), conflictResolution.c_str());
+ destFile.c_str(), conflict_resolution_.c_str());
- if (conflictResolution == CONFLICT_RESOLUTION_STRATEGY_REPLACE) {
+ if (conflict_resolution_ == CONFLICT_RESOLUTION_STRATEGY_REPLACE) {
putFile(session, flowFile, tmpFile, destFile);
- } else if (conflictResolution == CONFLICT_RESOLUTION_STRATEGY_IGNORE) {
+ } else if (conflict_resolution_ == CONFLICT_RESOLUTION_STRATEGY_IGNORE) {
session->transfer(flowFile, Success);
} else {
session->transfer(flowFile, Failure);
@@ -136,8 +142,8 @@ void PutFile::onTrigger(
}
bool PutFile::putFile(core::ProcessSession *session,
- std::shared_ptr<FlowFileRecord> flowFile, const std::string &tmpFile,
- const std::string &destFile) {
+ std::shared_ptr<FlowFileRecord> flowFile,
+ const std::string &tmpFile, const std::string &destFile) {
ReadCallback cb(tmpFile, destFile);
session->read(flowFile, &cb);
@@ -205,7 +211,6 @@ PutFile::ReadCallback::~ReadCallback() {
unlink(_tmpFile.c_str());
}
-
} /* namespace processors */
} /* namespace minifi */
} /* namespace nifi */
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/3676468f/libminifi/src/processors/TailFile.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/processors/TailFile.cpp b/libminifi/src/processors/TailFile.cpp
index 859daa6..bcdd8fd 100644
--- a/libminifi/src/processors/TailFile.cpp
+++ b/libminifi/src/processors/TailFile.cpp
@@ -145,10 +145,10 @@ static bool sortTailMatchedFileItem(TailMatchedFileItem i,
TailMatchedFileItem j) {
return (i.modifiedTime < j.modifiedTime);
}
-void TailFile::checkRollOver() {
+void TailFile::checkRollOver(const std::string &fileLocation, const std::string &fileName) {
struct stat statbuf;
std::vector<TailMatchedFileItem> matchedFiles;
- std::string fullPath = this->_fileLocation + "/" + _currentTailFileName;
+ std::string fullPath = fileLocation + "/" + _currentTailFileName;
if (stat(fullPath.c_str(), &statbuf) == 0) {
if (statbuf.st_size > this->_currentTailFilePosition)
@@ -157,12 +157,12 @@ void TailFile::checkRollOver() {
uint64_t modifiedTimeCurrentTailFile =
((uint64_t) (statbuf.st_mtime) * 1000);
- std::string pattern = _fileName;
- std::size_t found = _fileName.find_last_of(".");
+ std::string pattern = fileName;
+ std::size_t found = fileName.find_last_of(".");
if (found != std::string::npos)
- pattern = _fileName.substr(0, found);
+ pattern = fileName.substr(0, found);
DIR *d;
- d = opendir(this->_fileLocation.c_str());
+ d = opendir(fileLocation.c_str());
if (!d)
return;
while (1) {
@@ -173,7 +173,7 @@ void TailFile::checkRollOver() {
std::string d_name = entry->d_name;
if (!(entry->d_type & DT_DIR)) {
std::string fileName = d_name;
- std::string fileFullName = this->_fileLocation + "/" + d_name;
+ std::string fileFullName = fileLocation + "/" + d_name;
if (fileFullName.find(pattern) != std::string::npos
&& stat(fileFullName.c_str(), &statbuf) == 0) {
if (((uint64_t) (statbuf.st_mtime) * 1000)
@@ -215,24 +215,28 @@ void TailFile::checkRollOver() {
void TailFile::onTrigger(
core::ProcessContext *context,
core::ProcessSession *session) {
+
+ std::lock_guard<std::mutex> tail_lock(tail_file_mutex_);
std::string value;
+ std::string fileLocation="";
+ std::string fileName="";
if (context->getProperty(FileName.getName(), value)) {
std::size_t found = value.find_last_of("/\\");
- this->_fileLocation = value.substr(0, found);
- this->_fileName = value.substr(found + 1);
+ fileLocation = value.substr(0, found);
+ fileName = value.substr(found + 1);
}
if (context->getProperty(StateFile.getName(), value)) {
_stateFile = value + "." + getUUIDStr();
}
if (!this->_stateRecovered) {
_stateRecovered = true;
- this->_currentTailFileName = _fileName;
+ this->_currentTailFileName = fileName;
this->_currentTailFilePosition = 0;
// recover the state if we have not done so
this->recoverState();
}
- checkRollOver();
- std::string fullPath = this->_fileLocation + "/" + _currentTailFileName;
+ checkRollOver(fileLocation,fileName);
+ std::string fullPath = fileLocation + "/" + _currentTailFileName;
struct stat statbuf;
if (stat(fullPath.c_str(), &statbuf) == 0) {
if (statbuf.st_size <= this->_currentTailFilePosition)
@@ -241,13 +245,13 @@ void TailFile::onTrigger(
context->yield();
return;
}
- std::shared_ptr<FlowFileRecord> flowFile = std::static_pointer_cast<FlowFileRecord>(session->create());;
+ std::shared_ptr<FlowFileRecord> flowFile = std::static_pointer_cast<FlowFileRecord>(session->create());
if (!flowFile)
return;
std::size_t found = _currentTailFileName.find_last_of(".");
std::string baseName = _currentTailFileName.substr(0, found);
std::string extension = _currentTailFileName.substr(found + 1);
- flowFile->updateKeyedAttribute(PATH, _fileLocation);
+ flowFile->updateKeyedAttribute(PATH, fileLocation);
flowFile->addKeyedAttribute(ABSOLUTE_PATH, fullPath);
session->import(fullPath, flowFile, true, this->_currentTailFilePosition);
session->transfer(flowFile, Success);
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/3676468f/libminifi/test/TestExecuteProcess.cpp
----------------------------------------------------------------------
diff --git a/libminifi/test/TestExecuteProcess.cpp b/libminifi/test/TestExecuteProcess.cpp
new file mode 100644
index 0000000..6aa51fc
--- /dev/null
+++ b/libminifi/test/TestExecuteProcess.cpp
@@ -0,0 +1,131 @@
+/**
+ *
+ * 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 <uuid/uuid.h>
+#include <fstream>
+#include "FlowController.h"
+#include "unit/ProvenanceTestHelper.h"
+#include "core/logging/LogAppenders.h"
+#include "core/logging/BaseLogger.h"
+#include "processors/GetFile.h"
+#include "core/core.h"
+#include "core/FlowFile.h"
+#include "core/Processor.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+#include "core/ProcessorNode.h"
+
+int main(int argc, char **argv)
+{
+
+ std::ostringstream oss;
+ std::unique_ptr<logging::BaseLogger> outputLogger = std::unique_ptr<
+ logging::BaseLogger>(
+ new org::apache::nifi::minifi::core::logging::OutputStreamAppender(oss,
+ 0));
+ std::shared_ptr<logging::Logger> logger = logging::Logger::getLogger();
+ logger->updateLogger(std::move(outputLogger));
+
+
+ outputLogger = std::unique_ptr<logging::BaseLogger>(
+ new org::apache::nifi::minifi::core::logging::NullAppender());
+ logger->updateLogger(std::move(outputLogger));
+
+ std::shared_ptr<core::Processor> processor = std::make_shared<
+ org::apache::nifi::minifi::processors::ExecuteProcess>("executeProcess");
+ processor->setMaxConcurrentTasks(1);
+
+ std::shared_ptr<core::Repository> test_repo =
+ std::make_shared<TestRepository>();
+
+ std::shared_ptr<TestRepository> repo =
+ std::static_pointer_cast<TestRepository>(test_repo);
+ std::shared_ptr<minifi::FlowController> controller = std::make_shared<
+ TestFlowController>(test_repo, test_repo);
+
+ uuid_t processoruuid;
+ assert(true == processor->getUUID(processoruuid));
+
+ std::shared_ptr<minifi::Connection> connection = std::make_shared<
+ minifi::Connection>(test_repo, "executeProcessConnection");
+ connection->setRelationship(core::Relationship("success", "description"));
+
+ // link the connections so that we can test results at the end for this
+ connection->setSource(processor);
+ connection->setDestination(processor);
+
+ connection->setSourceUUID(processoruuid);
+ connection->setDestinationUUID(processoruuid);
+
+ processor->addConnection(connection);
+ assert(processor->getName() == "executeProcess");
+
+ std::shared_ptr<core::FlowFile> record;
+ processor->setScheduledState(core::ScheduledState::RUNNING);
+
+ processor->initialize();
+
+ std::atomic<bool> is_ready(false);
+
+ std::vector<std::thread> processor_workers;
+
+ core::ProcessorNode node2(processor);
+ std::shared_ptr<core::ProcessContext> contextset = std::make_shared<
+ core::ProcessContext>(node2, test_repo);
+ core::ProcessSessionFactory factory(contextset.get());
+ processor->onSchedule(contextset.get(), &factory);
+
+ for (int i = 0; i < 1; i++) {
+ //
+ processor_workers.push_back(
+ std::thread(
+ [processor,test_repo,&is_ready]()
+ {
+ core::ProcessorNode node(processor);
+ std::shared_ptr<core::ProcessContext> context = std::make_shared<core::ProcessContext>(node, test_repo);
+ context->setProperty(org::apache::nifi::minifi::processors::ExecuteProcess::Command,"sleep 0.5");
+ //context->setProperty(org::apache::nifi::minifi::processors::ExecuteProcess::CommandArguments," 1 >>" + ss.str());
+ std::shared_ptr<core::ProcessSession> session = std::make_shared<core::ProcessSession>(context.get());
+ while(!is_ready.load(std::memory_order_relaxed)) {
+
+ }
+
+ processor->onTrigger(context.get(), session.get());
+
+ }));
+ }
+
+ is_ready.store(true, std::memory_order_relaxed);
+ //is_ready.store(true);
+
+ std::for_each(processor_workers.begin(), processor_workers.end(),
+ [](std::thread &t)
+ {
+ t.join();
+ });
+
+ outputLogger = std::unique_ptr<logging::BaseLogger>(
+ new org::apache::nifi::minifi::core::logging::NullAppender());
+ logger->updateLogger(std::move(outputLogger));
+
+
+ std::shared_ptr<org::apache::nifi::minifi::processors::ExecuteProcess> execp =
+ std::static_pointer_cast<
+ org::apache::nifi::minifi::processors::ExecuteProcess>(processor);
+
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/3676468f/libminifi/test/unit/ProcessorTests.cpp
----------------------------------------------------------------------
diff --git a/libminifi/test/unit/ProcessorTests.cpp b/libminifi/test/unit/ProcessorTests.cpp
index c040e4d..91a55f7 100644
--- a/libminifi/test/unit/ProcessorTests.cpp
+++ b/libminifi/test/unit/ProcessorTests.cpp
@@ -41,17 +41,16 @@ TEST_CASE("Test Find file", "[getfileCreate2]") {
testController.enableDebug();
-
-
std::shared_ptr<core::Processor> processor = std::make_shared<
org::apache::nifi::minifi::processors::GetFile>("getfileCreate2");
- std::shared_ptr<core::Repository> test_repo = std::make_shared<TestRepository>();
-
- std::shared_ptr<TestRepository> repo = std::static_pointer_cast<TestRepository>(test_repo);
- std::shared_ptr<minifi::FlowController> controller = std::make_shared<TestFlowController>(test_repo, test_repo);
+ std::shared_ptr<core::Repository> test_repo =
+ std::make_shared<TestRepository>();
-
+ std::shared_ptr<TestRepository> repo =
+ std::static_pointer_cast<TestRepository>(test_repo);
+ std::shared_ptr<minifi::FlowController> controller = std::make_shared<
+ TestFlowController>(test_repo, test_repo);
char format[] = "/tmp/gt.XXXXXX";
char *dir = testController.createTempDirectory(format);
@@ -59,9 +58,8 @@ TEST_CASE("Test Find file", "[getfileCreate2]") {
uuid_t processoruuid;
REQUIRE(true == processor->getUUID(processoruuid));
-
std::shared_ptr<minifi::Connection> connection = std::make_shared<
- minifi::Connection>(test_repo,"getfileCreate2Connection");
+ minifi::Connection>(test_repo, "getfileCreate2Connection");
connection->setRelationship(core::Relationship("success", "description"));
// link the connections so that we can test results at the end for this
@@ -77,11 +75,12 @@ TEST_CASE("Test Find file", "[getfileCreate2]") {
core::ProcessorNode node(processor);
core::ProcessContext context(node, test_repo);
+ core::ProcessSessionFactory factory(&context);
context.setProperty(org::apache::nifi::minifi::processors::GetFile::Directory,
dir);
core::ProcessSession session(&context);
-
+ processor->onSchedule(&context, &factory);
REQUIRE(processor->getName() == "getfileCreate2");
std::shared_ptr<core::FlowFile> record;
@@ -107,7 +106,7 @@ TEST_CASE("Test Find file", "[getfileCreate2]") {
unlink(ss.str().c_str());
reporter = session.getProvenanceReporter();
- REQUIRE( processor->getName() == "getfileCreate2");
+ REQUIRE(processor->getName() == "getfileCreate2");
records = reporter->getEvents();
@@ -140,7 +139,104 @@ TEST_CASE("Test Find file", "[getfileCreate2]") {
if (!found)
throw std::runtime_error("Did not find record");
+ }
+
+}
+
+TEST_CASE("Test GetFileLikeIt'sThreaded", "[getfileCreate3]") {
+
+ TestController testController;
+
+ testController.enableDebug();
+
+ std::shared_ptr<core::Processor> processor = std::make_shared<
+ org::apache::nifi::minifi::processors::GetFile>("getfileCreate2");
+
+ std::shared_ptr<core::Repository> test_repo =
+ std::make_shared<TestRepository>();
+
+ std::shared_ptr<TestRepository> repo =
+ std::static_pointer_cast<TestRepository>(test_repo);
+ std::shared_ptr<minifi::FlowController> controller = std::make_shared<
+ TestFlowController>(test_repo, test_repo);
+
+ char format[] = "/tmp/gt.XXXXXX";
+ char *dir = testController.createTempDirectory(format);
+
+ uuid_t processoruuid;
+ REQUIRE(true == processor->getUUID(processoruuid));
+
+ std::shared_ptr<minifi::Connection> connection = std::make_shared<
+ minifi::Connection>(test_repo, "getfileCreate2Connection");
+ connection->setRelationship(core::Relationship("success", "description"));
+
+ // link the connections so that we can test results at the end for this
+ connection->setSource(processor);
+ connection->setDestination(processor);
+
+ connection->setSourceUUID(processoruuid);
+ connection->setDestinationUUID(processoruuid);
+
+ processor->addConnection(connection);
+ REQUIRE(dir != NULL);
+
+ core::ProcessorNode node(processor);
+ core::ProcessContext context(node, test_repo);
+ core::ProcessSessionFactory factory(&context);
+ context.setProperty(org::apache::nifi::minifi::processors::GetFile::Directory,
+ dir);
+ // replicate 10 threads
+ processor->setScheduledState(core::ScheduledState::RUNNING);
+ processor->onSchedule(&context, &factory);
+
+ for (int i = 0; i < 10; i++) {
+
+ core::ProcessSession session(&context);
+ REQUIRE(processor->getName() == "getfileCreate2");
+
+ std::shared_ptr<core::FlowFile> record;
+
+ processor->onTrigger(&context, &session);
+
+ provenance::ProvenanceReporter *reporter = session.getProvenanceReporter();
+ std::set<provenance::ProvenanceEventRecord*> records =
+ reporter->getEvents();
+ record = session.get();
+ REQUIRE(record == nullptr);
+ REQUIRE(records.size() == 0);
+
+ std::fstream file;
+ std::stringstream ss;
+ ss << dir << "/" << "tstFile.ext";
+ file.open(ss.str(), std::ios::out);
+ file << "tempFile";
+ file.close();
+ processor->incrementActiveTasks();
+ processor->setScheduledState(core::ScheduledState::RUNNING);
+ processor->onTrigger(&context, &session);
+ unlink(ss.str().c_str());
+ rmdir(dir);
+ reporter = session.getProvenanceReporter();
+
+ REQUIRE(processor->getName() == "getfileCreate2");
+
+ records = reporter->getEvents();
+
+ for (provenance::ProvenanceEventRecord *provEventRecord : records) {
+ REQUIRE(provEventRecord->getComponentType() == processor->getName());
+ }
+ session.commit();
+ std::shared_ptr<core::FlowFile> ffr = session.get();
+
+ REQUIRE(2 == repo->getRepoMap().size());
+
+ for (auto entry : repo->getRepoMap()) {
+ provenance::ProvenanceEventRecord newRecord;
+ newRecord.DeSerialize((uint8_t*) entry.second.data(),
+ entry.second.length());
+
+ }
}
}
@@ -158,9 +254,7 @@ TEST_CASE("LogAttributeTest", "[getfileCreate3]") {
testController.enableDebug();
-
- std::shared_ptr<core::Repository> repo = std::make_shared<
- TestRepository>();
+ std::shared_ptr<core::Repository> repo = std::make_shared<TestRepository>();
std::shared_ptr<core::Processor> processor = std::make_shared<
org::apache::nifi::minifi::processors::GetFile>("getfileCreate2");
@@ -177,25 +271,22 @@ TEST_CASE("LogAttributeTest", "[getfileCreate3]") {
uuid_t logattribute_uuid;
REQUIRE(true == logAttribute->getUUID(logattribute_uuid));
-
std::shared_ptr<minifi::Connection> connection = std::make_shared<
- minifi::Connection>(repo,"getfileCreate2Connection");
+ minifi::Connection>(repo, "getfileCreate2Connection");
connection->setRelationship(core::Relationship("success", "description"));
std::shared_ptr<minifi::Connection> connection2 = std::make_shared<
- minifi::Connection>(repo,"logattribute");
+ minifi::Connection>(repo, "logattribute");
connection2->setRelationship(core::Relationship("success", "description"));
// link the connections so that we can test results at the end for this
connection->setSource(processor);
-
// link the connections so that we can test results at the end for this
connection->setDestination(logAttribute);
connection2->setSource(logAttribute);
-
connection2->setSourceUUID(logattribute_uuid);
connection->setSourceUUID(processoruuid);
connection->setDestinationUUID(logattribute_uuid);
@@ -219,10 +310,15 @@ TEST_CASE("LogAttributeTest", "[getfileCreate3]") {
std::shared_ptr<core::FlowFile> record;
processor->setScheduledState(core::ScheduledState::RUNNING);
+
+ core::ProcessSessionFactory factory(&context);
+ processor->onSchedule(&context, &factory);
processor->onTrigger(&context, &session);
logAttribute->incrementActiveTasks();
logAttribute->setScheduledState(core::ScheduledState::RUNNING);
+ core::ProcessSessionFactory factory2(&context2);
+ logAttribute->onSchedule(&context2, &factory2);
logAttribute->onTrigger(&context2, &session2);
provenance::ProvenanceReporter *reporter = session.getProvenanceReporter();
@@ -270,3 +366,13 @@ TEST_CASE("LogAttributeTest", "[getfileCreate3]") {
logger->updateLogger(std::move(outputLogger));
}
+
+int fileSize(const char *add) {
+ std::ifstream mySource;
+ mySource.open(add, std::ios_base::binary);
+ mySource.seekg(0, std::ios_base::end);
+ int size = mySource.tellg();
+ mySource.close();
+ return size;
+}
+