You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ab...@apache.org on 2020/04/14 13:33:37 UTC
[nifi-minifi-cpp] branch master updated: MINIFICPP-1119 unify
win/posix sockets
This is an automated email from the ASF dual-hosted git repository.
aboda pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
The following commit(s) were added to refs/heads/master by this push:
new 3116eb8 MINIFICPP-1119 unify win/posix sockets
3116eb8 is described below
commit 3116eb817a7182486bf3e50b87c3032ac2074031
Author: Marton Szasz <sz...@gmail.com>
AuthorDate: Mon Feb 10 20:10:48 2020 +0100
MINIFICPP-1119 unify win/posix sockets
This closes #713
Squashed commit of the following:
commit 3bcf17a30750c655b1b77b701eab6bd2ebe969aa
Author: Marton Szasz <sz...@gmail.com>
Date: Fri Feb 7 12:17:11 2020 -0500
MINIFICPP-1119 ClientSocket/win: move lib pragma to .cpps, iwyu
commit ebe03abb40d193cf4ec57efb5751bfe6b4b6e862
Author: Marton Szasz <sz...@gmail.com>
Date: Thu Feb 6 17:12:35 2020 +0100
MINIFICPP-1119 fix compilation error on windows
commit 0d82b44cdd4cb99667bd7a203ce95ba82f0419b3
Author: Marton Szasz <sz...@gmail.com>
Date: Tue Feb 4 15:55:02 2020 +0100
MINIFICPP-1119 review feedback, simplify init, remove insane code
commit 5b1d2ece436ffdd938695278d83a028fa3576804
Author: Marton Szasz <sz...@gmail.com>
Date: Fri Jan 24 19:26:02 2020 +0100
MINIFICPP-1119 exclude ClientSocket from CPPLINT
commit f3606494a0bbda98e8c082da6c4de20a1a424da3
Author: Marton Szasz <sz...@gmail.com>
Date: Fri Jan 24 17:04:03 2020 +0000
MINIFICPP-1119 add missing error check to writeData
commit a702a153f82153bb9ef62250680ac9d05d99c896
Author: Marton Szasz <sz...@gmail.com>
Date: Thu Jan 23 16:53:56 2020 +0000
MINIFICPP-1119 fix some windows issues
commit 2d8ac695eadd7121c377414ee64a5952ca00a755
Author: Marton Szasz <sz...@gmail.com>
Date: Thu Jan 23 17:42:43 2020 +0100
MINIFICPP-1119 fix compilation issue on linux
commit 5cbf93022fcccb685e9895bf3248c181873bc3f3
Author: Marton Szasz <sz...@gmail.com>
Date: Tue Jan 21 18:41:59 2020 +0100
MINIFICPP-1119 unify win/posix sockets + clean up issues
MINIFICPP-1119 -1154 socket: fix name lookup, review feedback, IPv6
breaks: canonical name lookup for server sockets
Signed-off-by: Arpad Boda <ab...@apache.org>
---
extensions/coap/controllerservice/CoapConnector.h | 36 +-
extensions/coap/protocols/CoapC2Protocol.h | 10 +-
extensions/http-curl/client/HTTPStream.h | 10 +-
extensions/http-curl/tests/CivetStream.h | 17 +-
extensions/rocksdb-repos/RocksDbStream.cpp | 13 +-
extensions/rocksdb-repos/RocksDbStream.h | 28 +-
.../standard-processors/processors/PutFile.cpp | 3 +-
.../standard-processors/processors/PutFile.h | 22 +-
libminifi/include/core/ProcessSession.h | 7 +-
libminifi/include/core/PropertyValidation.h | 105 ++--
libminifi/include/core/PropertyValue.h | 28 +-
libminifi/include/core/StreamManager.h | 4 +-
libminifi/include/io/AtomicEntryStream.h | 16 +-
libminifi/include/io/BaseStream.h | 20 +-
libminifi/include/io/CRCStream.h | 39 +-
.../{opsys/posix => include}/io/ClientSocket.h | 146 +++--
libminifi/include/io/DataStream.h | 4 +-
libminifi/include/io/DescriptorStream.h | 30 +-
libminifi/include/io/FileStream.h | 16 +-
libminifi/include/io/NetworkPrioritizer.h | 57 +-
libminifi/include/io/NonConvertingStream.h | 36 +-
libminifi/include/io/Serializable.h | 2 +-
libminifi/include/io/ServerSocket.h | 9 +-
libminifi/include/io/StreamFactory.h | 9 +-
libminifi/include/io/tls/SecureDescriptorStream.h | 30 +-
libminifi/include/io/tls/TLSSocket.h | 41 +-
libminifi/include/sitetosite/Peer.h | 4 +-
libminifi/include/utils/Deleters.h | 36 +-
libminifi/include/utils/GeneralUtils.h | 23 +-
libminifi/include/utils/Monitors.h | 3 +-
libminifi/opsys/win/io/ClientSocket.h | 350 ------------
libminifi/src/core/PropertyValidation.cpp | 1 +
libminifi/src/io/CPPLINT.cfg | 3 +
libminifi/src/io/ClientSocket.cpp | 634 +++++++++++++++++++++
libminifi/src/io/ServerSocket.cpp | 16 +-
libminifi/src/io/StreamFactory.cpp | 29 +-
libminifi/src/io/posix/ClientSocket.cpp | 495 ----------------
libminifi/src/io/tls/TLSServerSocket.cpp | 5 +-
libminifi/src/io/tls/TLSSocket.cpp | 150 ++---
libminifi/src/io/win/ClientSocket.cpp | 556 ------------------
libminifi/test/RandomServerSocket.cpp | 4 +
libminifi/test/TestBase.h | 4 +-
nanofi/include/cxx/CallbackProcessor.h | 28 +-
nanofi/include/cxx/Instance.h | 14 +-
nanofi/tests/CLogAggregatorTests.cpp | 26 +-
45 files changed, 1154 insertions(+), 1965 deletions(-)
diff --git a/extensions/coap/controllerservice/CoapConnector.h b/extensions/coap/controllerservice/CoapConnector.h
index 1570b90..3c43000 100644
--- a/extensions/coap/controllerservice/CoapConnector.h
+++ b/extensions/coap/controllerservice/CoapConnector.h
@@ -28,6 +28,7 @@
#include "coap_message.h"
#include <memory>
#include <unordered_map>
+#include <utility>
namespace org {
namespace apache {
@@ -48,26 +49,17 @@ class CoapConnectorService : public core::controller::ControllerService {
* Constructors for the controller service.
*/
explicit CoapConnectorService(const std::string &name, const std::string &id)
- : ControllerService(name, id),
- port_(0),
- initialized_(false),
- logger_(logging::LoggerFactory<CoapConnectorService>::getLogger()) {
+ : ControllerService(name, id) {
initialize();
}
explicit CoapConnectorService(const std::string &name, utils::Identifier uuid = utils::Identifier())
- : ControllerService(name, uuid),
- port_(0),
- initialized_(false),
- logger_(logging::LoggerFactory<CoapConnectorService>::getLogger()) {
+ : ControllerService(name, std::move(uuid)) {
initialize();
}
explicit CoapConnectorService(const std::string &name, const std::shared_ptr<Configure> &configuration)
- : ControllerService(name),
- port_(0),
- initialized_(false),
- logger_(logging::LoggerFactory<CoapConnectorService>::getLogger()) {
+ : ControllerService(name) {
setConfiguration(configuration);
initialize();
}
@@ -79,21 +71,19 @@ class CoapConnectorService : public core::controller::ControllerService {
static core::Property Port;
static core::Property MaxQueueSize;
- virtual void initialize();
+ void initialize() override;
- void yield() {
+ void yield() override { }
- }
-
- bool isRunning() {
+ bool isRunning() override {
return getState() == core::controller::ControllerServiceState::ENABLED;
}
- bool isWorkAvailable() {
+ bool isWorkAvailable() override {
return false;
}
- virtual void onEnable();
+ void onEnable() override;
/**
* Sends the payload to the endpoint, returning the response as we await. Will retry transmission
@@ -103,7 +93,7 @@ class CoapConnectorService : public core::controller::ControllerService {
* @param size size of the payload to be sent
* @return CoAPMessage that contains the response code and data, if any.
*/
- CoapResponse sendPayload(uint8_t type, const std::string &endpoint, const CoapMessage * const message);
+ CoapResponse sendPayload(uint8_t type, const std::string &endpoint, const CoapMessage * message);
protected:
@@ -115,16 +105,16 @@ class CoapConnectorService : public core::controller::ControllerService {
// initialization mutex.
std::mutex initialization_mutex_;
- std::atomic<bool> initialized_;
+ std::atomic<bool> initialized_{ false };
private:
// host connecting to.
std::string host_;
// port connecting to
- unsigned int port_;
+ unsigned int port_{ 0 };
- std::shared_ptr<logging::Logger> logger_;
+ std::shared_ptr<logging::Logger> logger_{ logging::LoggerFactory<CoapConnectorService>::getLogger() };
};
diff --git a/extensions/coap/protocols/CoapC2Protocol.h b/extensions/coap/protocols/CoapC2Protocol.h
index 0fe9395..ff63f86 100644
--- a/extensions/coap/protocols/CoapC2Protocol.h
+++ b/extensions/coap/protocols/CoapC2Protocol.h
@@ -69,7 +69,7 @@ class CoapProtocol : public minifi::c2::RESTSender {
public:
explicit CoapProtocol(const std::string &name, const utils::Identifier &uuid = utils::Identifier());
- virtual ~CoapProtocol();
+ ~CoapProtocol() override;
/**
* Consume the payload.
@@ -77,17 +77,17 @@ class CoapProtocol : public minifi::c2::RESTSender {
* @param payload payload to consume.
* @param direction direction of operation.
*/
- virtual minifi::c2::C2Payload consumePayload(const std::string &url, const minifi::c2::C2Payload &payload, minifi::c2::Direction direction, bool async) override;
+ minifi::c2::C2Payload consumePayload(const std::string &url, const minifi::c2::C2Payload &payload, minifi::c2::Direction direction, bool async) override;
- virtual minifi::c2::C2Payload consumePayload(const minifi::c2::C2Payload &payload, minifi::c2::Direction direction, bool async) override {
+ minifi::c2::C2Payload consumePayload(const minifi::c2::C2Payload &payload, minifi::c2::Direction direction, bool async) override {
return serialize(payload);
}
- virtual void update(const std::shared_ptr<Configure> &configure) override {
+ void update(const std::shared_ptr<Configure> &configure) override {
// no op.
}
- virtual void initialize(const std::shared_ptr<core::controller::ControllerServiceProvider> &controller, const std::shared_ptr<Configure> &configure) override;
+ void initialize(const std::shared_ptr<core::controller::ControllerServiceProvider> &controller, const std::shared_ptr<Configure> &configure) override;
// Supported Properties
diff --git a/extensions/http-curl/client/HTTPStream.h b/extensions/http-curl/client/HTTPStream.h
index 3829e94..7e5a6c2 100644
--- a/extensions/http-curl/client/HTTPStream.h
+++ b/extensions/http-curl/client/HTTPStream.h
@@ -79,7 +79,7 @@ class HttpStream : public io::BaseStream {
*/
virtual void seek(uint64_t offset) override;
- virtual const uint64_t getSize() const override {
+ const uint64_t getSize() const override {
return written;
}
@@ -136,11 +136,9 @@ class HttpStream : public io::BaseStream {
}
inline bool isFinished(int seconds = 0) {
- if (http_client_future_.wait_for(std::chrono::seconds(seconds)) == std::future_status::ready && (http_read_callback_.getSize() == 0 && http_read_callback_.waitingOps())) {
- return true;
- } else {
- return false;
- }
+ return http_client_future_.wait_for(std::chrono::seconds(seconds)) == std::future_status::ready
+ && http_read_callback_.getSize() == 0
+ && http_read_callback_.waitingOps();
}
/**
diff --git a/extensions/http-curl/tests/CivetStream.h b/extensions/http-curl/tests/CivetStream.h
index 4ce54ec..2e58fd9 100644
--- a/extensions/http-curl/tests/CivetStream.h
+++ b/extensions/http-curl/tests/CivetStream.h
@@ -44,17 +44,14 @@ class CivetStream : public io::BaseStream {
}
- virtual ~CivetStream() {
- }
+ ~CivetStream() override = default;
/**
* Skip to the specified offset.
* @param offset offset to which we will skip
*/
- void seek(uint64_t offset){
-
- }
+ void seek(uint64_t offset) override { }
- const uint64_t getSize() const {
+ const uint64_t getSize() const override {
return BaseStream::readBuffer;
}
@@ -64,7 +61,7 @@ class CivetStream : public io::BaseStream {
* @param buf buffer in which we extract data
* @param buflen
*/
- virtual int readData(std::vector<uint8_t> &buf, int buflen) {
+ int readData(std::vector<uint8_t> &buf, int buflen) override {
if (buflen < 0) {
throw minifi::Exception{ExceptionType::GENERAL_EXCEPTION, "negative buflen"};
}
@@ -85,7 +82,7 @@ class CivetStream : public io::BaseStream {
* @param buf buffer in which we extract data
* @param buflen
*/
- virtual int readData(uint8_t *buf, int buflen) {
+ int readData(uint8_t *buf, int buflen) override {
return mg_read(conn,buf,buflen);
}
@@ -104,7 +101,7 @@ class CivetStream : public io::BaseStream {
* @param value value to write
* @param size size of value
*/
- virtual int writeData(uint8_t *value, int size) {
+ int writeData(uint8_t *value, int size) override {
return 0;
}
@@ -124,8 +121,6 @@ class CivetStream : public io::BaseStream {
return buf;
}
- void reset();
-
//size_t pos;
struct mg_connection *conn;
diff --git a/extensions/rocksdb-repos/RocksDbStream.cpp b/extensions/rocksdb-repos/RocksDbStream.cpp
index c55eedb..2a2cd95 100644
--- a/extensions/rocksdb-repos/RocksDbStream.cpp
+++ b/extensions/rocksdb-repos/RocksDbStream.cpp
@@ -18,6 +18,7 @@
#include "RocksDbStream.h"
#include <fstream>
+#include <utility>
#include <vector>
#include <memory>
#include <string>
@@ -29,19 +30,13 @@ namespace nifi {
namespace minifi {
namespace io {
-RocksDbStream::RocksDbStream(const std::string &path, rocksdb::DB *db, bool write_enable)
+RocksDbStream::RocksDbStream(std::string path, rocksdb::DB *db, bool write_enable)
: BaseStream(),
- path_(path),
+ path_(std::move(path)),
write_enable_(write_enable),
db_(db),
logger_(logging::LoggerFactory<RocksDbStream>::getLogger()) {
- rocksdb::Status status;
- status = db_->Get(rocksdb::ReadOptions(), path_, &value_);
- if (status.ok()) {
- exists_ = true;
- } else {
- exists_ = false;
- }
+ exists_ = db_->Get(rocksdb::ReadOptions(), path_, &value_).ok();
offset_ = 0;
size_ = value_.size();
}
diff --git a/extensions/rocksdb-repos/RocksDbStream.h b/extensions/rocksdb-repos/RocksDbStream.h
index 6b0df01..924ee0c 100644
--- a/extensions/rocksdb-repos/RocksDbStream.h
+++ b/extensions/rocksdb-repos/RocksDbStream.h
@@ -46,30 +46,24 @@ class RocksDbStream : public io::BaseStream {
* File Stream constructor that accepts an fstream shared pointer.
* It must already be initialized for read and write.
*/
- explicit RocksDbStream(const std::string &path, rocksdb::DB *db, bool write_enable = false);
+ explicit RocksDbStream(std::string path, rocksdb::DB *db, bool write_enable = false);
- /**
- * File Stream constructor that accepts an fstream shared pointer.
- * It must already be initialized for read and write.
- */
- explicit RocksDbStream(const std::string &path);
-
- virtual ~RocksDbStream() {
+ ~RocksDbStream() override {
closeStream();
}
- virtual void closeStream();
+ void closeStream() override;
/**
* Skip to the specified offset.
* @param offset offset to which we will skip
*/
- void seek(uint64_t offset);
+ void seek(uint64_t offset) override;
- const uint64_t getSize() const {
+ const uint64_t getSize() const override {
return size_;
}
- virtual int read(uint16_t &value, bool is_little_endian) {
+ int read(uint16_t &value, bool is_little_endian) override {
uint8_t buf[2];
if (readData(&buf[0], 2) < 0)
return -1;
@@ -81,7 +75,7 @@ class RocksDbStream : public io::BaseStream {
return 2;
}
- virtual int read(uint32_t &value, bool is_little_endian) {
+ int read(uint32_t &value, bool is_little_endian) override {
uint8_t buf[4];
if (readData(&buf[0], 4) < 0)
return -1;
@@ -94,7 +88,7 @@ class RocksDbStream : public io::BaseStream {
return 4;
}
- virtual int read(uint64_t &value, bool is_little_endian) {
+ int read(uint64_t &value, bool is_little_endian) override {
uint8_t buf[8];
if (readData(&buf[0], 8) < 0)
return -1;
@@ -114,13 +108,13 @@ class RocksDbStream : public io::BaseStream {
* @param buf buffer in which we extract data
* @param buflen
*/
- virtual int readData(std::vector<uint8_t> &buf, int buflen);
+ int readData(std::vector<uint8_t> &buf, int buflen) override;
/**
* Reads data and places it into buf
* @param buf buffer in which we extract data
* @param buflen
*/
- virtual int readData(uint8_t *buf, int buflen);
+ int readData(uint8_t *buf, int buflen) override;
/**
* Write value to the stream using std::vector
@@ -135,7 +129,7 @@ class RocksDbStream : public io::BaseStream {
* @param value value to write
* @param size size of value
*/
- virtual int writeData(uint8_t *value, int size);
+ int writeData(uint8_t *value, int size) override;
/**
* Returns the underlying buffer
diff --git a/extensions/standard-processors/processors/PutFile.cpp b/extensions/standard-processors/processors/PutFile.cpp
index e7ec1eb..0d89355 100644
--- a/extensions/standard-processors/processors/PutFile.cpp
+++ b/extensions/standard-processors/processors/PutFile.cpp
@@ -244,8 +244,7 @@ bool PutFile::putFile(core::ProcessSession *session, std::shared_ptr<FlowFileRec
PutFile::ReadCallback::ReadCallback(const std::string &tmp_file, const std::string &dest_file)
: tmp_file_(tmp_file),
- dest_file_(dest_file),
- logger_(logging::LoggerFactory<PutFile::ReadCallback>::getLogger()) {
+ dest_file_(dest_file) {
}
// Copy the entire file contents to the temporary file
diff --git a/extensions/standard-processors/processors/PutFile.h b/extensions/standard-processors/processors/PutFile.h
index cfc8924..fdfcace 100644
--- a/extensions/standard-processors/processors/PutFile.h
+++ b/extensions/standard-processors/processors/PutFile.h
@@ -20,6 +20,8 @@
#ifndef __PUT_FILE_H__
#define __PUT_FILE_H__
+#include <utility>
+
#include "FlowFileRecord.h"
#include "core/Processor.h"
#include "core/ProcessSession.h"
@@ -47,11 +49,11 @@ class PutFile : public core::Processor {
* Create a new processor
*/
PutFile(std::string name, utils::Identifier uuid = utils::Identifier())
- : core::Processor(name, uuid),
+ : core::Processor(std::move(name), uuid),
logger_(logging::LoggerFactory<PutFile>::getLogger()) {
}
- virtual ~PutFile() = default;
+ ~PutFile() override = default;
// Supported Properties
static core::Property Directory;
@@ -68,25 +70,23 @@ class PutFile : public core::Processor {
* @param sessionFactory process session factory that is used when creating
* ProcessSession objects.
*/
- virtual void onSchedule(core::ProcessContext *context, core::ProcessSessionFactory *sessionFactory);
+ void onSchedule(core::ProcessContext *context, core::ProcessSessionFactory *sessionFactory) override;
- virtual void onTrigger(core::ProcessContext *context, core::ProcessSession *session);
- virtual void initialize(void);
+ void onTrigger(core::ProcessContext *context, core::ProcessSession *session) override;
+ void initialize() override;
class ReadCallback : public InputStreamCallback {
public:
- ReadCallback(const std::string &tmp_file,
- const std::string &dest_file);
- ~ReadCallback();
- virtual int64_t process(std::shared_ptr<io::BaseStream> stream);
+ ReadCallback(const std::string &tmp_file, const std::string &dest_file);
+ ~ReadCallback() override;
+ int64_t process(std::shared_ptr<io::BaseStream> stream) override;
bool commit();
private:
- std::shared_ptr<logging::Logger> logger_;
+ std::shared_ptr<logging::Logger> logger_{ logging::LoggerFactory<PutFile::ReadCallback>::getLogger() };
bool write_succeeded_ = false;
std::string tmp_file_;
std::string dest_file_;
- std::string dest_dir_;
};
/**
diff --git a/libminifi/include/core/ProcessSession.h b/libminifi/include/core/ProcessSession.h
index 969a36c..7ae75ef 100644
--- a/libminifi/include/core/ProcessSession.h
+++ b/libminifi/include/core/ProcessSession.h
@@ -18,6 +18,7 @@
#ifndef __PROCESS_SESSION_H__
#define __PROCESS_SESSION_H__
+#include <utility>
#include <vector>
#include <queue>
#include <map>
@@ -49,10 +50,10 @@ class ProcessSession : public ReferenceContainer {
* Create a new process session
*/
ProcessSession(std::shared_ptr<ProcessContext> processContext = nullptr)
- : process_context_(processContext),
+ : process_context_(std::move(processContext)),
logger_(logging::LoggerFactory<ProcessSession>::getLogger()) {
logger_->log_trace("ProcessSession created for %s", process_context_->getProcessorNode()->getName());
- auto repo = processContext->getProvenanceRepository();
+ auto repo = process_context_->getProvenanceRepository();
//provenance_report_ = new provenance::ProvenanceReporter(repo, process_context_->getProcessorNode()->getName(), process_context_->getProcessorNode()->getName());
provenance_report_ = std::make_shared<provenance::ProvenanceReporter>(repo, process_context_->getProcessorNode()->getName(), process_context_->getProcessorNode()->getName());
}
@@ -83,8 +84,6 @@ class ProcessSession : public ReferenceContainer {
std::shared_ptr<core::FlowFile> clone(const std::shared_ptr<core::FlowFile> &parent);
// Clone a new UUID FlowFile from parent for attributes and sub set of parent content resource claim
std::shared_ptr<core::FlowFile> clone(const std::shared_ptr<core::FlowFile> &parent, int64_t offset, int64_t size);
- // Duplicate a FlowFile with the same UUID and all attributes and content resource claim for the roll back of the session
- std::shared_ptr<core::FlowFile> duplicate(const std::shared_ptr<core::FlowFile> &original);
// Transfer the FlowFile to the relationship
virtual void transfer(const std::shared_ptr<core::FlowFile> &flow, Relationship relationship);
// Put Attribute
diff --git a/libminifi/include/core/PropertyValidation.h b/libminifi/include/core/PropertyValidation.h
index f221b48..3ed03ac 100644
--- a/libminifi/include/core/PropertyValidation.h
+++ b/libminifi/include/core/PropertyValidation.h
@@ -24,6 +24,7 @@
#include "utils/StringUtils.h"
#include <limits>
#include <memory>
+#include <utility>
namespace org {
namespace apache {
@@ -62,7 +63,7 @@ class ValidationResult {
}
protected:
- bool valid_;
+ bool valid_{ false };
std::string subject_;
std::string input_;
friend class ValidationResult;
@@ -85,12 +86,10 @@ class ValidationResult {
class PropertyValidator {
public:
- PropertyValidator(const std::string &name)
- : name_(name) {
- }
- virtual ~PropertyValidator() {
-
+ PropertyValidator(std::string name)
+ : name_(std::move(name)) {
}
+ virtual ~PropertyValidator() = default;
std::string getName() const {
return name_;
@@ -120,17 +119,17 @@ class AlwaysValid : public PropertyValidator {
bool always_valid_;
public:
AlwaysValid(bool isalwaysvalid, const std::string &name)
- : always_valid_(isalwaysvalid),
- PropertyValidator(name) {
-
- }
- virtual ~AlwaysValid() {
+ : PropertyValidator(name),
+ always_valid_(isalwaysvalid) {
}
- ValidationResult validate(const std::string &subject, const std::shared_ptr<minifi::state::response::Value> &input) const {
+
+ ~AlwaysValid() override = default;
+
+ ValidationResult validate(const std::string &subject, const std::shared_ptr<minifi::state::response::Value> &input) const override {
return ValidationResult::Builder::createBuilder().withSubject(subject).withInput(input->getStringValue()).isValid(always_valid_).build();
}
- ValidationResult validate(const std::string &subject, const std::string &input) const {
+ ValidationResult validate(const std::string &subject, const std::string &input) const override {
return ValidationResult::Builder::createBuilder().withSubject(subject).withInput(input).isValid(always_valid_).build();
}
@@ -141,15 +140,14 @@ class BooleanValidator : public PropertyValidator {
BooleanValidator(const std::string &name)
: PropertyValidator(name) {
}
- virtual ~BooleanValidator() {
- }
+ ~BooleanValidator() override = default;
- ValidationResult validate(const std::string &subject, const std::shared_ptr<minifi::state::response::Value> &input) const {
+ ValidationResult validate(const std::string &subject, const std::shared_ptr<minifi::state::response::Value> &input) const override {
return PropertyValidator::_validate_internal<minifi::state::response::BoolValue>(subject, input);
}
- ValidationResult validate(const std::string &subject, const std::string &input) const {
+ ValidationResult validate(const std::string &subject, const std::string &input) const override {
if (utils::StringUtils::equalsIgnoreCase(input, "true") || utils::StringUtils::equalsIgnoreCase(input, "false"))
return ValidationResult::Builder::createBuilder().withSubject(subject).withInput(input).isValid(true).build();
else
@@ -162,19 +160,17 @@ class IntegerValidator : public PropertyValidator {
IntegerValidator(const std::string &name)
: PropertyValidator(name) {
}
- virtual ~IntegerValidator() {
- }
+ ~IntegerValidator() override = default;
- ValidationResult validate(const std::string &subject, const std::shared_ptr<minifi::state::response::Value> &input) const {
+ ValidationResult validate(const std::string &subject, const std::shared_ptr<minifi::state::response::Value> &input) const override {
return PropertyValidator::_validate_internal<minifi::state::response::IntValue>(subject, input);
}
- ValidationResult validate(const std::string &subject, const std::string &input) const {
+ ValidationResult validate(const std::string &subject, const std::string &input) const override {
try {
std::stoi(input);
return ValidationResult::Builder::createBuilder().withSubject(subject).withInput(input).isValid(true).build();
} catch (...) {
-
}
return ValidationResult::Builder::createBuilder().withSubject(subject).withInput(input).isValid(false).build();
}
@@ -185,14 +181,13 @@ class UnsignedIntValidator : public PropertyValidator {
explicit UnsignedIntValidator(const std::string &name)
: PropertyValidator(name) {
}
- virtual ~UnsignedIntValidator() {
+ ~UnsignedIntValidator() override = default;
- }
- ValidationResult validate(const std::string &subject, const std::shared_ptr<minifi::state::response::Value> &input) const {
+ ValidationResult validate(const std::string &subject, const std::shared_ptr<minifi::state::response::Value> &input) const override {
return PropertyValidator::_validate_internal<minifi::state::response::UInt32Value>(subject, input);
}
- ValidationResult validate(const std::string &subject, const std::string &input) const {
+ ValidationResult validate(const std::string &subject, const std::string &input) const override {
try {
auto negative = input.find_first_of('-') != std::string::npos;
if (negative){
@@ -215,10 +210,9 @@ class LongValidator : public PropertyValidator {
min_(min),
max_(max) {
}
- virtual ~LongValidator() {
+ ~LongValidator() override = default;
- }
- ValidationResult validate(const std::string &subject, const std::shared_ptr<minifi::state::response::Value> &input) const {
+ ValidationResult validate(const std::string &subject, const std::shared_ptr<minifi::state::response::Value> &input) const override {
auto in64 = std::dynamic_pointer_cast<minifi::state::response::Int64Value>(input);
if (in64) {
return ValidationResult::Builder::createBuilder().withSubject(subject).withInput(in64->getStringValue()).isValid(in64->getValue() >= min_ && in64->getValue() <= max_).build();
@@ -228,7 +222,7 @@ class LongValidator : public PropertyValidator {
}
}
- ValidationResult validate(const std::string &subject, const std::string &input) const {
+ ValidationResult validate(const std::string &subject, const std::string &input) const override {
try {
auto res = std::stoll(input);
@@ -249,14 +243,13 @@ class UnsignedLongValidator : public PropertyValidator {
explicit UnsignedLongValidator(const std::string &name)
: PropertyValidator(name) {
}
- virtual ~UnsignedLongValidator() {
+ ~UnsignedLongValidator() override = default;
- }
- ValidationResult validate(const std::string &subject, const std::shared_ptr<minifi::state::response::Value> &input) const {
+ ValidationResult validate(const std::string &subject, const std::shared_ptr<minifi::state::response::Value> &input) const override {
return PropertyValidator::_validate_internal<minifi::state::response::UInt64Value>(subject, input);
}
- ValidationResult validate(const std::string &subject, const std::string &input) const {
+ ValidationResult validate(const std::string &subject, const std::string &input) const override {
try {
auto negative = input.find_first_of('-') != std::string::npos;
if (negative){
@@ -265,11 +258,9 @@ class UnsignedLongValidator : public PropertyValidator {
std::stoull(input);
return ValidationResult::Builder::createBuilder().withSubject(subject).withInput(input).isValid(true).build();
} catch (...) {
-
}
return ValidationResult::Builder::createBuilder().withSubject(subject).withInput(input).isValid(false).build();
}
-
};
class DataSizeValidator : public PropertyValidator {
@@ -277,19 +268,15 @@ class DataSizeValidator : public PropertyValidator {
DataSizeValidator(const std::string &name)
: PropertyValidator(name) {
}
- virtual ~DataSizeValidator() {
+ ~DataSizeValidator() override = default;
- }
- ValidationResult validate(const std::string &subject, const std::shared_ptr<minifi::state::response::Value> &input) const {
+ ValidationResult validate(const std::string &subject, const std::shared_ptr<minifi::state::response::Value> &input) const override {
return PropertyValidator::_validate_internal<core::DataSizeValue>(subject, input);
}
- ValidationResult validate(const std::string &subject, const std::string &input) const {
+ ValidationResult validate(const std::string &subject, const std::string &input) const override {
uint64_t out;
- if (core::DataSizeValue::StringToInt(input, out))
- return ValidationResult::Builder::createBuilder().withSubject(subject).withInput(input).isValid(true).build();
- else
- return ValidationResult::Builder::createBuilder().withSubject(subject).withInput(input).isValid(false).build();
+ return ValidationResult::Builder::createBuilder().withSubject(subject).withInput(input).isValid(core::DataSizeValue::StringToInt(input, out)).build();
}
};
@@ -298,9 +285,7 @@ class PortValidator : public LongValidator {
PortValidator(const std::string &name)
: LongValidator(name, 1, 65535) {
}
- virtual ~PortValidator() {
-
- }
+ ~PortValidator() override = default;
};
//Use only for specifying listen ports, where 0 means a randomly chosen one!
@@ -309,9 +294,7 @@ public:
ListenPortValidator(const std::string &name)
: LongValidator(name, 0, 65535) {
}
- virtual ~ListenPortValidator() {
-
- }
+ ~ListenPortValidator() override = default;
};
class TimePeriodValidator : public PropertyValidator {
@@ -319,20 +302,16 @@ class TimePeriodValidator : public PropertyValidator {
TimePeriodValidator(const std::string &name)
: PropertyValidator(name) {
}
- virtual ~TimePeriodValidator() {
+ ~TimePeriodValidator() override = default;
- }
- ValidationResult validate(const std::string &subject, const std::shared_ptr<minifi::state::response::Value> &input) const {
+ ValidationResult validate(const std::string &subject, const std::shared_ptr<minifi::state::response::Value> &input) const override {
return PropertyValidator::_validate_internal<core::TimePeriodValue>(subject, input);
}
- ValidationResult validate(const std::string &subject, const std::string &input) const {
+ ValidationResult validate(const std::string &subject, const std::string &input) const override {
uint64_t out;
TimeUnit outTimeUnit;
- if (core::TimePeriodValue::StringToTime(input, out, outTimeUnit))
- return ValidationResult::Builder::createBuilder().withSubject(subject).withInput(input).isValid(true).build();
- else
- return ValidationResult::Builder::createBuilder().withSubject(subject).withInput(input).isValid(false).build();
+ return ValidationResult::Builder::createBuilder().withSubject(subject).withInput(input).isValid(core::TimePeriodValue::StringToTime(input, out, outTimeUnit)).build();
}
};
@@ -340,7 +319,6 @@ class TimePeriodValidator : public PropertyValidator {
class StandardValidators {
public:
-
static std::shared_ptr<PropertyValidator> VALID;
static const std::shared_ptr<PropertyValidator> &getValidator(const std::shared_ptr<minifi::state::response::Value> &input) {
@@ -362,16 +340,16 @@ class StandardValidators {
} else if (std::dynamic_pointer_cast<minifi::state::response::UInt64Value>(input) != nullptr) {
return init.UNSIGNED_LONG_VALIDATOR;
} else {
- return init.VALID;
+ return org::apache::nifi::minifi::core::StandardValidators::VALID;
}
}
- static const std::shared_ptr<PropertyValidator> PORT_VALIDATOR(){
+ static std::shared_ptr<PropertyValidator> PORT_VALIDATOR(){
static std::shared_ptr<PropertyValidator> validator = std::make_shared<PortValidator>("PORT_VALIDATOR");
return validator;
}
- static const std::shared_ptr<PropertyValidator> LISTEN_PORT_VALIDATOR(){
+ static std::shared_ptr<PropertyValidator> LISTEN_PORT_VALIDATOR(){
static std::shared_ptr<PropertyValidator> validator = std::make_shared<ListenPortValidator>("PORT_VALIDATOR");
return validator;
}
@@ -387,8 +365,7 @@ class StandardValidators {
std::shared_ptr<PropertyValidator> TIME_PERIOD_VALIDATOR;
StandardValidators();
-}
-;
+};
} /* namespace core */
} /* namespace minifi */
diff --git a/libminifi/include/core/PropertyValue.h b/libminifi/include/core/PropertyValue.h
index be62226..bbdaf98 100644
--- a/libminifi/include/core/PropertyValue.h
+++ b/libminifi/include/core/PropertyValue.h
@@ -59,21 +59,12 @@ static inline std::shared_ptr<state::response::Value> convert(const std::shared_
class PropertyValue : public state::response::ValueNode {
public:
PropertyValue()
- : type_id(std::type_index(typeid(std::string))),
- ValueNode() {
+ : type_id(std::type_index(typeid(std::string))) {
validator_ = StandardValidators::VALID;
}
- PropertyValue(const PropertyValue &o)
- : type_id(o.type_id),
- validator_(o.validator_),
- state::response::ValueNode(o) {
- }
- PropertyValue(PropertyValue &&o) noexcept
- : type_id(o.type_id),
- validator_(std::move(o.validator_)),
- state::response::ValueNode(std::move(o)) {
- }
+ PropertyValue(const PropertyValue &o) = default;
+ PropertyValue(PropertyValue &&o) noexcept = default;
void setValidator(const std::shared_ptr<PropertyValidator> &val) {
validator_ = val;
@@ -84,12 +75,11 @@ class PropertyValue : public state::response::ValueNode {
}
ValidationResult validate(const std::string &subject) const {
- if (validator_) {
- return validator_->validate(subject, getValue());
- }
- else {
- return ValidationResult::Builder::createBuilder().isValid(true).build();
- }
+ if (validator_) {
+ return validator_->validate(subject, getValue());
+ } else {
+ return ValidationResult::Builder::createBuilder().isValid(true).build();
+ }
}
operator uint64_t() const {
@@ -147,7 +137,7 @@ class PropertyValue : public state::response::ValueNode {
* createValue
*/
template<typename T>
- auto operator=(const T ref) -> typename std::enable_if<std::is_same<T, std::string >::value,PropertyValue&>::type {
+ auto operator=(const T ref) -> typename std::enable_if<std::is_same<T, std::string>::value,PropertyValue&>::type {
if (value_ == nullptr) {
type_id = std::type_index(typeid(T));
value_ = minifi::state::response::createValue(ref);
diff --git a/libminifi/include/core/StreamManager.h b/libminifi/include/core/StreamManager.h
index a2a2e78..cdbda53 100644
--- a/libminifi/include/core/StreamManager.h
+++ b/libminifi/include/core/StreamManager.h
@@ -37,9 +37,7 @@ namespace core {
template<typename T>
class StreamManager {
public:
- virtual ~StreamManager() {
-
- }
+ virtual ~StreamManager() = default;
virtual std::string getStoragePath() const = 0;
diff --git a/libminifi/include/io/AtomicEntryStream.h b/libminifi/include/io/AtomicEntryStream.h
index 9f9043e..145991c 100644
--- a/libminifi/include/io/AtomicEntryStream.h
+++ b/libminifi/include/io/AtomicEntryStream.h
@@ -50,19 +50,18 @@ class AtomicEntryStream : public BaseStream {
}
}
- virtual ~AtomicEntryStream();
-
- virtual void closeStream() {
+ ~AtomicEntryStream() override;
+ void closeStream() override {
}
/**
* Skip to the specified offset.
* @param offset offset to which we will skip
*/
- void seek(uint64_t offset);
+ void seek(uint64_t offset) override;
- virtual const uint64_t getSize() const {
+ const uint64_t getSize() const override {
return length_;
}
@@ -72,19 +71,18 @@ class AtomicEntryStream : public BaseStream {
* @param buf buffer in which we extract data
* @param buflen
*/
- virtual int readData(std::vector<uint8_t> &buf, int buflen);
+ int readData(std::vector<uint8_t> &buf, int buflen) override;
/**
* Reads data and places it into buf
* @param buf buffer in which we extract data
* @param buflen
*/
- virtual int readData(uint8_t *buf, int buflen);
+ int readData(uint8_t *buf, int buflen) override;
/**
* Write value to the stream using std::vector
* @param buf incoming buffer
* @param buflen buffer to write
- *
*/
virtual int writeData(std::vector<uint8_t> &buf, int buflen);
@@ -93,7 +91,7 @@ class AtomicEntryStream : public BaseStream {
* @param value value to write
* @param size size of value
*/
- virtual int writeData(uint8_t *value, int size);
+ int writeData(uint8_t *value, int size) override;
/**
* Returns the underlying buffer
diff --git a/libminifi/include/io/BaseStream.h b/libminifi/include/io/BaseStream.h
index 8311a30..ab4a8a1 100644
--- a/libminifi/include/io/BaseStream.h
+++ b/libminifi/include/io/BaseStream.h
@@ -50,9 +50,8 @@ class BaseStream : public DataStream, public Serializable {
: composable_stream_(other) {
}
- virtual ~BaseStream() {
+ ~BaseStream() override = default;
- }
/**
* write 4 bytes to stream
* @param base_value non encoded value
@@ -62,9 +61,9 @@ class BaseStream : public DataStream, public Serializable {
**/
virtual int write(uint32_t base_value, bool is_little_endian = EndiannessCheck::IS_LITTLE);
- virtual int writeData(uint8_t *value, int size);
+ int writeData(uint8_t *value, int size) override;
- virtual void seek(uint64_t offset) {
+ void seek(uint64_t offset) override {
if (LIKELY(composable_stream_ != this)) {
composable_stream_->seek(offset);
} else {
@@ -126,13 +125,13 @@ class BaseStream : public DataStream, public Serializable {
* @param buf buffer in which we extract data
* @param buflen
*/
- virtual int readData(std::vector<uint8_t> &buf, int buflen);
+ int readData(std::vector<uint8_t> &buf, int buflen) override;
/**
* Reads data and places it into buf
* @param buf buffer in which we extract data
* @param buflen
*/
- virtual int readData(uint8_t *buf, int buflen);
+ int readData(uint8_t *buf, int buflen) override;
/**
* reads two bytes from the stream
@@ -140,7 +139,7 @@ class BaseStream : public DataStream, public Serializable {
* @param stream stream from which we will read
* @return resulting read size
**/
- virtual int read(uint16_t &base_value, bool is_little_endian = EndiannessCheck::IS_LITTLE);
+ int read(uint16_t &base_value, bool is_little_endian = EndiannessCheck::IS_LITTLE) override;
/**
* reads a byte from the stream
@@ -165,7 +164,7 @@ class BaseStream : public DataStream, public Serializable {
* @param stream stream from which we will read
* @return resulting read size
**/
- virtual int read(uint32_t &value, bool is_little_endian = EndiannessCheck::IS_LITTLE);
+ int read(uint32_t &value, bool is_little_endian = EndiannessCheck::IS_LITTLE) override;
/**
* reads eight byte from the stream
@@ -173,15 +172,14 @@ class BaseStream : public DataStream, public Serializable {
* @param stream stream from which we will read
* @return resulting read size
**/
- virtual int read(uint64_t &value, bool is_little_endian = EndiannessCheck::IS_LITTLE);
+ int read(uint64_t &value, bool is_little_endian = EndiannessCheck::IS_LITTLE) override;
- virtual const uint64_t getSize() const {
+ const uint64_t getSize() const override {
if (LIKELY(composable_stream_ == this)) {
return buffer.size();
} else {
return composable_stream_->getSize();
}
-
}
/**
diff --git a/libminifi/include/io/CRCStream.h b/libminifi/include/io/CRCStream.h
index cd789e1..6b63f95 100644
--- a/libminifi/include/io/CRCStream.h
+++ b/libminifi/include/io/CRCStream.h
@@ -44,14 +44,13 @@ class CRCStream : public BaseStream {
* Raw pointer because the caller guarantees that
* it will exceed our lifetime.
*/
- explicit CRCStream(T *stream);
+ explicit CRCStream(T *child_stream);
- explicit CRCStream(CRCStream<T> &&move);
+ CRCStream(CRCStream<T>&&) noexcept;
- virtual ~CRCStream() {
- }
+ ~CRCStream() override = default;
- T *getstream() {
+ T *getstream() const {
return child_stream_;
}
@@ -64,19 +63,18 @@ class CRCStream : public BaseStream {
* @param buf buffer in which we extract data
* @param buflen
*/
- virtual int readData(std::vector<uint8_t> &buf, int buflen);
+ int readData(std::vector<uint8_t> &buf, int buflen) override;
/**
* Reads data and places it into buf
* @param buf buffer in which we extract data
* @param buflen
*/
- virtual int readData(uint8_t *buf, int buflen);
+ int readData(uint8_t *buf, int buflen) override;
/**
* Write value to the stream using std::vector
* @param buf incoming buffer
* @param buflen buffer to write
- *
*/
virtual int writeData(std::vector<uint8_t> &buf, int buflen);
@@ -85,7 +83,7 @@ class CRCStream : public BaseStream {
* @param value value to write
* @param size size of value
*/
- virtual int writeData(uint8_t *value, int size);
+ int writeData(uint8_t *value, int size) override;
/**
* write 4 bytes to stream
@@ -94,7 +92,7 @@ class CRCStream : public BaseStream {
* @param is_little_endian endianness determination
* @return resulting write size
**/
- virtual int write(uint32_t base_value, bool is_little_endian = EndiannessCheck::IS_LITTLE);
+ int write(uint32_t base_value, bool is_little_endian = EndiannessCheck::IS_LITTLE) override;
/**
* write 2 bytes to stream
* @param base_value non encoded value
@@ -102,7 +100,7 @@ class CRCStream : public BaseStream {
* @param is_little_endian endianness determination
* @return resulting write size
**/
- virtual int write(uint16_t base_value, bool is_little_endian = EndiannessCheck::IS_LITTLE);
+ int write(uint16_t base_value, bool is_little_endian = EndiannessCheck::IS_LITTLE) override;
/**
* write 8 bytes to stream
@@ -111,27 +109,27 @@ class CRCStream : public BaseStream {
* @param is_little_endian endianness determination
* @return resulting write size
**/
- virtual int write(uint64_t base_value, bool is_little_endian = EndiannessCheck::IS_LITTLE);
+ int write(uint64_t base_value, bool is_little_endian = EndiannessCheck::IS_LITTLE) override;
/**
* Reads a system word
* @param value value to write
*/
- virtual int read(uint64_t &value, bool is_little_endian = EndiannessCheck::IS_LITTLE);
+ int read(uint64_t &value, bool is_little_endian = EndiannessCheck::IS_LITTLE) override;
/**
* Reads a uint32_t
* @param value value to write
*/
- virtual int read(uint32_t &value, bool is_little_endian = EndiannessCheck::IS_LITTLE);
+ int read(uint32_t &value, bool is_little_endian = EndiannessCheck::IS_LITTLE) override;
/**
* Reads a system short
* @param value value to write
*/
- virtual int read(uint16_t &value, bool is_little_endian = EndiannessCheck::IS_LITTLE);
+ int read(uint16_t &value, bool is_little_endian = EndiannessCheck::IS_LITTLE) override;
- virtual short initialize() {
+ short initialize() override {
child_stream_->initialize();
reset();
return 0;
@@ -166,14 +164,14 @@ class CRCStream : public BaseStream {
};
template<typename T>
-CRCStream<T>::CRCStream(T *other)
- : child_stream_(other),
+CRCStream<T>::CRCStream(T *child_stream)
+ : child_stream_(child_stream),
disable_encoding_(false) {
crc_ = crc32(0L, Z_NULL, 0);
}
template<typename T>
-CRCStream<T>::CRCStream(CRCStream<T> &&move)
+CRCStream<T>::CRCStream(CRCStream<T> &&move) noexcept
: crc_(std::move(move.crc_)),
child_stream_(std::move(move.child_stream_)),
disable_encoding_(false) {
@@ -229,7 +227,6 @@ void CRCStream<T>::updateCRC(uint8_t *buffer, uint32_t length) {
template<typename T>
int CRCStream<T>::write(uint64_t base_value, bool is_little_endian) {
-
if (disable_encoding_)
is_little_endian = false;
const uint64_t value = is_little_endian == 1 ? htonll_r(base_value) : base_value;
@@ -284,7 +281,6 @@ int CRCStream<T>::read(uint32_t &value, bool is_little_endian) {
value = (buf[0] << 24) | (buf[1] << 16) | (buf[2] << 8) | buf[3];
} else {
value = buf[0] | buf[1] << 8 | buf[2] << 16 | buf[3] << 24;
-
}
return sizeof(value);
@@ -300,7 +296,6 @@ int CRCStream<T>::read(uint16_t &value, bool is_little_endian) {
value = (buf[0] << 8) | buf[1];
} else {
value = buf[0] | buf[1] << 8;
-
}
return sizeof(value);
}
diff --git a/libminifi/opsys/posix/io/ClientSocket.h b/libminifi/include/io/ClientSocket.h
similarity index 70%
rename from libminifi/opsys/posix/io/ClientSocket.h
rename to libminifi/include/io/ClientSocket.h
index cba975d..b1ca262 100644
--- a/libminifi/opsys/posix/io/ClientSocket.h
+++ b/libminifi/include/io/ClientSocket.h
@@ -15,17 +15,26 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#ifndef LIBMINIFI_INCLUDE_IO_POSIX_CLIENTSOCKET_H_
-#define LIBMINIFI_INCLUDE_IO_POSIX_CLIENTSOCKET_H_
-
+#ifndef LIBMINIFI_INCLUDE_IO_CLIENTSOCKET_H_
+#define LIBMINIFI_INCLUDE_IO_CLIENTSOCKET_H_
#include <cstdint>
+#ifdef WIN32
+#ifndef WIN32_LEAN_AND_MEAN
+#define WIN32_LEAN_AND_MEAN
+#endif /* WIN32_LEAN_AND_MEAN */
+#include <WinSock2.h>
+#else
#include <sys/types.h>
#include <sys/socket.h>
#include <netdb.h>
#include <unistd.h>
+#endif /* WIN32 */
#include <mutex>
#include <atomic>
+#include <string>
+#include <memory>
+#include <vector>
#include "io/BaseStream.h"
#include "core/Core.h"
#include "core/logging/Logger.h"
@@ -39,6 +48,26 @@ namespace nifi {
namespace minifi {
namespace io {
+#ifdef WIN32
+using SocketDescriptor = SOCKET;
+using ip4addr = in_addr;
+#else
+using SocketDescriptor = int;
+using ip4addr = in_addr_t;
+static constexpr SocketDescriptor INVALID_SOCKET = -1;
+static constexpr int SOCKET_ERROR = -1;
+#endif /* WIN32 */
+
+/**
+ * Return the last socket error message, based on errno on posix and WSAGetLastError() on windows
+ */
+std::string get_last_socket_error_message();
+
+/**
+ * @return >= 0 on posix, != INVALID_SOCKET on windows
+ */
+bool valid_socket(SocketDescriptor) noexcept;
+
/**
* Context class for socket. This is currently only used as a parent class for TLSContext. It is necessary so the Socket and TLSSocket constructors
* can be the same. It also gives us a common place to set timeouts, etc from the Configure object in the future.
@@ -54,8 +83,6 @@ class SocketContext {
* connecting information from users
* Design: Extends DataStream and allows us to perform most streaming
* operations against a BSD socket
- *
- *
*/
class Socket : public BaseStream {
public:
@@ -65,18 +92,19 @@ class Socket : public BaseStream {
* @param hostname hostname we are connecting to.
* @param port port we are connecting to.
*/
- explicit Socket(const std::shared_ptr<SocketContext> &context, const std::string &hostname, const uint16_t port);
+ explicit Socket(const std::shared_ptr<SocketContext> &context, std::string hostname, uint16_t port);
- /**
- * Move constructor.
- */
- explicit Socket(const Socket &&);
+ Socket(const Socket&) = delete;
+ Socket(Socket&&) noexcept;
+
+ Socket& operator=(const Socket&) = delete;
+ Socket& operator=(Socket&& other) noexcept;
/**
* Static function to return the current machine's host name
*/
static std::string getMyHostName() {
- static std::string HOSTNAME = init_hostname();
+ static const std::string HOSTNAME = init_hostname();
return HOSTNAME;
}
@@ -93,7 +121,7 @@ class Socket : public BaseStream {
*/
virtual int16_t initialize();
- virtual void setInterface(io::NetworkInterface &&interface) {
+ virtual void setInterface(io::NetworkInterface interface) {
local_network_interface_ = std::move(interface);
}
@@ -112,6 +140,10 @@ class Socket : public BaseStream {
return port_;
}
+ void setPort(uint16_t port) {
+ port_ = port;
+ }
+
// data stream extensions
/**
* Reads data and places it into buf
@@ -198,22 +230,6 @@ class Socket : public BaseStream {
*/
virtual int read(uint16_t &value, bool is_little_endian = EndiannessCheck::IS_LITTLE);
- /**
- * Returns the underlying buffer
- * @return vector's array
- **/
- const uint8_t *getBuffer() const {
- return DataStream::getBuffer();
- }
-
- /**
- * Retrieve size of data stream
- * @return size of data stream
- **/
- const uint64_t getSize() const {
- return DataStream::getSize();
- }
-
protected:
/**
@@ -224,7 +240,7 @@ class Socket : public BaseStream {
* @param port connecting port
* @param listeners number of listeners in the queue
*/
- explicit Socket(const std::shared_ptr<SocketContext> &context, const std::string &hostname, const uint16_t port, const uint16_t listeners);
+ explicit Socket(const std::shared_ptr<SocketContext> &context, std::string hostname, uint16_t port, uint16_t listeners);
/**
* Creates a vector and returns the vector using the provided
@@ -236,62 +252,86 @@ class Socket : public BaseStream {
std::vector<uint8_t> readBuffer(const T&);
/**
- * Creates a connection using the address info object.
- * @param p addrinfo structure.
- * @returns fd.
+ * Creates a connection using the addr object
+ * @param ignored ignored
+ * @param addr The IPv4 address to connect to
+ * @returns 0 on success, -1 on error
+ */
+ virtual int8_t createConnection(const addrinfo *ignored, ip4addr &addr);
+
+ /**
+ * Iterates through {@code destination_addresses} and tries to connect to each address until it succeeds.
+ * Supports both IPv4 and IPv6.
+ * @param destination_addresses Destination addresses, typically from {@code getaddrinfo}.
+ * @return 0 on success, -1 on error
*/
- virtual int8_t createConnection(const addrinfo *p, in_addr_t &addr);
+ virtual int8_t createConnection(const addrinfo *destination_addresses);
/**
* Sets socket options depending on the instance.
* @param sock socket file descriptor.
*/
- virtual int16_t setSocketOptions(const int sock);
+ virtual int16_t setSocketOptions(SocketDescriptor sock);
/**
* Attempt to select the socket file descriptor
* @param msec timeout interval to wait
* @returns file descriptor
*/
- virtual int16_t select_descriptor(const uint16_t msec);
-
- addrinfo *addr_info_;
+ virtual int16_t select_descriptor(uint16_t msec);
std::recursive_mutex selection_mutex_;
std::string requested_hostname_;
std::string canonical_hostname_;
- uint16_t port_;
+ uint16_t port_{ 0 };
- bool is_loopback_only_;
+ bool is_loopback_only_{ false };
io::NetworkInterface local_network_interface_;
// connection information
- int32_t socket_file_descriptor_;
-
- fd_set total_list_;
- fd_set read_fds_;
- std::atomic<uint16_t> socket_max_;
- std::atomic<uint64_t> total_written_;
- std::atomic<uint64_t> total_read_;
- uint16_t listeners_;
+ SocketDescriptor socket_file_descriptor_{ INVALID_SOCKET }; // -1 on posix
+ fd_set total_list_{};
+ fd_set read_fds_{};
+ std::atomic<uint16_t> socket_max_{ 0 };
+ std::atomic<uint64_t> total_written_{ 0 };
+ std::atomic<uint64_t> total_read_{ 0 };
+ uint16_t listeners_{ 0 };
- bool nonBlocking_;
+ bool nonBlocking_{ false };
std::shared_ptr<logging::Logger> logger_;
- void setPort(uint16_t port) {
- port_ = port;
+ private:
+#ifdef WIN32
+ struct SocketInitializer
+ {
+ SocketInitializer() {
+ static WSADATA s_wsaData;
+ const int iWinSockInitResult = WSAStartup(MAKEWORD(2, 2), &s_wsaData);
+ if (0 != iWinSockInitResult) {
+ throw std::exception("Cannot start client");
+ }
+ }
+ ~SocketInitializer() noexcept {
+ WSACleanup();
+ }
+ };
+ static void initialize_socket() {
+ static SocketInitializer initialized;
}
+#else
+ static void initialize_socket(){}
+#endif /* WIN32 */
- private:
static std::string init_hostname() {
+ initialize_socket();
char hostname[1024];
gethostname(hostname, 1024);
Socket mySock(nullptr, hostname, 0);
mySock.initialize();
- auto resolved_hostname = mySock.getHostname();
+ const auto resolved_hostname = mySock.getHostname();
return !IsNullOrEmpty(resolved_hostname) ? resolved_hostname : hostname;
}
};
@@ -301,4 +341,4 @@ class Socket : public BaseStream {
} /* namespace nifi */
} /* namespace apache */
} /* namespace org */
-#endif /* LIBMINIFI_INCLUDE_IO_POSIX_CLIENTSOCKET_H_ */
+#endif /* LIBMINIFI_INCLUDE_IO_CLIENTSOCKET_H_ */
diff --git a/libminifi/include/io/DataStream.h b/libminifi/include/io/DataStream.h
index a5700b3..b5be417 100644
--- a/libminifi/include/io/DataStream.h
+++ b/libminifi/include/io/DataStream.h
@@ -36,10 +36,8 @@ namespace io {
*/
class DataStream {
public:
-
DataStream() = default;
-
- virtual ~DataStream() = default;
+ virtual ~DataStream() noexcept = default;
/**
* Constructor
diff --git a/libminifi/include/io/DescriptorStream.h b/libminifi/include/io/DescriptorStream.h
index e6d843c..8389c08 100644
--- a/libminifi/include/io/DescriptorStream.h
+++ b/libminifi/include/io/DescriptorStream.h
@@ -47,17 +47,15 @@ class DescriptorStream : public io::BaseStream {
*/
explicit DescriptorStream(int fd);
- virtual ~DescriptorStream() {
-
- }
+ ~DescriptorStream() override = default;
/**
* Skip to the specified offset.
* @param offset offset to which we will skip
*/
- void seek(uint64_t offset);
+ void seek(uint64_t offset) override;
- const uint64_t getSize() const {
+ const uint64_t getSize() const override {
return -1;
}
@@ -67,13 +65,13 @@ class DescriptorStream : public io::BaseStream {
* @param buf buffer in which we extract data
* @param buflen
*/
- virtual int readData(std::vector<uint8_t> &buf, int buflen);
+ int readData(std::vector<uint8_t> &buf, int buflen) override;
/**
* Reads data and places it into buf
* @param buf buffer in which we extract data
* @param buflen
*/
- virtual int readData(uint8_t *buf, int buflen);
+ int readData(uint8_t *buf, int buflen) override;
/**
* Write value to the stream using std::vector
@@ -88,12 +86,12 @@ class DescriptorStream : public io::BaseStream {
* @param value value to write
* @param size size of value
*/
- virtual int writeData(uint8_t *value, int size);
+ int writeData(uint8_t *value, int size) override;
/**
* Returns the underlying buffer
* @return vector's array
- **/
+ */
const uint8_t *getBuffer() const {
throw std::runtime_error("Stream does not support this operation");
}
@@ -104,7 +102,7 @@ class DescriptorStream : public io::BaseStream {
* @param stream stream from which we will read
* @return resulting read size
**/
- virtual int read(uint8_t &value);
+ int read(uint8_t &value) override;
/**
* reads two bytes from the stream
@@ -112,7 +110,7 @@ class DescriptorStream : public io::BaseStream {
* @param stream stream from which we will read
* @return resulting read size
**/
- virtual int read(uint16_t &base_value, bool is_little_endian = false);
+ int read(uint16_t &base_value, bool is_little_endian = false) override;
/**
* reads a byte from the stream
@@ -120,7 +118,7 @@ class DescriptorStream : public io::BaseStream {
* @param stream stream from which we will read
* @return resulting read size
**/
- virtual int read(char &value);
+ int read(char &value) override;
/**
* reads a byte array from the stream
@@ -129,7 +127,7 @@ class DescriptorStream : public io::BaseStream {
* @param stream stream from which we will read
* @return resulting read size
**/
- virtual int read(uint8_t *value, int len);
+ int read(uint8_t *value, int len) override;
/**
* reads four bytes from the stream
@@ -137,7 +135,7 @@ class DescriptorStream : public io::BaseStream {
* @param stream stream from which we will read
* @return resulting read size
**/
- virtual int read(uint32_t &value, bool is_little_endian = false);
+ int read(uint32_t &value, bool is_little_endian = false) override;
/**
* reads eight byte from the stream
@@ -145,7 +143,7 @@ class DescriptorStream : public io::BaseStream {
* @param stream stream from which we will read
* @return resulting read size
**/
- virtual int read(uint64_t &value, bool is_little_endian = false);
+ int read(uint64_t &value, bool is_little_endian = false) override;
/**
@@ -154,7 +152,7 @@ class DescriptorStream : public io::BaseStream {
* @param stream stream from which we will read
* @return resulting read size
**/
- virtual int readUTF(std::string &str, bool widen = false);
+ int readUTF(std::string &str, bool widen = false) override;
protected:
diff --git a/libminifi/include/io/FileStream.h b/libminifi/include/io/FileStream.h
index 04dc41b..52675d6 100644
--- a/libminifi/include/io/FileStream.h
+++ b/libminifi/include/io/FileStream.h
@@ -52,18 +52,18 @@ class FileStream : public io::BaseStream {
*/
explicit FileStream(const std::string &path, bool append = false);
- virtual ~FileStream() {
+ ~FileStream() override {
closeStream();
}
- virtual void closeStream();
+ void closeStream() override;
/**
* Skip to the specified offset.
* @param offset offset to which we will skip
*/
- void seek(uint64_t offset);
+ void seek(uint64_t offset) override;
- const uint64_t getSize() const {
+ const uint64_t getSize() const override {
return length_;
}
@@ -73,13 +73,13 @@ class FileStream : public io::BaseStream {
* @param buf buffer in which we extract data
* @param buflen
*/
- virtual int readData(std::vector<uint8_t> &buf, int buflen);
+ int readData(std::vector<uint8_t> &buf, int buflen) override;
/**
* Reads data and places it into buf
* @param buf buffer in which we extract data
* @param buflen
*/
- virtual int readData(uint8_t *buf, int buflen);
+ int readData(uint8_t *buf, int buflen) override;
/**
* Write value to the stream using std::vector
@@ -94,7 +94,7 @@ class FileStream : public io::BaseStream {
* @param value value to write
* @param size size of value
*/
- virtual int writeData(uint8_t *value, int size);
+ int writeData(uint8_t *value, int size) override;
/**
* Returns the underlying buffer
@@ -121,9 +121,7 @@ class FileStream : public io::BaseStream {
size_t length_;
private:
-
std::shared_ptr<logging::Logger> logger_;
-
};
} /* namespace io */
diff --git a/libminifi/include/io/NetworkPrioritizer.h b/libminifi/include/io/NetworkPrioritizer.h
index 7f9fcad..e1aa770 100644
--- a/libminifi/include/io/NetworkPrioritizer.h
+++ b/libminifi/include/io/NetworkPrioritizer.h
@@ -20,6 +20,7 @@
#include <iostream>
#include <memory>
+#include <utility>
namespace org {
namespace apache {
@@ -29,48 +30,30 @@ namespace io {
class NetworkInterface;
-class NetworkPrioritizer {
- public:
-
- virtual ~NetworkPrioritizer() {
- }
-
+struct NetworkPrioritizer {
+ virtual ~NetworkPrioritizer() noexcept = default;
virtual NetworkInterface getInterface(uint32_t size) = 0;
protected:
friend class NetworkInterface;
virtual void reduce_tokens(uint32_t size) = 0;
-
};
class NetworkInterface {
public:
+ NetworkInterface() = default;
- NetworkInterface()
- : prioritizer_(nullptr) {
- }
+ NetworkInterface(std::string ifc, std::shared_ptr<NetworkPrioritizer> prioritizer)
+ : ifc_{std::move(ifc)}, prioritizer_{std::move(prioritizer)}
+ { }
- virtual ~NetworkInterface() {
- }
+ NetworkInterface(const NetworkInterface &other) = default;
+ NetworkInterface(NetworkInterface &&other) noexcept = default;
- explicit NetworkInterface(const std::string &ifc, const std::shared_ptr<NetworkPrioritizer> &prioritizer)
- : prioritizer_(prioritizer) {
- ifc_ = ifc;
- }
+ virtual ~NetworkInterface() = default;
- NetworkInterface(const NetworkInterface &other)
- : prioritizer_(other.prioritizer_) {
- ifc_ = other.ifc_;
- }
+ std::string getInterface() const { return ifc_; }
- explicit NetworkInterface(const NetworkInterface &&other)
- : ifc_(std::move(other.ifc_)),
- prioritizer_(std::move(other.prioritizer_)) {
- }
-
- std::string getInterface() const {
- return ifc_;
- }
void log_write(uint32_t size) {
if (nullptr != prioritizer_) {
prioritizer_->reduce_tokens(size);
@@ -83,29 +66,26 @@ class NetworkInterface {
}
}
- NetworkInterface &operator=(const NetworkInterface &&other) {
- ifc_ = std::move(other.ifc_);
- prioritizer_ = std::move(other.prioritizer_);
- return *this;
- }
+ NetworkInterface &operator=(const NetworkInterface &other) = default;
+ NetworkInterface &operator=(NetworkInterface &&other) = default;
+
private:
- friend class NetworkPrioritizer;
+ friend struct NetworkPrioritizer;
+
std::string ifc_;
std::shared_ptr<NetworkPrioritizer> prioritizer_;
};
class NetworkPrioritizerFactory {
public:
- NetworkPrioritizerFactory()
- : np_(nullptr) {
+ NetworkPrioritizerFactory() = default;
- }
static std::shared_ptr<NetworkPrioritizerFactory> getInstance() {
static std::shared_ptr<NetworkPrioritizerFactory> fa = std::make_shared<NetworkPrioritizerFactory>();
return fa;
}
- void clearPrioritizer(){
+ void clearPrioritizer() {
np_ = nullptr;
}
@@ -119,6 +99,7 @@ class NetworkPrioritizerFactory {
std::shared_ptr<NetworkPrioritizer> getPrioritizer() {
return np_;
}
+
private:
std::shared_ptr<NetworkPrioritizer> np_;
};
diff --git a/libminifi/include/io/NonConvertingStream.h b/libminifi/include/io/NonConvertingStream.h
index aa16995..64eeb7f 100644
--- a/libminifi/include/io/NonConvertingStream.h
+++ b/libminifi/include/io/NonConvertingStream.h
@@ -56,11 +56,11 @@ class NonConvertingStream : public BaseStream {
* @param is_little_endian endianness determination
* @return resulting write size
**/
- virtual int write(uint32_t base_value, bool is_little_endian = false);
+ int write(uint32_t base_value, bool is_little_endian = false) override;
- int writeData(uint8_t *value, int size);
+ int writeData(uint8_t *value, int size) override;
- virtual void seek(uint64_t offset) {
+ void seek(uint64_t offset) override {
if (composable_stream_ != this) {
composable_stream_->seek(offset);
} else {
@@ -75,7 +75,7 @@ class NonConvertingStream : public BaseStream {
* @param is_little_endian endianness determination
* @return resulting write size
**/
- virtual int write(uint16_t base_value, bool is_little_endian = false);
+ int write(uint16_t base_value, bool is_little_endian = false) override;
/**
* write valueto stream
@@ -84,7 +84,7 @@ class NonConvertingStream : public BaseStream {
* @param strema output stream
* @return resulting write size
**/
- virtual int write(uint8_t *value, int len);
+ int write(uint8_t *value, int len) override;
/**
* write 8 bytes to stream
@@ -93,21 +93,21 @@ class NonConvertingStream : public BaseStream {
* @param is_little_endian endianness determination
* @return resulting write size
**/
- virtual int write(uint64_t base_value, bool is_little_endian = false);
+ int write(uint64_t base_value, bool is_little_endian = false) override;
/**
* write bool to stream
* @param value non encoded value
* @return resulting write size
**/
- virtual int write(bool value);
+ int write(bool value) override;
/**
* write UTF string to stream
* @param str string to write
* @return resulting write size
**/
- virtual int writeUTF(std::string str, bool widen = false);
+ int writeUTF(std::string str, bool widen = false) override;
/**
* reads a byte from the stream
@@ -115,20 +115,20 @@ class NonConvertingStream : public BaseStream {
* @param stream stream from which we will read
* @return resulting read size
**/
- virtual int read(uint8_t &value);
+ int read(uint8_t &value) override;
/**
* Reads data and places it into buf
* @param buf buffer in which we extract data
* @param buflen
*/
- virtual int readData(std::vector<uint8_t> &buf, int buflen);
+ int readData(std::vector<uint8_t> &buf, int buflen) override;
/**
* Reads data and places it into buf
* @param buf buffer in which we extract data
* @param buflen
*/
- virtual int readData(uint8_t *buf, int buflen);
+ int readData(uint8_t *buf, int buflen) override;
/**
* reads two bytes from the stream
@@ -136,7 +136,7 @@ class NonConvertingStream : public BaseStream {
* @param stream stream from which we will read
* @return resulting read size
**/
- virtual int read(uint16_t &base_value, bool is_little_endian = false);
+ int read(uint16_t &base_value, bool is_little_endian = false) override;
/**
* reads a byte from the stream
@@ -144,7 +144,7 @@ class NonConvertingStream : public BaseStream {
* @param stream stream from which we will read
* @return resulting read size
**/
- virtual int read(char &value);
+ int read(char &value) override;
/**
* reads a byte array from the stream
@@ -153,7 +153,7 @@ class NonConvertingStream : public BaseStream {
* @param stream stream from which we will read
* @return resulting read size
**/
- virtual int read(uint8_t *value, int len);
+ int read(uint8_t *value, int len) override;
/**
* reads four bytes from the stream
@@ -161,7 +161,7 @@ class NonConvertingStream : public BaseStream {
* @param stream stream from which we will read
* @return resulting read size
**/
- virtual int read(uint32_t &value, bool is_little_endian = false);
+ int read(uint32_t &value, bool is_little_endian = false) override;
/**
* reads eight byte from the stream
@@ -169,9 +169,9 @@ class NonConvertingStream : public BaseStream {
* @param stream stream from which we will read
* @return resulting read size
**/
- virtual int read(uint64_t &value, bool is_little_endian = false);
+ int read(uint64_t &value, bool is_little_endian = false) override;
- virtual const uint64_t getSize() const {
+ const uint64_t getSize() const override {
if (composable_stream_ == this){
return buffer.size();
}
@@ -187,7 +187,7 @@ class NonConvertingStream : public BaseStream {
* @param stream stream from which we will read
* @return resulting read size
**/
- virtual int readUTF(std::string &str, bool widen = false);
+ int readUTF(std::string &str, bool widen = false) override;
protected:
DataStream *composable_stream_;
};
diff --git a/libminifi/include/io/Serializable.h b/libminifi/include/io/Serializable.h
index c569c75..4088257 100644
--- a/libminifi/include/io/Serializable.h
+++ b/libminifi/include/io/Serializable.h
@@ -85,7 +85,7 @@ class Serializable {
* @param strema output stream
* @return resulting write size
**/
- int write(const uint8_t * const value, int len, DataStream *stream);
+ int write(const uint8_t *value, int len, DataStream *stream);
/**
* write bool to stream
diff --git a/libminifi/include/io/ServerSocket.h b/libminifi/include/io/ServerSocket.h
index fefea0f..7d2ee8a 100644
--- a/libminifi/include/io/ServerSocket.h
+++ b/libminifi/include/io/ServerSocket.h
@@ -28,17 +28,12 @@ namespace io {
class BaseServerSocket {
-
public:
-
- virtual ~BaseServerSocket(){
-
- }
+ virtual ~BaseServerSocket()= default;
virtual int16_t initialize(bool loopbackOnly) = 0;
virtual void registerCallback(std::function<bool()> accept_function, std::function<void(io::BaseStream *)> handler) = 0;
-
};
/**
* Purpose: Server socket abstraction that makes focusing the accept/block paradigm
@@ -46,7 +41,7 @@ class BaseServerSocket {
*/
class ServerSocket : public BaseServerSocket, public Socket {
public:
- explicit ServerSocket(const std::shared_ptr<SocketContext> &context, const std::string &hostname, const uint16_t port, const uint16_t listeners);
+ explicit ServerSocket(const std::shared_ptr<SocketContext> &context, const std::string &hostname, uint16_t port, uint16_t listeners);
virtual ~ServerSocket();
diff --git a/libminifi/include/io/StreamFactory.h b/libminifi/include/io/StreamFactory.h
index 6f71434..3e61a05 100644
--- a/libminifi/include/io/StreamFactory.h
+++ b/libminifi/include/io/StreamFactory.h
@@ -31,18 +31,16 @@ namespace io {
class AbstractStreamFactory {
public:
- virtual ~AbstractStreamFactory() {
- }
+ virtual ~AbstractStreamFactory() = default;
- virtual std::unique_ptr<Socket> createSocket(const std::string &host, const uint16_t port) = 0;
+ virtual std::unique_ptr<Socket> createSocket(const std::string &host, uint16_t port) = 0;
- virtual std::unique_ptr<Socket> createSecureSocket(const std::string &host, const uint16_t port, const std::shared_ptr<minifi::controllers::SSLContextService> &ssl_service) = 0;
+ virtual std::unique_ptr<Socket> createSecureSocket(const std::string &host, uint16_t port, const std::shared_ptr<minifi::controllers::SSLContextService> &ssl_service) = 0;
};
/**
Purpose: Due to the current design this is the only mechanism by which we can
inject different socket types
-
**/
class StreamFactory {
public:
@@ -90,7 +88,6 @@ class StreamFactory {
}
protected:
-
StreamFactory(const std::shared_ptr<Configure> &configure);
std::shared_ptr<AbstractStreamFactory> delegate_;
diff --git a/libminifi/include/io/tls/SecureDescriptorStream.h b/libminifi/include/io/tls/SecureDescriptorStream.h
index c791ae8..2528323 100644
--- a/libminifi/include/io/tls/SecureDescriptorStream.h
+++ b/libminifi/include/io/tls/SecureDescriptorStream.h
@@ -49,17 +49,15 @@ class SecureDescriptorStream : public io::BaseStream {
*/
explicit SecureDescriptorStream(int fd, SSL *s);
- virtual ~SecureDescriptorStream() {
-
- }
+ ~SecureDescriptorStream() override = default;
/**
* Skip to the specified offset.
* @param offset offset to which we will skip
*/
- void seek(uint64_t offset);
+ void seek(uint64_t offset) override;
- const uint64_t getSize() const {
+ const uint64_t getSize() const override {
return -1;
}
@@ -69,13 +67,13 @@ class SecureDescriptorStream : public io::BaseStream {
* @param buf buffer in which we extract data
* @param buflen
*/
- virtual int readData(std::vector<uint8_t> &buf, int buflen);
+ int readData(std::vector<uint8_t> &buf, int buflen) override;
/**
* Reads data and places it into buf
* @param buf buffer in which we extract data
* @param buflen
*/
- virtual int readData(uint8_t *buf, int buflen);
+ int readData(uint8_t *buf, int buflen) override;
/**
* Write value to the stream using std::vector
@@ -90,12 +88,12 @@ class SecureDescriptorStream : public io::BaseStream {
* @param value value to write
* @param size size of value
*/
- virtual int writeData(uint8_t *value, int size);
+ int writeData(uint8_t *value, int size) override;
/**
* Returns the underlying buffer
* @return vector's array
- **/
+ */
const uint8_t *getBuffer() const {
throw std::runtime_error("Stream does not support this operation");
}
@@ -106,7 +104,7 @@ class SecureDescriptorStream : public io::BaseStream {
* @param stream stream from which we will read
* @return resulting read size
**/
- virtual int read(uint8_t &value);
+ int read(uint8_t &value) override;
/**
* reads two bytes from the stream
@@ -114,7 +112,7 @@ class SecureDescriptorStream : public io::BaseStream {
* @param stream stream from which we will read
* @return resulting read size
**/
- virtual int read(uint16_t &base_value, bool is_little_endian = false);
+ int read(uint16_t &base_value, bool is_little_endian = false) override;
/**
* reads a byte from the stream
@@ -122,7 +120,7 @@ class SecureDescriptorStream : public io::BaseStream {
* @param stream stream from which we will read
* @return resulting read size
**/
- virtual int read(char &value);
+ int read(char &value) override;
/**
* reads a byte array from the stream
@@ -131,7 +129,7 @@ class SecureDescriptorStream : public io::BaseStream {
* @param stream stream from which we will read
* @return resulting read size
**/
- virtual int read(uint8_t *value, int len);
+ int read(uint8_t *value, int len) override;
/**
* reads four bytes from the stream
@@ -139,7 +137,7 @@ class SecureDescriptorStream : public io::BaseStream {
* @param stream stream from which we will read
* @return resulting read size
**/
- virtual int read(uint32_t &value, bool is_little_endian = false);
+ int read(uint32_t &value, bool is_little_endian = false) override;
/**
* reads eight byte from the stream
@@ -147,7 +145,7 @@ class SecureDescriptorStream : public io::BaseStream {
* @param stream stream from which we will read
* @return resulting read size
**/
- virtual int read(uint64_t &value, bool is_little_endian = false);
+ int read(uint64_t &value, bool is_little_endian = false) override;
/**
@@ -156,7 +154,7 @@ class SecureDescriptorStream : public io::BaseStream {
* @param stream stream from which we will read
* @return resulting read size
**/
- virtual int readUTF(std::string &str, bool widen = false);
+ int readUTF(std::string &str, bool widen = false) override;
protected:
diff --git a/libminifi/include/io/tls/TLSSocket.h b/libminifi/include/io/tls/TLSSocket.h
index 74794ea..723a687 100644
--- a/libminifi/include/io/tls/TLSSocket.h
+++ b/libminifi/include/io/tls/TLSSocket.h
@@ -43,18 +43,8 @@ namespace io {
class OpenSSLInitializer {
public:
static OpenSSLInitializer *getInstance() {
- OpenSSLInitializer* atomic_context = context_instance.load(std::memory_order_relaxed);
- std::atomic_thread_fence(std::memory_order_acquire);
- if (atomic_context == nullptr) {
- std::lock_guard<std::mutex> lock(context_mutex);
- atomic_context = context_instance.load(std::memory_order_relaxed);
- if (atomic_context == nullptr) {
- atomic_context = new OpenSSLInitializer();
- std::atomic_thread_fence(std::memory_order_release);
- context_instance.store(atomic_context, std::memory_order_relaxed);
- }
- }
- return atomic_context;
+ static OpenSSLInitializer openssl_initializer;
+ return &openssl_initializer;
}
OpenSSLInitializer() {
@@ -62,18 +52,14 @@ class OpenSSLInitializer {
OpenSSL_add_all_algorithms();
SSL_load_error_strings();
}
- private:
- static std::atomic<OpenSSLInitializer*> context_instance;
- static std::mutex context_mutex;
};
class TLSContext : public SocketContext {
-
public:
- TLSContext(const std::shared_ptr<Configure> &configure, const std::shared_ptr<minifi::controllers::SSLContextService> &ssl_service = nullptr);
+ TLSContext(const std::shared_ptr<Configure> &configure, std::shared_ptr<minifi::controllers::SSLContextService> ssl_service = nullptr);
virtual ~TLSContext() {
- if (0 != ctx)
+ if (nullptr != ctx)
SSL_CTX_free(ctx);
}
@@ -88,8 +74,6 @@ class TLSContext : public SocketContext {
int16_t initialize(bool server_method = false);
private:
-
-
std::shared_ptr<logging::Logger> logger_;
std::shared_ptr<Configure> configure_;
std::shared_ptr<minifi::controllers::SSLContextService> ssl_service_;
@@ -109,7 +93,7 @@ class TLSSocket : public Socket {
* @param port connecting port
* @param listeners number of listeners in the queue
*/
- explicit TLSSocket(const std::shared_ptr<TLSContext> &context, const std::string &hostname, const uint16_t port, const uint16_t listeners);
+ explicit TLSSocket(const std::shared_ptr<TLSContext> &context, const std::string &hostname, uint16_t port, uint16_t listeners);
/**
* Constructor that creates a client socket.
@@ -117,12 +101,14 @@ class TLSSocket : public Socket {
* @param hostname hostname we are connecting to.
* @param port port we are connecting to.
*/
- explicit TLSSocket(const std::shared_ptr<TLSContext> &context, const std::string &hostname, const uint16_t port);
+ explicit TLSSocket(const std::shared_ptr<TLSContext> &context, const std::string &hostname, uint16_t port);
/**
* Move constructor.
*/
- explicit TLSSocket(const TLSSocket &&);
+ TLSSocket(TLSSocket &&);
+
+ TLSSocket& operator=(TLSSocket&&);
virtual ~TLSSocket();
@@ -130,7 +116,7 @@ class TLSSocket : public Socket {
* Initializes the socket
* @return result of the creation operation.
*/
- int16_t initialize(){
+ int16_t initialize() {
return initialize(true);
}
@@ -141,7 +127,7 @@ class TLSSocket : public Socket {
* @param msec timeout interval to wait
* @returns file descriptor
*/
- virtual int16_t select_descriptor(const uint16_t msec);
+ virtual int16_t select_descriptor(uint16_t msec);
virtual int readData(std::vector<uint8_t> &buf, int buflen);
@@ -189,9 +175,10 @@ class TLSSocket : public Socket {
}
void close_ssl(int fd);
- std::atomic<bool> connected_;
+
+ std::atomic<bool> connected_{ false };
std::shared_ptr<TLSContext> context_;
- SSL* ssl_;
+ SSL* ssl_{ nullptr };
std::mutex ssl_mutex_;
std::map<int, SSL*> ssl_map_;
};
diff --git a/libminifi/include/sitetosite/Peer.h b/libminifi/include/sitetosite/Peer.h
index 05bb4aa..4c164a8 100644
--- a/libminifi/include/sitetosite/Peer.h
+++ b/libminifi/include/sitetosite/Peer.h
@@ -161,7 +161,7 @@ class SiteToSitePeer : public org::apache::nifi::minifi::io::BaseStream {
url_ = "nifi://" + host_ + ":" + std::to_string(port_);
yield_expiration_ = 0;
timeout_ = 30000; // 30 seconds
- local_network_interface_= std::move(io::NetworkInterface(ifc, nullptr));
+ local_network_interface_= io::NetworkInterface(ifc, nullptr);
}
explicit SiteToSitePeer(SiteToSitePeer &&ss)
@@ -189,7 +189,7 @@ class SiteToSitePeer : public org::apache::nifi::minifi::io::BaseStream {
}
// setInterface
void setInterface(std::string &ifc) {
- local_network_interface_ = std::move(io::NetworkInterface(ifc,nullptr));
+ local_network_interface_ = io::NetworkInterface(ifc,nullptr);
}
std::string getInterface() {
return local_network_interface_.getInterface();
diff --git a/libminifi/include/utils/Deleters.h b/libminifi/include/utils/Deleters.h
index 5a56dfb..b6d9a3e 100644
--- a/libminifi/include/utils/Deleters.h
+++ b/libminifi/include/utils/Deleters.h
@@ -17,20 +17,42 @@
#ifndef LIBMINIFI_INCLUDE_UTILS_DELETERS_H
#define LIBMINIFI_INCLUDE_UTILS_DELETERS_H
+#include <cstdlib>
+#ifdef WIN32
+#include <WS2tcpip.h>
+#else
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <netdb.h>
+#include <ifaddrs.h>
+#endif /* WIN32 */
+
namespace org {
namespace apache {
namespace nifi {
namespace minifi {
namespace utils {
-class FreeDeleter
-{
-public:
- void operator()(void* ptr) {
- // free(null) is guaranteed to be safe, so no need to be defensive.
- free(ptr);
- }
+struct FreeDeleter {
+ void operator()(void* const ptr) const noexcept {
+ // free(null) is guaranteed to be safe, so no need to be defensive.
+ free(ptr);
+ }
+};
+
+struct addrinfo_deleter {
+ void operator()(addrinfo* const p) const noexcept {
+ freeaddrinfo(p);
+ }
+};
+
+#ifndef WIN32
+struct ifaddrs_deleter {
+ void operator()(ifaddrs* const p) const noexcept {
+ freeifaddrs(p);
+ }
};
+#endif /* !WIN32 */
} /* namespace utils */
} /* namespace minifi */
diff --git a/libminifi/include/utils/GeneralUtils.h b/libminifi/include/utils/GeneralUtils.h
index a5e2ff9..42b1e89 100644
--- a/libminifi/include/utils/GeneralUtils.h
+++ b/libminifi/include/utils/GeneralUtils.h
@@ -21,6 +21,8 @@
#include <memory>
#include <type_traits>
+#include <utility>
+#include <functional>
namespace org {
namespace apache {
@@ -28,23 +30,40 @@ namespace nifi {
namespace minifi {
namespace utils {
+#if __cplusplus < 201402L
template<typename T, typename... Args>
std::unique_ptr<T> make_unique(Args&&... args) {
return std::unique_ptr<T>{ new T{ std::forward<Args>(args)... } };
}
+#else
+using std::make_unique;
+#endif /* < C++14 */
-template<typename T, typename std::enable_if<std::is_integral<T>::value>::type* = nullptr>
+template<typename T, typename = typename std::enable_if<std::is_integral<T>::value>::type>
T intdiv_ceil(T numerator, T denominator) {
// note: division and remainder is 1 instruction on x86
return numerator / denominator + (numerator % denominator > 0);
}
-template <typename T, typename std::enable_if<std::is_pointer<T>::value>::type* = nullptr>
+template <typename T, typename = typename std::enable_if<std::is_pointer<T>::value>::type>
using owner = T;
+#if __cplusplus < 201402L
+// from https://en.cppreference.com/w/cpp/utility/exchange
+template<typename T, typename U = T>
+T exchange(T& obj, U&& new_value) {
+ T old_value = std::move(obj);
+ obj = std::forward<U>(new_value);
+ return old_value;
+}
+#else
+using std::exchange;
+#endif /* < C++14 */
+
} /* namespace utils */
} /* namespace minifi */
} /* namespace nifi */
} /* namespace apache */
} /* namespace org */
+
#endif /* LIBMINIFI_INCLUDE_UTILS_GENERAL_UTILS_H */
diff --git a/libminifi/include/utils/Monitors.h b/libminifi/include/utils/Monitors.h
index a9ff485..f023252 100644
--- a/libminifi/include/utils/Monitors.h
+++ b/libminifi/include/utils/Monitors.h
@@ -114,7 +114,8 @@ class SingleRunMonitor : public utils::AfterExecute<bool>{
struct TaskRescheduleInfo {
TaskRescheduleInfo(bool result, std::chrono::milliseconds wait_time)
- : finished_(result), wait_time_(wait_time){}
+ : wait_time_(wait_time), finished_(result) {}
+
std::chrono::milliseconds wait_time_;
bool finished_;
diff --git a/libminifi/opsys/win/io/ClientSocket.h b/libminifi/opsys/win/io/ClientSocket.h
deleted file mode 100644
index 2144441..0000000
--- a/libminifi/opsys/win/io/ClientSocket.h
+++ /dev/null
@@ -1,350 +0,0 @@
-/**
- *
- * 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_IO_WIN__CLIENTSOCKET_H_
-#define LIBMINIFI_INCLUDE_IO_WIN__CLIENTSOCKET_H_
-
-#include <cstdint>
-#include <sys/types.h>
-#ifdef WIN32
-#ifndef WIN32_LEAN_AND_MEAN
-#define WIN32_LEAN_AND_MEAN
-#endif
-//#include <WinSock.h>
-//#include <WS2tcpip.h>
-#include <Windows.h>
-#include <WS2tcpip.h>
-#pragma comment(lib, "Ws2_32.lib")
-#else
-#include <sys/types.h>
-#include <sys/socket.h>
-#include <netdb.h>
-#include <unistd.h>
-#endif
-#include <mutex>
-#include <atomic>
-#include "io/BaseStream.h"
-#include "core/Core.h"
-#include "core/logging/Logger.h"
-#include "io/validation.h"
-#include "properties/Configure.h"
-#include "io/NetworkPrioritizer.h"
-
-
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-namespace io {
-
-#ifdef WIN32
- typedef SOCKET SocketDescriptor;
-#else
- typedef int SocketDescriptor;
- #define INVALID_SOCKET -1
-#endif
-
-
-/**
- * Context class for socket. This is currently only used as a parent class for TLSContext. It is necessary so the Socket and TLSSocket constructors
- * can be the same. It also gives us a common place to set timeouts, etc from the Configure object in the future.
- */
-class SocketContext {
- public:
- SocketContext(const std::shared_ptr<Configure> &configure) {
- }
-};
-/**
- * Socket class.
- * Purpose: Provides a general purpose socket interface that abstracts
- * connecting information from users
- * Design: Extends DataStream and allows us to perform most streaming
- * operations against a BSD socket
- *
- *
- */
-class Socket : public BaseStream {
-public:
- /**
- * Constructor that creates a client socket.
- * @param context the SocketContext
- * @param hostname hostname we are connecting to.
- * @param port port we are connecting to.
- */
- explicit Socket(const std::shared_ptr<SocketContext> &context, const std::string &hostname, const uint16_t port);
-
- /**
- * Move constructor.
- */
- explicit Socket(const Socket &&);
-
- /**
- * Static function to return the current machine's host name
- */
- static std::string getMyHostName() {
- static std::string HOSTNAME = init_hostname();
- return HOSTNAME;
- }
-
- /**
- * Destructor
- */
-
- virtual ~Socket();
-
- virtual void closeStream();
- /**
- * Initializes the socket
- * @return result of the creation operation.
- */
- virtual int16_t initialize();
-
- virtual void setInterface(io::NetworkInterface &&ifc) {
- local_network_interface_ = std::move(ifc);
- }
-
- /**
- * Sets the non blocking flag on the file descriptor.
- */
- void setNonBlocking();
-
- std::string getHostname() const;
-
- /**
- * Return the port for this socket
- * @returns port
- */
- uint16_t getPort() const;
-
- // data stream extensions
- /**
- * Reads data and places it into buf
- * @param buf buffer in which we extract data
- * @param buflen
- * @param retrieve_all_bytes determines if we should read all bytes before returning
- */
- virtual int readData(std::vector<uint8_t> &buf, int buflen) {
- return readData(buf, buflen, true);
- }
- /**
- * Reads data and places it into buf
- * @param buf buffer in which we extract data
- * @param buflen
- * @param retrieve_all_bytes determines if we should read all bytes before returning
- */
- virtual int readData(uint8_t *buf, int buflen) {
- return readData(buf, buflen, true);
- }
-
- /**
- * Reads data and places it into buf
- * @param buf buffer in which we extract data
- * @param buflen
- * @param retrieve_all_bytes determines if we should read all bytes before returning
- */
- virtual int readData(std::vector<uint8_t> &buf, int buflen, bool retrieve_all_bytes);
- /**
- * Reads data and places it into buf
- * @param buf buffer in which we extract data
- * @param buflen
- * @param retrieve_all_bytes determines if we should read all bytes before returning
- */
- virtual int readData(uint8_t *buf, int buflen, bool retrieve_all_bytes);
-
- /**
- * Write value to the stream using std::vector
- * @param buf incoming buffer
- * @param buflen buffer to write
- *
- */
- virtual int writeData(std::vector<uint8_t> &buf, int buflen);
-
- /**
- * writes value to stream
- * @param value value to write
- * @param size size of value
- */
- virtual int writeData(uint8_t *value, int size);
-
- /**
- * Writes a system word
- * @param value value to write
- */
- virtual int write(uint64_t value, bool is_little_endian = EndiannessCheck::IS_LITTLE);
-
- /**
- * Writes a uint32_t
- * @param value value to write
- */
- virtual int write(uint32_t value, bool is_little_endian = EndiannessCheck::IS_LITTLE);
-
- /**
- * Writes a system short
- * @param value value to write
- */
- virtual int write(uint16_t value, bool is_little_endian = EndiannessCheck::IS_LITTLE);
-
- /**
- * Reads a system word
- * @param value value to write
- */
- virtual int read(uint64_t &value, bool is_little_endian = EndiannessCheck::IS_LITTLE);
-
- /**
- * Reads a uint32_t
- * @param value value to write
- */
- virtual int read(uint32_t &value, bool is_little_endian = EndiannessCheck::IS_LITTLE);
-
- /**
- * Reads a system short
- * @param value value to write
- */
- virtual int read(uint16_t &value, bool is_little_endian = EndiannessCheck::IS_LITTLE);
-
- /**
- * Returns the underlying buffer
- * @return vector's array
- **/
- const uint8_t *getBuffer() const {
- return DataStream::getBuffer();
- }
-
- /**
- * Retrieve size of data stream
- * @return size of data stream
- **/
- const uint64_t getSize() const {
- return DataStream::getSize();
- }
-
-protected:
-
- void setPort(uint16_t port) {
- port_ = port;
- }
-
-
- /**
- * Constructor that accepts host name, port and listeners. With this
- * contructor we will be creating a server socket
- * @param context the SocketContext
- * @param hostname our host name
- * @param port connecting port
- * @param listeners number of listeners in the queue
- */
- explicit Socket(const std::shared_ptr<SocketContext> &context, const std::string &hostname, const uint16_t port, const uint16_t listeners);
-
- /**
- * Creates a vector and returns the vector using the provided
- * type name.
- * @param t incoming object
- * @returns vector.
- */
- template<typename T>
- std::vector<uint8_t> readBuffer(const T&);
-
- /**
- * Creates a connection using the address info object.
- * @param p addrinfo structure.
- * @returns fd.
- */
-#ifdef WIN32
- virtual int8_t createConnection(const addrinfo *p, struct in_addr &addr);
-#else
- virtual int8_t createConnection(const addrinfo *p, in_addr_t &addr);
-#endif
-
- /**
- * Sets socket options depending on the instance.
- * @param sock socket file descriptor.
- */
- virtual int16_t setSocketOptions(const SocketDescriptor sock);
-
- /**
- * Attempt to select the socket file descriptor
- * @param msec timeout interval to wait
- * @returns file descriptor
- */
- virtual int16_t select_descriptor(const uint16_t msec);
-
-
- std::recursive_mutex selection_mutex_;
-
- std::string requested_hostname_;
- std::string canonical_hostname_;
- uint16_t port_;
-
- bool is_loopback_only_;
- io::NetworkInterface local_network_interface_;
-
- // connection information
- SocketDescriptor socket_file_descriptor_;
-
- fd_set total_list_;
- fd_set read_fds_;
- std::atomic<uint16_t> socket_max_;
- std::atomic<uint64_t> total_written_;
- std::atomic<uint64_t> total_read_;
- uint16_t listeners_;
-
- bool nonBlocking_;
-
- std::shared_ptr<logging::Logger> logger_;
-
-private:
-
- class SocketInitializer
- {
- public:
- SocketInitializer() {
- #ifdef WIN32
- static WSADATA s_wsaData;
- int iWinSockInitResult = WSAStartup(MAKEWORD(2, 2), &s_wsaData);
- if (0 != iWinSockInitResult) {
- throw std::exception("Cannot start client");
- }
- #endif
- }
- ~SocketInitializer() {
- #ifdef WIN32
- WSACleanup();
- #endif
- }
- };
- static void initialize_socket() {
- static SocketInitializer initialized;
- }
-
-
- static std::string init_hostname() {
- initialize_socket();
- char hostname[1024];
- gethostname(hostname, 1024);
- Socket mySock(nullptr, hostname, 0);
- mySock.initialize();
- auto resolved_hostname = mySock.getHostname();
- return !IsNullOrEmpty(resolved_hostname) ? resolved_hostname : hostname;
- }
-};
-
-} /* namespace io */
-} /* namespace minifi */
-} /* namespace nifi */
-} /* namespace apache */
-} /* namespace org */
-#endif /* LIBMINIFI_INCLUDE_IO_WIN__CLIENTSOCKET_H_ */
diff --git a/libminifi/src/core/PropertyValidation.cpp b/libminifi/src/core/PropertyValidation.cpp
index d476416..177a3c7 100644
--- a/libminifi/src/core/PropertyValidation.cpp
+++ b/libminifi/src/core/PropertyValidation.cpp
@@ -24,6 +24,7 @@ namespace minifi {
namespace core {
std::shared_ptr<PropertyValidator> StandardValidators::VALID = std::make_shared<AlwaysValid>(true, "VALID");
+
StandardValidators::StandardValidators() {
INVALID = std::make_shared<AlwaysValid>(false, "INVALID");
INTEGER_VALIDATOR = std::make_shared<IntegerValidator>("INTEGER_VALIDATOR");
diff --git a/libminifi/src/io/CPPLINT.cfg b/libminifi/src/io/CPPLINT.cfg
new file mode 100644
index 0000000..48cd471
--- /dev/null
+++ b/libminifi/src/io/CPPLINT.cfg
@@ -0,0 +1,3 @@
+set noparent
+filter=-build/include_order,-build/include_alpha
+exclude_files=ClientSocket\.cpp
diff --git a/libminifi/src/io/ClientSocket.cpp b/libminifi/src/io/ClientSocket.cpp
new file mode 100644
index 0000000..cf64928
--- /dev/null
+++ b/libminifi/src/io/ClientSocket.cpp
@@ -0,0 +1,634 @@
+/**
+ *
+ * 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 "io/ClientSocket.h"
+#ifndef WIN32
+#include <netinet/tcp.h>
+#include <sys/types.h>
+#include <netinet/in.h>
+#include <ifaddrs.h>
+#include <unistd.h>
+#else
+#include <WS2tcpip.h>
+#pragma comment(lib, "Ws2_32.lib")
+#endif /* !WIN32 */
+
+#include <memory>
+#include <utility>
+#include <vector>
+#include <cerrno>
+#include <string>
+#include "Exception.h"
+#include <system_error>
+#include <cinttypes>
+#include <Exception.h>
+#include <utils/Deleters.h>
+#include "io/validation.h"
+#include "core/logging/LoggerConfiguration.h"
+#include "utils/GeneralUtils.h"
+
+namespace util = org::apache::nifi::minifi::utils;
+namespace mio = org::apache::nifi::minifi::io;
+
+namespace {
+std::string get_last_getaddrinfo_err_str(int getaddrinfo_result) {
+#ifdef WIN32
+ (void)getaddrinfo_result; // against unused warnings on windows
+ return mio::get_last_socket_error_message();
+#else
+ return gai_strerror(getaddrinfo_result);
+#endif /* WIN32 */
+}
+
+std::string sockaddr_ntop(const sockaddr* const sa) {
+ std::string result;
+ if (sa->sa_family == AF_INET) {
+ sockaddr_in sa_in{};
+ std::memcpy(reinterpret_cast<void*>(&sa_in), sa, sizeof(sockaddr_in));
+ result.resize(INET_ADDRSTRLEN);
+ if (inet_ntop(AF_INET, &sa_in.sin_addr, &result[0], INET_ADDRSTRLEN) == nullptr) {
+ throw minifi::Exception{ minifi::ExceptionType::GENERAL_EXCEPTION, mio::get_last_socket_error_message() };
+ }
+ } else if (sa->sa_family == AF_INET6) {
+ sockaddr_in6 sa_in6{};
+ std::memcpy(reinterpret_cast<void*>(&sa_in6), sa, sizeof(sockaddr_in6));
+ result.resize(INET6_ADDRSTRLEN);
+ if (inet_ntop(AF_INET6, &sa_in6.sin6_addr, &result[0], INET6_ADDRSTRLEN) == nullptr) {
+ throw minifi::Exception{ minifi::ExceptionType::GENERAL_EXCEPTION, mio::get_last_socket_error_message() };
+ }
+ } else {
+ throw minifi::Exception{ minifi::ExceptionType::GENERAL_EXCEPTION, "sockaddr_ntop: unknown address family" };
+ }
+ result.resize(strlen(result.c_str())); // discard remaining null bytes at the end
+ return result;
+}
+
+template<typename T, typename Pred, typename Adv>
+auto find_if_custom_linked_list(T* const list, const Adv advance_func, const Pred predicate) ->
+ typename std::enable_if<std::is_convertible<decltype(advance_func(std::declval<T*>())), T*>::value && std::is_convertible<decltype(predicate(std::declval<T*>())), bool>::value, T*>::type
+{
+ for (T* it = list; it; it = advance_func(it)) {
+ if (predicate(it)) return it;
+ }
+ return nullptr;
+}
+
+#ifndef WIN32
+std::error_code bind_to_local_network_interface(const minifi::io::SocketDescriptor fd, const minifi::io::NetworkInterface& interface) {
+ using ifaddrs_uniq_ptr = std::unique_ptr<ifaddrs, util::ifaddrs_deleter>;
+ const auto if_list_ptr = []() -> ifaddrs_uniq_ptr {
+ ifaddrs *list = nullptr;
+ const auto get_ifa_success = getifaddrs(&list) == 0;
+ assert(get_ifa_success || !list);
+ (void)get_ifa_success; // unused in release builds
+ return ifaddrs_uniq_ptr{ list };
+ }();
+ if (!if_list_ptr) { return { errno, std::generic_category() }; }
+
+ const auto advance_func = [](const ifaddrs *const p) { return p->ifa_next; };
+ const auto predicate = [&interface](const ifaddrs *const item) {
+ return item->ifa_addr && item->ifa_name && (item->ifa_addr->sa_family == AF_INET || item->ifa_addr->sa_family == AF_INET6)
+ && item->ifa_name == interface.getInterface();
+ };
+ const auto *const itemFound = find_if_custom_linked_list(if_list_ptr.get(), advance_func, predicate);
+ if (itemFound == nullptr) { return std::make_error_code(std::errc::no_such_device_or_address); }
+
+ const socklen_t addrlen = itemFound->ifa_addr->sa_family == AF_INET ? sizeof(struct sockaddr_in) : sizeof(struct sockaddr_in6);
+ if (bind(fd, itemFound->ifa_addr, addrlen) != 0) { return { errno, std::generic_category() }; }
+ return {};
+}
+#endif /* !WIN32 */
+
+std::error_code set_non_blocking(const minifi::io::SocketDescriptor fd) noexcept {
+#ifndef WIN32
+ if (fcntl(fd, F_SETFL, O_NONBLOCK) < 0) {
+ return { errno, std::generic_category() };
+ }
+#else
+ u_long iMode = 1;
+ if (ioctlsocket(fd, FIONBIO, &iMode) == SOCKET_ERROR) {
+ return { WSAGetLastError(), std::system_category() };
+ }
+#endif /* !WIN32 */
+ return {};
+}
+} // namespace
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace io {
+
+std::string get_last_socket_error_message() {
+#ifdef WIN32
+ const auto error_code = WSAGetLastError();
+#else
+ const auto error_code = errno;
+#endif /* WIN32 */
+ return std::system_category().message(error_code);
+}
+
+bool valid_socket(const SocketDescriptor fd) noexcept {
+#ifdef WIN32
+ return fd != INVALID_SOCKET && fd >= 0;
+#else
+ return fd >= 0;
+#endif /* WIN32 */
+}
+
+Socket::Socket(const std::shared_ptr<SocketContext>& /*context*/, std::string hostname, const uint16_t port, const uint16_t listeners)
+ : requested_hostname_(std::move(hostname)),
+ port_(port),
+ listeners_(listeners),
+ logger_(logging::LoggerFactory<Socket>::getLogger()) {
+ FD_ZERO(&total_list_);
+ FD_ZERO(&read_fds_);
+ initialize_socket();
+}
+
+Socket::Socket(const std::shared_ptr<SocketContext>& context, std::string hostname, const uint16_t port)
+ : Socket(context, std::move(hostname), port, 0) {
+}
+
+// total_list_ and read_fds_ have to use parentheses for initialization due to CWG 1467
+// http://www.open-std.org/jtc1/sc22/wg21/docs/cwg_defects.html#1467
+// Language defect fix was applied to GCC 5 and Clang 4, but at the time of writing this comment, we support GCC 4.8
+Socket::Socket(Socket &&other) noexcept
+ : requested_hostname_{ std::move(other.requested_hostname_) },
+ canonical_hostname_{ std::move(other.canonical_hostname_) },
+ port_{ other.port_ },
+ is_loopback_only_{ other.is_loopback_only_ },
+ local_network_interface_{ std::move(other.local_network_interface_) },
+ socket_file_descriptor_{ other.socket_file_descriptor_ },
+ total_list_(other.total_list_),
+ read_fds_(other.read_fds_),
+ socket_max_{ other.socket_max_.load() },
+ total_written_{ other.total_written_.load() },
+ total_read_{ other.total_read_.load() },
+ listeners_{ other.listeners_ },
+ nonBlocking_{ other.nonBlocking_ },
+ logger_{ other.logger_ }
+{
+ other = Socket{ {}, {}, {} };
+}
+
+Socket& Socket::operator=(Socket &&other) noexcept {
+ if (&other == this) return *this;
+ requested_hostname_ = util::exchange(other.requested_hostname_, "");
+ canonical_hostname_ = util::exchange(other.canonical_hostname_, "");
+ port_ = util::exchange(other.port_, 0);
+ is_loopback_only_ = util::exchange(other.is_loopback_only_, false);
+ local_network_interface_ = util::exchange(other.local_network_interface_, {});
+ socket_file_descriptor_ = util::exchange(other.socket_file_descriptor_, INVALID_SOCKET);
+ total_list_ = other.total_list_;
+ FD_ZERO(&other.total_list_);
+ read_fds_ = other.read_fds_;
+ FD_ZERO(&other.read_fds_);
+ socket_max_.exchange(other.socket_max_);
+ other.socket_max_.exchange(0);
+ total_written_.exchange(other.total_written_);
+ other.total_written_.exchange(0);
+ total_read_.exchange(other.total_read_);
+ other.total_read_.exchange(0);
+ listeners_ = util::exchange(other.listeners_, 0);
+ nonBlocking_ = util::exchange(other.nonBlocking_, false);
+ logger_ = other.logger_;
+ return *this;
+}
+
+Socket::~Socket() {
+ Socket::closeStream();
+}
+
+void Socket::closeStream() {
+ if (valid_socket(socket_file_descriptor_)) {
+ logging::LOG_DEBUG(logger_) << "Closing " << socket_file_descriptor_;
+#ifdef WIN32
+ closesocket(socket_file_descriptor_);
+#else
+ close(socket_file_descriptor_);
+#endif
+ socket_file_descriptor_ = INVALID_SOCKET;
+ }
+ if (total_written_ > 0) {
+ local_network_interface_.log_write(total_written_);
+ total_written_ = 0;
+ }
+ if (total_read_ > 0) {
+ local_network_interface_.log_read(total_read_);
+ total_read_ = 0;
+ }
+}
+
+void Socket::setNonBlocking() {
+ if (listeners_ <= 0) {
+ nonBlocking_ = true;
+ }
+}
+
+int8_t Socket::createConnection(const addrinfo* const destination_addresses) {
+ for (const auto *current_addr = destination_addresses; current_addr; current_addr = current_addr->ai_next) {
+ if (!valid_socket(socket_file_descriptor_ = socket(current_addr->ai_family, current_addr->ai_socktype, current_addr->ai_protocol))) {
+ logger_->log_warn("socket: %s", get_last_socket_error_message());
+ continue;
+ }
+ setSocketOptions(socket_file_descriptor_);
+
+ if (listeners_ > 0) {
+ // server socket
+ const auto bind_result = bind(socket_file_descriptor_, current_addr->ai_addr, current_addr->ai_addrlen);
+ if (bind_result == SOCKET_ERROR) {
+ logger_->log_warn("bind: %s", get_last_socket_error_message());
+ closeStream();
+ continue;
+ }
+
+ const auto listen_result = listen(socket_file_descriptor_, listeners_);
+ if (listen_result == SOCKET_ERROR) {
+ logger_->log_warn("listen: %s", get_last_socket_error_message());
+ closeStream();
+ continue;
+ }
+
+ logger_->log_info("Listening on %s:%" PRIu16 " with backlog %" PRIu16, sockaddr_ntop(current_addr->ai_addr), port_, listeners_);
+ } else {
+ // client socket
+#ifndef WIN32
+ if (!local_network_interface_.getInterface().empty()) {
+ const auto err = bind_to_local_network_interface(socket_file_descriptor_, local_network_interface_);
+ if (err) logger_->log_info("Bind to interface %s failed %s", local_network_interface_.getInterface(), err.message());
+ else logger_->log_info("Bind to interface %s", local_network_interface_.getInterface());
+ }
+#endif /* !WIN32 */
+
+ const auto connect_result = connect(socket_file_descriptor_, current_addr->ai_addr, current_addr->ai_addrlen);
+ if (connect_result == SOCKET_ERROR) {
+ logger_->log_warn("Couldn't connect to %s:%" PRIu16 ": %s", sockaddr_ntop(current_addr->ai_addr), port_, get_last_socket_error_message());
+ closeStream();
+ continue;
+ }
+
+ logger_->log_info("Connected to %s:%" PRIu16, sockaddr_ntop(current_addr->ai_addr), port_);
+ }
+
+ FD_SET(socket_file_descriptor_, &total_list_);
+ socket_max_ = socket_file_descriptor_;
+ return 0;
+ }
+ return -1;
+}
+
+int8_t Socket::createConnection(const addrinfo *, ip4addr &addr) {
+ if (!valid_socket(socket_file_descriptor_ = socket(AF_INET, SOCK_STREAM, 0))) {
+ logger_->log_error("error while connecting to server socket");
+ return -1;
+ }
+
+ setSocketOptions(socket_file_descriptor_);
+
+ if (listeners_ > 0) {
+ // server socket
+ sockaddr_in sa{};
+ memset(&sa, 0, sizeof(struct sockaddr_in));
+ sa.sin_family = AF_INET;
+ sa.sin_port = htons(port_);
+ sa.sin_addr.s_addr = htonl(is_loopback_only_ ? INADDR_LOOPBACK : INADDR_ANY);
+ if (bind(socket_file_descriptor_, reinterpret_cast<const sockaddr*>(&sa), sizeof(struct sockaddr_in)) == SOCKET_ERROR) {
+ logger_->log_error("Could not bind to socket, reason %s", get_last_socket_error_message());
+ return -1;
+ }
+
+ if (listen(socket_file_descriptor_, listeners_) == -1) {
+ return -1;
+ }
+ logger_->log_debug("Created connection with %d listeners", listeners_);
+ } else {
+ // client socket
+#ifndef WIN32
+ if (!local_network_interface_.getInterface().empty()) {
+ const auto err = bind_to_local_network_interface(socket_file_descriptor_, local_network_interface_);
+ if (err) logger_->log_info("Bind to interface %s failed %s", local_network_interface_.getInterface(), err.message());
+ else logger_->log_info("Bind to interface %s", local_network_interface_.getInterface());
+ }
+#endif /* !WIN32 */
+ sockaddr_in sa_loc{};
+ memset(&sa_loc, 0x00, sizeof(sa_loc));
+ sa_loc.sin_family = AF_INET;
+ sa_loc.sin_port = htons(port_);
+ // use any address if you are connecting to the local machine for testing
+ // otherwise we must use the requested hostname
+ if (IsNullOrEmpty(requested_hostname_) || requested_hostname_ == "localhost") {
+ sa_loc.sin_addr.s_addr = htonl(is_loopback_only_ ? INADDR_LOOPBACK : INADDR_ANY);
+ } else {
+#ifdef WIN32
+ sa_loc.sin_addr.s_addr = addr.s_addr;
+ }
+ if (connect(socket_file_descriptor_, reinterpret_cast<const sockaddr*>(&sa_loc), sizeof(sockaddr_in)) == SOCKET_ERROR) {
+ int err = WSAGetLastError();
+ if (err == WSAEADDRNOTAVAIL) {
+ logger_->log_error("invalid or unknown IP");
+ } else if (err == WSAECONNREFUSED) {
+ logger_->log_error("Connection refused");
+ } else {
+ logger_->log_error("Unknown error");
+ }
+#else
+ sa_loc.sin_addr.s_addr = addr;
+ }
+ if (connect(socket_file_descriptor_, reinterpret_cast<const sockaddr *>(&sa_loc), sizeof(sockaddr_in)) < 0) {
+#endif /* WIN32 */
+ closeStream();
+ return -1;
+ }
+ }
+
+ // add the listener to the total set
+ FD_SET(socket_file_descriptor_, &total_list_);
+ socket_max_ = socket_file_descriptor_;
+ logger_->log_debug("Created connection with file descriptor %d", socket_file_descriptor_);
+ return 0;
+}
+
+int16_t Socket::initialize() {
+ addrinfo hints{};
+ memset(&hints, 0, sizeof hints); // make sure the struct is empty
+ hints.ai_family = AF_UNSPEC;
+ hints.ai_socktype = SOCK_STREAM;
+ hints.ai_flags = AI_CANONNAME;
+ if (listeners_ > 0 && !is_loopback_only_)
+ hints.ai_flags = AI_PASSIVE;
+ hints.ai_protocol = 0; /* any protocol */
+
+ const char* const gai_node = [this]() -> const char* {
+ if (is_loopback_only_) return "localhost";
+ if (!is_loopback_only_ && listeners_ > 0) return nullptr; // all non-localhost server sockets listen on wildcard address
+ if (!requested_hostname_.empty()) return requested_hostname_.c_str();
+ return nullptr;
+ }();
+ const auto gai_service = std::to_string(port_);
+ addrinfo* getaddrinfo_result = nullptr;
+ const int errcode = getaddrinfo(gai_node, gai_service.c_str(), &hints, &getaddrinfo_result);
+ const std::unique_ptr<addrinfo, util::addrinfo_deleter> addr_info{ getaddrinfo_result };
+ getaddrinfo_result = nullptr;
+ if (errcode != 0) {
+ logger_->log_error("getaddrinfo: %s", get_last_getaddrinfo_err_str(errcode));
+ return -1;
+ }
+ socket_file_descriptor_ = INVALID_SOCKET;
+
+ // AI_CANONNAME always sets ai_canonname of the first addrinfo structure
+ canonical_hostname_ = !IsNullOrEmpty(addr_info->ai_canonname) ? addr_info->ai_canonname : requested_hostname_;
+
+ const auto conn_result = port_ > 0 ? createConnection(addr_info.get()) : -1;
+ if (conn_result == 0 && nonBlocking_) {
+ // Put the socket in non-blocking mode:
+ const auto err = set_non_blocking(socket_file_descriptor_);
+ if (err) logger_->log_info("Couldn't make socket non-blocking: %s", err.message());
+ else logger_->log_debug("Successfully applied O_NONBLOCK to fd");
+ }
+ return conn_result;
+}
+
+int16_t Socket::select_descriptor(const uint16_t msec) {
+ if (listeners_ == 0) {
+ return socket_file_descriptor_;
+ }
+
+ struct timeval tv{};
+
+ read_fds_ = total_list_;
+
+ tv.tv_sec = msec / 1000;
+ tv.tv_usec = (msec % 1000) * 1000;
+
+ std::lock_guard<std::recursive_mutex> guard(selection_mutex_);
+
+ if (msec > 0)
+ select(socket_max_ + 1, &read_fds_, nullptr, nullptr, &tv);
+ else
+ select(socket_max_ + 1, &read_fds_, nullptr, nullptr, nullptr);
+
+ for (int i = 0; i <= socket_max_; i++) {
+ if (FD_ISSET(i, &read_fds_)) {
+ if (i == socket_file_descriptor_) {
+ if (listeners_ > 0) {
+ struct sockaddr_storage remoteaddr; // client address
+ socklen_t addrlen = sizeof remoteaddr;
+ int newfd = accept(socket_file_descriptor_, (struct sockaddr *) &remoteaddr, &addrlen);
+ FD_SET(newfd, &total_list_); // add to master set
+ if (newfd > socket_max_) { // keep track of the max
+ socket_max_ = newfd;
+ }
+ return newfd;
+ } else {
+ return socket_file_descriptor_;
+ }
+ // we have a new connection
+ } else {
+ // data to be received on i
+ return i;
+ }
+ }
+ }
+
+ logger_->log_debug("Could not find a suitable file descriptor or select timed out");
+
+ return -1;
+}
+
+int16_t Socket::setSocketOptions(const SocketDescriptor sock) {
+ int opt = 1;
+#ifndef WIN32
+#ifndef __MACH__
+ if (setsockopt(sock, SOL_TCP, TCP_NODELAY, static_cast<void*>(&opt), sizeof(opt)) < 0) {
+ logger_->log_error("setsockopt() TCP_NODELAY failed");
+ close(sock);
+ return -1;
+ }
+ if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, reinterpret_cast<char*>(&opt), sizeof(opt)) < 0) {
+ logger_->log_error("setsockopt() SO_REUSEADDR failed");
+ close(sock);
+ return -1;
+ }
+
+ int sndsize = 256 * 1024;
+ if (setsockopt(sock, SOL_SOCKET, SO_SNDBUF, reinterpret_cast<char *>(&sndsize), sizeof(sndsize)) < 0) {
+ logger_->log_error("setsockopt() SO_SNDBUF failed");
+ close(sock);
+ return -1;
+ }
+
+#else
+ if (listeners_ > 0) {
+ // lose the pesky "address already in use" error message
+ if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, reinterpret_cast<char *>(&opt), sizeof(opt)) < 0) {
+ logger_->log_error("setsockopt() SO_REUSEADDR failed");
+ close(sock);
+ return -1;
+ }
+ }
+#endif /* !__MACH__ */
+#endif /* !WIN32 */
+ return 0;
+}
+
+std::string Socket::getHostname() const {
+ return canonical_hostname_;
+}
+
+int Socket::writeData(std::vector<uint8_t> &buf, int buflen) {
+ if (buflen < 0) {
+ throw minifi::Exception{ExceptionType::GENERAL_EXCEPTION, "negative buflen"};
+ }
+
+ if (buf.size() < static_cast<size_t>(buflen))
+ return -1;
+ return writeData(buf.data(), buflen);
+}
+
+// data stream overrides
+
+int Socket::writeData(uint8_t *value, int size) {
+ int ret = 0, bytes = 0;
+
+ int fd = select_descriptor(1000);
+ if (fd < 0) { return -1; }
+ while (bytes < size) {
+ ret = send(fd, reinterpret_cast<const char*>(value) + bytes, size - bytes, 0);
+ // check for errors
+ if (ret <= 0) {
+ close(fd);
+ logger_->log_error("Could not send to %d, error: %s", fd, get_last_socket_error_message());
+ return ret;
+ }
+ bytes += ret;
+ }
+
+ if (ret)
+ logger_->log_trace("Send data size %d over socket %d", size, fd);
+ total_written_ += bytes;
+ return bytes;
+}
+
+template<typename T>
+inline std::vector<uint8_t> Socket::readBuffer(const T& t) {
+ std::vector<uint8_t> buf;
+ buf.resize(sizeof t);
+ readData(reinterpret_cast<uint8_t *>(&buf[0]), sizeof(t));
+ return buf;
+}
+
+int Socket::write(uint64_t base_value, bool is_little_endian) {
+ return Serializable::write(base_value, this, is_little_endian);
+}
+
+int Socket::write(uint32_t base_value, bool is_little_endian) {
+ return Serializable::write(base_value, this, is_little_endian);
+}
+
+int Socket::write(uint16_t base_value, bool is_little_endian) {
+ return Serializable::write(base_value, this, is_little_endian);
+}
+
+int Socket::read(uint64_t &value, bool is_little_endian) {
+ auto buf = readBuffer(value);
+
+ if (is_little_endian) {
+ value = ((uint64_t) buf[0] << 56) | ((uint64_t) (buf[1] & 255) << 48) | ((uint64_t) (buf[2] & 255) << 40) | ((uint64_t) (buf[3] & 255) << 32) | ((uint64_t) (buf[4] & 255) << 24)
+ | ((uint64_t) (buf[5] & 255) << 16) | ((uint64_t) (buf[6] & 255) << 8) | ((uint64_t) (buf[7] & 255) << 0);
+ } else {
+ value = ((uint64_t) buf[0] << 0) | ((uint64_t) (buf[1] & 255) << 8) | ((uint64_t) (buf[2] & 255) << 16) | ((uint64_t) (buf[3] & 255) << 24) | ((uint64_t) (buf[4] & 255) << 32)
+ | ((uint64_t) (buf[5] & 255) << 40) | ((uint64_t) (buf[6] & 255) << 48) | ((uint64_t) (buf[7] & 255) << 56);
+ }
+ return sizeof(value);
+}
+
+int Socket::read(uint32_t &value, bool is_little_endian) {
+ auto buf = readBuffer(value);
+
+ if (is_little_endian) {
+ value = (buf[0] << 24) | (buf[1] << 16) | (buf[2] << 8) | buf[3];
+ } else {
+ value = buf[0] | buf[1] << 8 | buf[2] << 16 | buf[3] << 24;
+ }
+ return sizeof(value);
+}
+
+int Socket::read(uint16_t &value, bool is_little_endian) {
+ auto buf = readBuffer(value);
+
+ if (is_little_endian) {
+ value = (buf[0] << 8) | buf[1];
+ } else {
+ value = buf[0] | buf[1] << 8;
+ }
+ return sizeof(value);
+}
+
+int Socket::readData(std::vector<uint8_t> &buf, int buflen, bool retrieve_all_bytes) {
+ if (buflen < 0) {
+ throw minifi::Exception{ExceptionType::GENERAL_EXCEPTION, "negative buflen"};
+ }
+
+ if (buf.size() < static_cast<size_t>(buflen)) {
+ buf.resize(buflen);
+ }
+ return readData(buf.data(), buflen, retrieve_all_bytes);
+}
+
+int Socket::readData(uint8_t *buf, int buflen, bool retrieve_all_bytes) {
+ int32_t total_read = 0;
+ while (buflen) {
+ int16_t fd = select_descriptor(1000);
+ if (fd < 0) {
+ if (listeners_ <= 0) {
+ logger_->log_debug("fd %d close %i", fd, buflen);
+ close(socket_file_descriptor_);
+ }
+ return -1;
+ }
+ int bytes_read = recv(fd, reinterpret_cast<char*>(buf), buflen, 0);
+ logger_->log_trace("Recv call %d", bytes_read);
+ if (bytes_read <= 0) {
+ if (bytes_read == 0) {
+ logger_->log_debug("Other side hung up on %d", fd);
+ } else {
+ if (errno == EAGAIN || errno == EWOULDBLOCK) {
+ // continue
+ return -2;
+ }
+ logger_->log_error("Could not recv on %d ( port %d), error: %s", fd, port_, strerror(errno));
+ }
+ return -1;
+ }
+ buflen -= bytes_read;
+ buf += bytes_read;
+ total_read += bytes_read;
+ if (!retrieve_all_bytes) {
+ break;
+ }
+ }
+ total_read_ += total_read;
+ return total_read;
+}
+
+} /* namespace io */
+} /* namespace minifi */
+} /* namespace nifi */
+} /* namespace apache */
+} /* namespace org */
diff --git a/libminifi/src/io/ServerSocket.cpp b/libminifi/src/io/ServerSocket.cpp
index cf79e0c..bc24f9a 100644
--- a/libminifi/src/io/ServerSocket.cpp
+++ b/libminifi/src/io/ServerSocket.cpp
@@ -18,21 +18,17 @@
#include "io/ServerSocket.h"
#include "io/DescriptorStream.h"
-#include <sys/types.h>
#ifndef WIN32
-#include <netinet/tcp.h>
-#include <netdb.h>
-#include <sys/socket.h>
+#include <sys/types.h>
#include <netinet/in.h>
#include <arpa/inet.h>
+#include <fcntl.h>
#include <unistd.h>
-#endif
-#include <cstdio>
+#else
+#pragma comment(lib, "Ws2_32.lib")
+#endif /* !WIN32 */
#include <memory>
#include <utility>
-#include <vector>
-#include <cerrno>
-#include <iostream>
#include <string>
#include "io/validation.h"
#include "core/logging/LoggerConfiguration.h"
@@ -70,7 +66,7 @@ void ServerSocket::registerCallback(std::function<bool()> accept_function, std::
}
}
};
- server_read_thread_ = std::thread(fx, accept_function, handler);
+ server_read_thread_ = std::thread(fx, std::move(accept_function), std::move(handler));
}
void ServerSocket::close_fd(int fd) {
diff --git a/libminifi/src/io/StreamFactory.cpp b/libminifi/src/io/StreamFactory.cpp
index bb021ae..457c6d3 100644
--- a/libminifi/src/io/StreamFactory.cpp
+++ b/libminifi/src/io/StreamFactory.cpp
@@ -20,6 +20,7 @@
#include <memory>
#include <mutex>
#include <string>
+#include <utils/GeneralUtils.h>
#ifdef OPENSSL_SUPPORT
#include "io/tls/TLSSocket.h"
@@ -38,21 +39,11 @@ namespace io {
template<typename T, typename V>
class SocketCreator : public AbstractStreamFactory {
- template<bool cond, typename U>
- using TypeCheck = typename std::enable_if< cond, U >::type;
-
- template<bool cond, typename Q>
- using ContextTypeCheck = typename std::enable_if< cond, Q >::type;
-
public:
template<typename Q = V>
- ContextTypeCheck<true, std::shared_ptr<Q>> create(const std::shared_ptr<Configure> &configure) {
+ std::shared_ptr<Q> create(const std::shared_ptr<Configure> &configure) {
return std::make_shared<V>(configure);
}
- template<typename Q = V>
- ContextTypeCheck<false, std::shared_ptr<Q>> create(const std::shared_ptr<Configure> &configure) {
- return std::make_shared<SocketContext>(configure);
- }
SocketCreator<T, V>(const std::shared_ptr<Configure> &configuration)
: configuration_(configuration) {
@@ -60,13 +51,9 @@ class SocketCreator : public AbstractStreamFactory {
}
template<typename U = T>
- TypeCheck<true, U> *create(const std::string &host, const uint16_t port) {
+ U* create(const std::string &host, const uint16_t port) {
return new T(context_, host, port);
}
- template<typename U = T>
- TypeCheck<false, U> *create(const std::string &host, const uint16_t port) {
- return new Socket(context_, host, port);
- }
std::unique_ptr<Socket> createSocket(const std::string &host, const uint16_t port) {
T *socket = create(host, port);
@@ -76,15 +63,11 @@ class SocketCreator : public AbstractStreamFactory {
std::unique_ptr<Socket> createSecureSocket(const std::string &host, const uint16_t port, const std::shared_ptr<minifi::controllers::SSLContextService> &ssl_service) {
#ifdef OPENSSL_SUPPORT
if (ssl_service != nullptr) {
- auto ptr = std::make_shared<TLSContext>(configuration_, ssl_service);
- TLSSocket *socket = new TLSSocket(ptr, host, port);
- return std::unique_ptr<Socket>(socket);
- } else {
- return nullptr;
+ auto context = std::make_shared<TLSContext>(configuration_, ssl_service);
+ return utils::make_unique<TLSSocket>(context, host, port);
}
-#else
+#endif /* OPENSSL_SUPPORT */
return nullptr;
-#endif
}
private:
diff --git a/libminifi/src/io/posix/ClientSocket.cpp b/libminifi/src/io/posix/ClientSocket.cpp
deleted file mode 100644
index d65d68d..0000000
--- a/libminifi/src/io/posix/ClientSocket.cpp
+++ /dev/null
@@ -1,495 +0,0 @@
-/**
- *
- * 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 "io/ClientSocket.h"
-#include <netinet/tcp.h>
-#include <sys/types.h>
-#include <netdb.h>
-#include <sys/socket.h>
-#include <netinet/in.h>
-#include <arpa/inet.h>
-#include <sys/ioctl.h>
-#include <net/if.h>
-#include <ifaddrs.h>
-#include <unistd.h>
-#include <cstdio>
-#include <memory>
-#include <utility>
-#include <vector>
-#include <cerrno>
-#include <iostream>
-#include <string>
-#include "Exception.h"
-#include "io/validation.h"
-#include "core/logging/LoggerConfiguration.h"
-
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-namespace io {
-
-Socket::Socket(const std::shared_ptr<SocketContext>& /*context*/, const std::string &hostname, const uint16_t port, const uint16_t listeners = -1)
- : requested_hostname_(hostname),
- port_(port),
- addr_info_(0),
- socket_file_descriptor_(-1),
- socket_max_(0),
- total_written_(0),
- total_read_(0),
- is_loopback_only_(false),
- listeners_(listeners),
- canonical_hostname_(""),
- nonBlocking_(false),
- logger_(logging::LoggerFactory<Socket>::getLogger()) {
- FD_ZERO(&total_list_);
- FD_ZERO(&read_fds_);
-}
-
-Socket::Socket(const std::shared_ptr<SocketContext>& context, const std::string &hostname, const uint16_t port)
- : Socket(context, hostname, port, 0) {
-}
-
-Socket::Socket(const Socket &&other)
- : requested_hostname_(std::move(other.requested_hostname_)),
- port_(std::move(other.port_)),
- is_loopback_only_(false),
- addr_info_(std::move(other.addr_info_)),
- socket_file_descriptor_(other.socket_file_descriptor_),
- socket_max_(other.socket_max_.load()),
- listeners_(other.listeners_),
- total_list_(other.total_list_),
- read_fds_(other.read_fds_),
- canonical_hostname_(std::move(other.canonical_hostname_)),
- nonBlocking_(false),
- logger_(std::move(other.logger_)) {
- total_written_ = other.total_written_.load();
- total_read_ = other.total_read_.load();
-}
-
-Socket::~Socket() {
- closeStream();
-}
-
-void Socket::closeStream() {
- if (0 != addr_info_) {
- freeaddrinfo(addr_info_);
- addr_info_ = 0;
- }
- if (socket_file_descriptor_ >= 0) {
- logging::LOG_DEBUG(logger_) << "Closing " << socket_file_descriptor_;
- close(socket_file_descriptor_);
- socket_file_descriptor_ = -1;
- }
- if (total_written_ > 0) {
- local_network_interface_.log_write(total_written_);
- total_written_ = 0;
- }
- if (total_read_ > 0) {
- local_network_interface_.log_read(total_read_);
- total_read_ = 0;
- }
-}
-
-void Socket::setNonBlocking() {
- if (listeners_ <= 0) {
- nonBlocking_ = true;
- }
-}
-
-int8_t Socket::createConnection(const addrinfo *p, in_addr_t &addr) {
- if ((socket_file_descriptor_ = socket(p->ai_family, p->ai_socktype, p->ai_protocol)) == -1) {
- logger_->log_error("error while connecting to server socket");
- return -1;
- }
-
- setSocketOptions(socket_file_descriptor_);
-
- if (listeners_ <= 0 && !local_network_interface_.getInterface().empty()) {
- // bind to local network interface
- ifaddrs* list = NULL;
- ifaddrs* item = NULL;
- ifaddrs* itemFound = NULL;
- int result = getifaddrs(&list);
- if (result == 0) {
- item = list;
- while (item) {
- if ((item->ifa_addr != NULL) && (item->ifa_name != NULL) && (AF_INET == item->ifa_addr->sa_family)) {
- if (strcmp(item->ifa_name, local_network_interface_.getInterface().c_str()) == 0) {
- itemFound = item;
- break;
- }
- }
- item = item->ifa_next;
- }
-
- if (itemFound != NULL) {
- result = bind(socket_file_descriptor_, itemFound->ifa_addr, sizeof(struct sockaddr_in));
- if (result < 0)
- logger_->log_info("Bind to interface %s failed %s", local_network_interface_.getInterface(), strerror(errno));
- else
- logger_->log_info("Bind to interface %s", local_network_interface_.getInterface());
- }
- freeifaddrs(list);
- }
- }
-
- if (listeners_ > 0) {
- struct sockaddr_in *sa_loc = (struct sockaddr_in*) p->ai_addr;
- sa_loc->sin_family = AF_INET;
- sa_loc->sin_port = htons(port_);
- if (is_loopback_only_) {
- sa_loc->sin_addr.s_addr = htonl(INADDR_LOOPBACK);
- } else {
- sa_loc->sin_addr.s_addr = htonl(INADDR_ANY);
- }
- if (bind(socket_file_descriptor_, p->ai_addr, p->ai_addrlen) == -1) {
- logger_->log_error("Could not bind to socket, reason %s", strerror(errno));
- return -1;
- }
- }
- {
- if (listeners_ <= 0) {
- struct sockaddr_in *sa_loc = (struct sockaddr_in*) p->ai_addr;
- sa_loc->sin_family = AF_INET;
- sa_loc->sin_port = htons(port_);
- // use any address if you are connecting to the local machine for testing
- // otherwise we must use the requested hostname
- if (IsNullOrEmpty(requested_hostname_) || requested_hostname_ == "localhost") {
- if (is_loopback_only_) {
- sa_loc->sin_addr.s_addr = htonl(INADDR_LOOPBACK);
- } else {
- sa_loc->sin_addr.s_addr = htonl(INADDR_ANY);
- }
- } else {
- sa_loc->sin_addr.s_addr = addr;
- }
- if (connect(socket_file_descriptor_, p->ai_addr, p->ai_addrlen) == -1) {
- close(socket_file_descriptor_);
- socket_file_descriptor_ = -1;
- return -1;
- }
- }
- }
-
- // listen
- if (listeners_ > 0) {
- if (listen(socket_file_descriptor_, listeners_) == -1) {
- return -1;
- } else {
- logger_->log_debug("Created connection with %d listeners", listeners_);
- }
- }
- // add the listener to the total set
- FD_SET(socket_file_descriptor_, &total_list_);
- socket_max_ = socket_file_descriptor_;
- logger_->log_debug("Created connection with file descriptor %d", socket_file_descriptor_);
- return 0;
-}
-
-int16_t Socket::initialize() {
- addrinfo hints = { sizeof(addrinfo) };
- memset(&hints, 0, sizeof hints); // make sure the struct is empty
- hints.ai_family = AF_UNSPEC;
- hints.ai_socktype = SOCK_STREAM;
- hints.ai_flags = AI_CANONNAME;
- if (listeners_ > 0)
- hints.ai_flags |= AI_PASSIVE;
- hints.ai_protocol = 0; /* any protocol */
-
- int errcode = getaddrinfo(requested_hostname_.c_str(), 0, &hints, &addr_info_);
-
- if (errcode != 0) {
- logger_->log_error("Saw error during getaddrinfo, error: %s", strerror(errno));
- return -1;
- }
-
- socket_file_descriptor_ = -1;
-
- in_addr_t addr;
- struct hostent *h;
-#ifdef __MACH__
- h = gethostbyname(requested_hostname_.c_str());
-#else
- const char *host;
-
- host = requested_hostname_.c_str();
- char buf[1024];
- struct hostent he;
- int hh_errno;
- gethostbyname_r(host, &he, buf, sizeof(buf), &h, &hh_errno);
-#endif
- if (h == nullptr) {
- logger_->log_error("hostname not defined for %s", requested_hostname_);
- return -1;
- }
- memcpy(reinterpret_cast<char*>(&addr), h->h_addr_list[0], h->h_length);
-
- auto p = addr_info_;
- for (; p != NULL; p = p->ai_next) {
- if (IsNullOrEmpty(canonical_hostname_)) {
- if (!IsNullOrEmpty(p) && !IsNullOrEmpty(p->ai_canonname))
- canonical_hostname_ = p->ai_canonname;
- }
- // we've successfully connected
- if (port_ > 0 && createConnection(p, addr) >= 0) {
- // Put the socket in non-blocking mode:
- if (nonBlocking_) {
- if (fcntl(socket_file_descriptor_, F_SETFL, O_NONBLOCK) < 0) {
- // handle error
- logger_->log_error("Could not create non blocking to socket", strerror(errno));
- } else {
- logger_->log_debug("Successfully applied O_NONBLOCK to fd");
- }
- }
- logger_->log_debug("Successfully created connection");
- return 0;
- break;
- }
- }
-
- logger_->log_debug("Could not find device for our connection");
- return -1;
-}
-
-int16_t Socket::select_descriptor(const uint16_t msec) {
- if (listeners_ == 0) {
- return socket_file_descriptor_;
- }
-
- struct timeval tv;
-
- read_fds_ = total_list_;
-
- tv.tv_sec = msec / 1000;
- tv.tv_usec = (msec % 1000) * 1000;
-
- std::lock_guard<std::recursive_mutex> guard(selection_mutex_);
-
- if (msec > 0)
- select(socket_max_ + 1, &read_fds_, NULL, NULL, &tv);
- else
- select(socket_max_ + 1, &read_fds_, NULL, NULL, NULL);
-
- for (int i = 0; i <= socket_max_; i++) {
- if (FD_ISSET(i, &read_fds_)) {
- if (i == socket_file_descriptor_) {
- if (listeners_ > 0) {
- struct sockaddr_storage remoteaddr; // client address
- socklen_t addrlen = sizeof remoteaddr;
- int newfd = accept(socket_file_descriptor_, (struct sockaddr *) &remoteaddr, &addrlen);
- FD_SET(newfd, &total_list_); // add to master set
- if (newfd > socket_max_) { // keep track of the max
- socket_max_ = newfd;
- }
- return newfd;
-
- } else {
- return socket_file_descriptor_;
- }
- // we have a new connection
- } else {
- // data to be received on i
- return i;
- }
- }
- }
-
- logger_->log_debug("Could not find a suitable file descriptor or select timed out");
-
- return -1;
-}
-
-int16_t Socket::setSocketOptions(const int sock) {
- int opt = 1;
-#ifndef __MACH__
- if (setsockopt(sock, SOL_TCP, TCP_NODELAY, static_cast<void*>(&opt), sizeof(opt)) < 0) {
- logger_->log_error("setsockopt() TCP_NODELAY failed");
- close(sock);
- return -1;
- }
- if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, reinterpret_cast<char*>(&opt), sizeof(opt)) < 0) {
- logger_->log_error("setsockopt() SO_REUSEADDR failed");
- close(sock);
- return -1;
- }
-
- int sndsize = 256 * 1024;
- if (setsockopt(sock, SOL_SOCKET, SO_SNDBUF, reinterpret_cast<char *>(&sndsize), sizeof(sndsize)) < 0) {
- logger_->log_error("setsockopt() SO_SNDBUF failed");
- close(sock);
- return -1;
- }
-
-#else
- if (listeners_ > 0) {
- // lose the pesky "address already in use" error message
- if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, reinterpret_cast<char *>(&opt), sizeof(opt)) < 0) {
- logger_->log_error("setsockopt() SO_REUSEADDR failed");
- close(sock);
- return -1;
- }
- }
-#endif
- return 0;
-}
-
-std::string Socket::getHostname() const {
- return canonical_hostname_;
-}
-
-int Socket::writeData(std::vector<uint8_t> &buf, int buflen) {
- if (buflen < 0) {
- throw minifi::Exception{ExceptionType::GENERAL_EXCEPTION, "negative buflen"};
- }
-
- if (buf.size() < static_cast<size_t>(buflen))
- return -1;
-
- return writeData(buf.data(), buflen);
-}
-
-// data stream overrides
-
-int Socket::writeData(uint8_t *value, int size) {
- int ret = 0, bytes = 0;
-
- int fd = select_descriptor(1000);
- while (bytes < size) {
- ret = send(fd, value + bytes, size - bytes, 0);
- // check for errors
- if (ret <= 0) {
- close(fd);
- logger_->log_error("Could not send to %d, error: %s", fd, strerror(errno));
- return ret;
- }
- bytes += ret;
- }
-
- if (ret)
- logger_->log_trace("Send data size %d over socket %d", size, fd);
- total_written_ += bytes;
- return bytes;
-}
-
-template<typename T>
-inline std::vector<uint8_t> Socket::readBuffer(const T& t) {
- std::vector<uint8_t> buf;
- buf.resize(sizeof t);
- readData(reinterpret_cast<uint8_t *>(&buf[0]), sizeof(t));
- return buf;
-}
-
-int Socket::write(uint64_t base_value, bool is_little_endian) {
- return Serializable::write(base_value, this, is_little_endian);
-}
-
-int Socket::write(uint32_t base_value, bool is_little_endian) {
- return Serializable::write(base_value, this, is_little_endian);
-}
-
-int Socket::write(uint16_t base_value, bool is_little_endian) {
- return Serializable::write(base_value, this, is_little_endian);
-}
-
-int Socket::read(uint64_t &value, bool is_little_endian) {
- auto buf = readBuffer(value);
-
- if (is_little_endian) {
- value = ((uint64_t) buf[0] << 56) | ((uint64_t) (buf[1] & 255) << 48) | ((uint64_t) (buf[2] & 255) << 40) | ((uint64_t) (buf[3] & 255) << 32) | ((uint64_t) (buf[4] & 255) << 24)
- | ((uint64_t) (buf[5] & 255) << 16) | ((uint64_t) (buf[6] & 255) << 8) | ((uint64_t) (buf[7] & 255) << 0);
- } else {
- value = ((uint64_t) buf[0] << 0) | ((uint64_t) (buf[1] & 255) << 8) | ((uint64_t) (buf[2] & 255) << 16) | ((uint64_t) (buf[3] & 255) << 24) | ((uint64_t) (buf[4] & 255) << 32)
- | ((uint64_t) (buf[5] & 255) << 40) | ((uint64_t) (buf[6] & 255) << 48) | ((uint64_t) (buf[7] & 255) << 56);
- }
- return sizeof(value);
-}
-
-int Socket::read(uint32_t &value, bool is_little_endian) {
- auto buf = readBuffer(value);
-
- if (is_little_endian) {
- value = (buf[0] << 24) | (buf[1] << 16) | (buf[2] << 8) | buf[3];
- } else {
- value = buf[0] | buf[1] << 8 | buf[2] << 16 | buf[3] << 24;
- }
- return sizeof(value);
-}
-
-int Socket::read(uint16_t &value, bool is_little_endian) {
- auto buf = readBuffer(value);
-
- if (is_little_endian) {
- value = (buf[0] << 8) | buf[1];
- } else {
- value = buf[0] | buf[1] << 8;
- }
- return sizeof(value);
-}
-
-int Socket::readData(std::vector<uint8_t> &buf, int buflen, bool retrieve_all_bytes) {
- if (buflen < 0) {
- throw minifi::Exception{ExceptionType::GENERAL_EXCEPTION, "negative buflen"};
- }
-
- if (buf.size() < static_cast<size_t>(buflen)) {
- buf.resize(buflen);
- }
- return readData(buf.data(), buflen, retrieve_all_bytes);
-}
-
-int Socket::readData(uint8_t *buf, int buflen, bool retrieve_all_bytes) {
- int32_t total_read = 0;
- while (buflen) {
- int16_t fd = select_descriptor(1000);
- if (fd < 0) {
- if (listeners_ <= 0) {
- logger_->log_debug("fd %d close %i", fd, buflen);
- close(socket_file_descriptor_);
- }
- return -1;
- }
- int bytes_read = recv(fd, buf, buflen, 0);
- logger_->log_trace("Recv call %d", bytes_read);
- if (bytes_read <= 0) {
- if (bytes_read == 0) {
- logger_->log_debug("Other side hung up on %d", fd);
- } else {
- if (errno == EAGAIN || errno == EWOULDBLOCK) {
- // continue
- return -2;
- }
- logger_->log_error("Could not recv on %d ( port %d), error: %s", fd, port_, strerror(errno));
- }
- return -1;
- }
- buflen -= bytes_read;
- buf += bytes_read;
- total_read += bytes_read;
- if (!retrieve_all_bytes) {
- break;
- }
- }
- total_read_ += total_read;
- return total_read;
-}
-
-} /* namespace io */
-} /* namespace minifi */
-} /* namespace nifi */
-} /* namespace apache */
-} /* namespace org */
diff --git a/libminifi/src/io/tls/TLSServerSocket.cpp b/libminifi/src/io/tls/TLSServerSocket.cpp
index 22b01a3..891d1b2 100644
--- a/libminifi/src/io/tls/TLSServerSocket.cpp
+++ b/libminifi/src/io/tls/TLSServerSocket.cpp
@@ -26,7 +26,10 @@
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
-#endif
+#else
+#pragma comment(lib, "Ws2_32.lib")
+#endif /* !WIN32 */
+
#include <cstdio>
#include <memory>
#include <chrono>
diff --git a/libminifi/src/io/tls/TLSSocket.cpp b/libminifi/src/io/tls/TLSSocket.cpp
index f251e73..66c004e 100644
--- a/libminifi/src/io/tls/TLSSocket.cpp
+++ b/libminifi/src/io/tls/TLSSocket.cpp
@@ -17,6 +17,10 @@
*/
#include <openssl/ssl.h>
#include <openssl/err.h>
+#ifdef WIN32
+#include <WS2tcpip.h>
+#pragma comment(lib, "Ws2_32.lib")
+#endif /* WIN32 */
#include <fstream>
#include <memory>
#include <utility>
@@ -30,6 +34,7 @@
#include "utils/StringUtils.h"
#include "core/Property.h"
#include "core/logging/LoggerConfiguration.h"
+#include "utils/GeneralUtils.h"
namespace org {
namespace apache {
@@ -37,23 +42,20 @@ namespace nifi {
namespace minifi {
namespace io {
-std::atomic<OpenSSLInitializer*> OpenSSLInitializer::context_instance;
-std::mutex OpenSSLInitializer::context_mutex;
-
-TLSContext::TLSContext(const std::shared_ptr<Configure> &configure, const std::shared_ptr<minifi::controllers::SSLContextService> &ssl_service)
+TLSContext::TLSContext(const std::shared_ptr<Configure> &configure, std::shared_ptr<minifi::controllers::SSLContextService> ssl_service)
: SocketContext(configure),
error_value(TLS_GOOD),
ctx(nullptr),
logger_(logging::LoggerFactory<TLSContext>::getLogger()),
configure_(configure),
- ssl_service_(ssl_service) {
+ ssl_service_(std::move(ssl_service)) {
}
/**
* The memory barrier is defined by the singleton
*/
int16_t TLSContext::initialize(bool server_method) {
- if (ctx != 0) {
+ if (ctx) {
return error_value;
}
@@ -71,7 +73,7 @@ int16_t TLSContext::initialize(bool server_method) {
const SSL_METHOD *method;
method = server_method ? TLSv1_2_server_method() : TLSv1_2_client_method();
ctx = SSL_CTX_new(method);
- if (ctx == NULL) {
+ if (ctx == nullptr) {
logger_->log_error("Could not create SSL context, error: %s.", std::strerror(errno));
error_value = TLS_ERROR_CONTEXT;
return error_value;
@@ -152,7 +154,7 @@ int16_t TLSContext::initialize(bool server_method) {
}
TLSSocket::~TLSSocket() {
- closeStream();
+ TLSSocket::closeStream();
}
void TLSSocket::closeStream() {
@@ -171,23 +173,38 @@ void TLSSocket::closeStream() {
* @param listeners number of listeners in the queue
*/
TLSSocket::TLSSocket(const std::shared_ptr<TLSContext> &context, const std::string &hostname, const uint16_t port, const uint16_t listeners)
- : Socket(context, hostname, port, listeners),
- ssl_(0) {
+ : Socket(context, hostname, port, listeners) {
logger_ = logging::LoggerFactory<TLSSocket>::getLogger();
context_ = context;
}
TLSSocket::TLSSocket(const std::shared_ptr<TLSContext> &context, const std::string &hostname, const uint16_t port)
- : Socket(context, hostname, port, 0),
- ssl_(0) {
+ : Socket(context, hostname, port, 0) {
logger_ = logging::LoggerFactory<TLSSocket>::getLogger();
context_ = context;
}
-TLSSocket::TLSSocket(const TLSSocket &&d)
- : Socket(std::move(d)),
- ssl_(0) {
- context_ = d.context_;
+TLSSocket::TLSSocket(TLSSocket &&other)
+ : Socket(std::move(other)),
+ context_{ utils::exchange(other.context_, nullptr) } {
+ std::lock_guard<std::mutex> lg{ other.ssl_mutex_ };
+
+ connected_.exchange(other.connected_.load());
+ other.connected_.exchange(false);
+ ssl_ = utils::exchange(other.ssl_, nullptr);
+ ssl_map_ = utils::exchange(other.ssl_map_, {});
+}
+
+TLSSocket& TLSSocket::operator=(TLSSocket&& other) {
+ if (&other == this) return *this;
+ this->Socket::operator=(static_cast<Socket&&>(other));
+ std::lock_guard<std::mutex> lg{ other.ssl_mutex_ };
+ connected_.exchange(other.connected_.load());
+ other.connected_.exchange(false);
+ context_ = utils::exchange(other.context_, nullptr);
+ ssl_ = utils::exchange(other.ssl_, nullptr);
+ ssl_map_ = utils::exchange(other.ssl_map_, {});
+ return *this;
}
int16_t TLSSocket::initialize(bool blocking) {
@@ -274,61 +291,60 @@ int16_t TLSSocket::select_descriptor(const uint16_t msec) {
select(socket_max_ + 1, &read_fds_, NULL, NULL, NULL);
for (int i = 0; i <= socket_max_; i++) {
- if (FD_ISSET(i, &read_fds_)) {
- if (i == socket_file_descriptor_) {
- if (listeners_ > 0) {
- struct sockaddr_in remoteaddr; // client address
- socklen_t addrlen = sizeof remoteaddr;
- int newfd = accept(socket_file_descriptor_, (struct sockaddr *) &remoteaddr, &addrlen);
- FD_SET(newfd, &total_list_); // add to master set
- if (newfd > socket_max_) { // keep track of the max
- socket_max_ = newfd;
- }
- auto ssl = SSL_new(context_->getContext());
- SSL_set_fd(ssl, newfd);
- auto accept_value = SSL_accept(ssl);
- if (accept_value != -1) {
- logger_->log_trace("Accepted on %d", newfd);
- ssl_map_[newfd] = ssl;
- return newfd;
- } else {
- int ssl_err = SSL_get_error(ssl, accept_value);
- logger_->log_error("Could not accept %d, error code %d", newfd, ssl_err);
- close_ssl(newfd);
- return -1;
- }
+ if (!FD_ISSET(i, &read_fds_)) continue;
+
+ if (i != socket_file_descriptor_) {
+ // data to be received on i
+ return i;
+ }
+
+ // listener can accept a new connection
+ if (listeners_ > 0) {
+ const auto newfd = accept(socket_file_descriptor_, nullptr, nullptr);
+ if (!valid_socket(newfd)) {
+ logger_->log_error("accept: %s", get_last_socket_error_message());
+ return -1;
+ }
+ FD_SET(newfd, &total_list_); // add to master set
+ if (newfd > socket_max_) { // keep track of the max
+ socket_max_ = newfd;
+ }
+ auto ssl = SSL_new(context_->getContext());
+ SSL_set_fd(ssl, newfd);
+ auto accept_value = SSL_accept(ssl);
+ if (accept_value != -1) {
+ logger_->log_trace("Accepted on %d", newfd);
+ ssl_map_[newfd] = ssl;
+ return newfd;
+ }
+ int ssl_err = SSL_get_error(ssl, accept_value);
+ logger_->log_error("Could not accept %d, error code %d", newfd, ssl_err);
+ close_ssl(newfd);
+ return -1;
+ }
+ if (!connected_) {
+ int rez = SSL_connect(ssl_);
+ if (rez < 0) {
+ ERR_print_errors_fp(stderr);
+ int ssl_error = SSL_get_error(ssl_, rez);
+ if (ssl_error == SSL_ERROR_WANT_WRITE) {
+ logger_->log_trace("want write");
+ return socket_file_descriptor_;
+ } else if (ssl_error == SSL_ERROR_WANT_READ) {
+ logger_->log_trace("want read");
+ return socket_file_descriptor_;
} else {
- if (!connected_) {
- int rez = SSL_connect(ssl_);
-
- if (rez < 0) {
- ERR_print_errors_fp(stderr);
- int ssl_error = SSL_get_error(ssl_, rez);
- if (ssl_error == SSL_ERROR_WANT_WRITE) {
- logger_->log_trace("want write");
- return socket_file_descriptor_;
- } else if (ssl_error == SSL_ERROR_WANT_READ) {
- logger_->log_trace("want read");
- return socket_file_descriptor_;
- } else {
- logger_->log_error("SSL socket connect failed to %s %d", requested_hostname_, port_);
- closeStream();
- return -1;
- }
- } else {
- connected_ = true;
- logger_->log_debug("SSL socket connect success to %s %d, on fd %d", requested_hostname_, port_, socket_file_descriptor_);
- return socket_file_descriptor_;
- }
- }
+ logger_->log_error("SSL socket connect failed (%d) to %s %d", ssl_error, requested_hostname_, port_);
+ closeStream();
+ return -1;
}
- return socket_file_descriptor_;
- // we have a new connection
- } else {
- // data to be received on i
- return i;
}
+ connected_ = true;
+ logger_->log_debug("SSL socket connect success to %s %d, on fd %d", requested_hostname_, port_, socket_file_descriptor_);
+ return socket_file_descriptor_;
}
+ return socket_file_descriptor_;
+ // we have a new connection
}
bool is_server = listeners_ > 0;
diff --git a/libminifi/src/io/win/ClientSocket.cpp b/libminifi/src/io/win/ClientSocket.cpp
deleted file mode 100644
index 5e82bfe..0000000
--- a/libminifi/src/io/win/ClientSocket.cpp
+++ /dev/null
@@ -1,556 +0,0 @@
-/**
- *
- * 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 "io/ClientSocket.h"
-#ifndef WIN32
-#include <netinet/tcp.h>
-#include <netdb.h>
-#include <sys/socket.h>
-#include <netinet/in.h>
-#include <arpa/inet.h>
-#include <sys/ioctl.h>
-#include <net/if.h>
-#include <ifaddrs.h>
-#include <unistd.h>
-#include <sys/types.h>
-#endif
-
-#include <memory>
-#include <utility>
-#include <vector>
-#include <cerrno>
-#include <string>
-#include "Exception.h"
-#include "io/validation.h"
-#include "core/logging/LoggerConfiguration.h"
-
-namespace {
-struct addrinfo_deleter {
- void operator()(addrinfo* const p) const noexcept {
- freeaddrinfo(p);
- }
-};
-} // namespace
-
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-namespace io {
-
-Socket::Socket(const std::shared_ptr<SocketContext> &context, const std::string &hostname, const uint16_t port, const uint16_t listeners = -1)
- : requested_hostname_(hostname),
- port_(port),
- socket_file_descriptor_(-1),
- socket_max_(0),
- total_written_(0),
- total_read_(0),
- is_loopback_only_(false),
- listeners_(listeners),
- canonical_hostname_(""),
- nonBlocking_(false),
- logger_(logging::LoggerFactory<Socket>::getLogger()) {
- FD_ZERO(&total_list_);
- FD_ZERO(&read_fds_);
- initialize_socket();
-}
-
-Socket::Socket(const std::shared_ptr<SocketContext> &context, const std::string &hostname, const uint16_t port)
- : Socket(context, hostname, port, 0) {
- initialize_socket();
-}
-
-Socket::Socket(const Socket &&other)
- : requested_hostname_(std::move(other.requested_hostname_)),
- port_(std::move(other.port_)),
- is_loopback_only_(false),
- socket_file_descriptor_(other.socket_file_descriptor_),
- socket_max_(other.socket_max_.load()),
- listeners_(other.listeners_),
- total_list_(other.total_list_),
- read_fds_(other.read_fds_),
- canonical_hostname_(std::move(other.canonical_hostname_)),
- nonBlocking_(false),
- logger_(std::move(other.logger_)) {
- total_written_ = other.total_written_.load();
- total_read_ = other.total_read_.load();
-}
-
-Socket::~Socket() {
- closeStream();
-}
-
-void Socket::closeStream() {
- if (socket_file_descriptor_ >= 0 && socket_file_descriptor_ != INVALID_SOCKET) {
- logging::LOG_DEBUG(logger_) << "Closing " << socket_file_descriptor_;
-#ifdef WIN32
- closesocket(socket_file_descriptor_);
-#else
- close(socket_file_descriptor_);
-#endif
- socket_file_descriptor_ = -1;
- }
- if (total_written_ > 0) {
- local_network_interface_.log_write(total_written_);
- total_written_ = 0;
- }
- if (total_read_ > 0) {
- local_network_interface_.log_read(total_read_);
- total_read_ = 0;
- }
-}
-
-void Socket::setNonBlocking() {
- if (listeners_ <= 0) {
- nonBlocking_ = true;
- }
-}
-#ifdef WIN32
-int8_t Socket::createConnection(const addrinfo *p, struct in_addr &addr) {
-#else
-int8_t Socket::createConnection(const addrinfo *p, in_addr_t &addr) {
-#endif
- if ((socket_file_descriptor_ = socket(p->ai_family, p->ai_socktype, p->ai_protocol)) == INVALID_SOCKET) {
- logger_->log_error("error while connecting to server socket");
- return -1;
- }
-
- setSocketOptions(socket_file_descriptor_);
-
- if (listeners_ <= 0 && !local_network_interface_.getInterface().empty()) {
- // bind to local network interface
-#ifndef WIN32
- ifaddrs* list = NULL;
- ifaddrs* item = NULL;
- ifaddrs* itemFound = NULL;
- int result = getifaddrs(&list);
- if (result == 0) {
- item = list;
- while (item) {
- if ((item->ifa_addr != NULL) && (item->ifa_name != NULL) && (AF_INET == item->ifa_addr->sa_family)) {
- if (strcmp(item->ifa_name, local_network_interface_.getInterface().c_str()) == 0) {
- itemFound = item;
- break;
- }
- }
- item = item->ifa_next;
- }
-
- if (itemFound != NULL) {
- result = bind(socket_file_descriptor_, itemFound->ifa_addr, sizeof(struct sockaddr_in));
- if (result < 0)
- logger_->log_info("Bind to interface %s failed %s", local_network_interface_.getInterface(), strerror(errno));
- else
- logger_->log_info("Bind to interface %s", local_network_interface_.getInterface());
- }
- freeifaddrs(list);
- }
-#endif
- }
-
- if (listeners_ > 0) {
- struct sockaddr_in *sa_loc = (struct sockaddr_in*) p->ai_addr;
- sa_loc->sin_family = AF_INET;
- sa_loc->sin_port = htons(port_);
- if (is_loopback_only_) {
- sa_loc->sin_addr.s_addr = htonl(INADDR_LOOPBACK);
- } else {
- sa_loc->sin_addr.s_addr = htonl(INADDR_ANY);
- }
- if (bind(socket_file_descriptor_, p->ai_addr, p->ai_addrlen) == -1) {
- logger_->log_error("Could not bind to socket, reason %s", strerror(errno));
- return -1;
- }
- }
- {
- if (listeners_ <= 0) {
- struct sockaddr_in sa_loc;
- memset(&sa_loc, 0x00, sizeof(sa_loc));
- sa_loc.sin_family = AF_INET;
- sa_loc.sin_port = htons(port_);
-
- // use any address if you are connecting to the local machine for testing
- // otherwise we must use the requested hostname
- if (IsNullOrEmpty(requested_hostname_) || requested_hostname_ == "localhost") {
- if (is_loopback_only_) {
- sa_loc.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
- } else {
- sa_loc.sin_addr.s_addr = htonl(INADDR_ANY);
- }
- } else {
-#ifdef WIN32
- sa_loc.sin_addr.s_addr = addr.s_addr;
-#else
- sa_loc.sin_addr.s_addr = addr;
-#endif
- }
- if (connect(socket_file_descriptor_, (struct sockaddr*) &sa_loc, sizeof(sa_loc)) == -1) {
-#ifdef WIN32
- int err = WSAGetLastError();
- if (err == WSAEADDRNOTAVAIL) {
- logger_->log_error("invalid or unknown IP");
- } else if (err == WSAECONNREFUSED) {
- logger_->log_error("Connection refused");
- } else {
- logger_->log_error("Unknown error");
- }
-#endif
- closeStream();
- return -1;
- }
- }
- }
-
- // listen
- if (listeners_ > 0) {
- if (listen(socket_file_descriptor_, listeners_) == -1) {
- return -1;
- } else {
- logger_->log_debug("Created connection with %d listeners", listeners_);
- }
- }
- // add the listener to the total set
- FD_SET(socket_file_descriptor_, &total_list_);
- socket_max_ = socket_file_descriptor_;
- logger_->log_debug("Created connection with file descriptor %d", socket_file_descriptor_);
- return 0;
-}
-
-int16_t Socket::initialize() {
- addrinfo hints = { sizeof(addrinfo) };
- memset(&hints, 0, sizeof hints); // make sure the struct is empty
- hints.ai_family = AF_UNSPEC;
- hints.ai_socktype = SOCK_STREAM;
- hints.ai_flags = AI_CANONNAME;
- if (listeners_ > 0)
- hints.ai_flags |= AI_PASSIVE;
- hints.ai_protocol = 0; /* any protocol */
-
- addrinfo* getaddrinfo_result = nullptr;
- const int errcode = getaddrinfo(requested_hostname_.c_str(), nullptr, &hints, &getaddrinfo_result);
- if (errcode != 0) {
- logger_->log_error("Saw error during getaddrinfo, error: %lu", WSAGetLastError());
- return -1;
- }
- const std::unique_ptr<addrinfo, addrinfo_deleter> addr_info{ getaddrinfo_result };
- getaddrinfo_result = nullptr;
- socket_file_descriptor_ = -1;
-
-#ifndef WIN32
- in_addr_t addr;
-#else
- in_addr addr;
-#endif
- struct hostent *h;
-#if defined( __MACH__ ) || defined(WIN32)
- h = gethostbyname(requested_hostname_.c_str());
-#else
- const char *host;
-
- host = requested_hostname_.c_str();
- char buf[1024];
- struct hostent he;
- int hh_errno;
- gethostbyname_r(host, &he, buf, sizeof(buf), &h, &hh_errno);
-#endif
- if (h == nullptr) {
- return -1;
- }
- memcpy(reinterpret_cast<char*>(&addr), h->h_addr_list[0], h->h_length);
-
- for (const auto* p = addr_info.get(); p; p = p->ai_next) {
- if (IsNullOrEmpty(canonical_hostname_)) {
- if (!IsNullOrEmpty(p) && !IsNullOrEmpty(p->ai_canonname))
- canonical_hostname_ = p->ai_canonname;
- }
- // we've successfully connected
- if (port_ > 0 && createConnection(p, addr) >= 0) {
- // Put the socket in non-blocking mode:
- if (nonBlocking_) {
-#ifndef WIN32
- if (fcntl(socket_file_descriptor_, F_SETFL, O_NONBLOCK) < 0) {
- // handle error
- logger_->log_error("Could not create non blocking to socket", strerror(errno));
- } else {
- logger_->log_debug("Successfully applied O_NONBLOCK to fd");
- }
-#else
- u_long iMode = 1;
- if (ioctlsocket(socket_file_descriptor_, FIONBIO, &iMode) == NO_ERROR) {
- logger_->log_debug("Successfully applied O_NONBLOCK to fd");
- }
-#endif
- }
- logger_->log_debug("Successfully created connection");
- return 0;
- break;
- }
- }
-
- logger_->log_debug("Could not find device for our connection");
- return -1;
-}
-
-int16_t Socket::select_descriptor(const uint16_t msec) {
- if (listeners_ == 0) {
- return socket_file_descriptor_;
- }
-
- struct timeval tv;
-
- read_fds_ = total_list_;
-
- tv.tv_sec = msec / 1000;
- tv.tv_usec = (msec % 1000) * 1000;
-
- std::lock_guard<std::recursive_mutex> guard(selection_mutex_);
-
- if (msec > 0)
- select(socket_max_ + 1, &read_fds_, NULL, NULL, &tv);
- else
- select(socket_max_ + 1, &read_fds_, NULL, NULL, NULL);
-
- for (int i = 0; i <= socket_max_; i++) {
- if (FD_ISSET(i, &read_fds_)) {
- if (i == socket_file_descriptor_) {
- if (listeners_ > 0) {
- struct sockaddr_storage remoteaddr; // client address
- socklen_t addrlen = sizeof remoteaddr;
- int newfd = accept(socket_file_descriptor_, (struct sockaddr *) &remoteaddr, &addrlen);
- FD_SET(newfd, &total_list_); // add to master set
- if (newfd > socket_max_) { // keep track of the max
- socket_max_ = newfd;
- }
- return newfd;
-
- } else {
- return socket_file_descriptor_;
- }
- // we have a new connection
- } else {
- // data to be received on i
- return i;
- }
- }
- }
-
- logger_->log_debug("Could not find a suitable file descriptor or select timed out");
-
- return -1;
-}
-
-int16_t Socket::setSocketOptions(const SocketDescriptor sock) {
- int opt = 1;
-#ifndef WIN32
-#ifndef __MACH__
- if (setsockopt(sock, SOL_TCP, TCP_NODELAY, static_cast<void*>(&opt), sizeof(opt)) < 0) {
- logger_->log_error("setsockopt() TCP_NODELAY failed");
- close(sock);
- return -1;
- }
- if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, reinterpret_cast<char*>(&opt), sizeof(opt)) < 0) {
- logger_->log_error("setsockopt() SO_REUSEADDR failed");
- close(sock);
- return -1;
- }
-
- int sndsize = 256 * 1024;
- if (setsockopt(sock, SOL_SOCKET, SO_SNDBUF, reinterpret_cast<char *>(&sndsize), sizeof(sndsize)) < 0) {
- logger_->log_error("setsockopt() SO_SNDBUF failed");
- close(sock);
- return -1;
- }
-
-#else
- if (listeners_ > 0) {
- // lose the pesky "address already in use" error message
- if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, reinterpret_cast<char *>(&opt), sizeof(opt)) < 0) {
- logger_->log_error("setsockopt() SO_REUSEADDR failed");
- close(sock);
- return -1;
- }
- }
-#endif
-#endif
- return 0;
-}
-
-std::string Socket::getHostname() const {
- return canonical_hostname_;
-}
-
-int Socket::writeData(std::vector<uint8_t> &buf, int buflen) {
- if (buflen < 0) {
- throw minifi::Exception{ExceptionType::GENERAL_EXCEPTION, "negative buflen"};
- }
-
- if (buf.size() < static_cast<size_t>(buflen))
- return -1;
- return writeData(buf.data(), buflen);
-}
-
-// data stream overrides
-
-int Socket::writeData(uint8_t *value, int size) {
- int ret = 0, bytes = 0;
-
- int fd = select_descriptor(1000);
- if (fd < 0) {
- return -1;
- }
- while (bytes < size) {
-#ifdef WIN32
- const char *vts = (const char*)value;
- ret = send(fd, vts + bytes, size - bytes, 0);
-#else
- ret = send(fd, value + bytes, size - bytes, 0);
-#endif
- // check for errors
- if (ret <= 0) {
- close(fd);
- logger_->log_error("Could not send to %d, error: %s", fd, strerror(errno));
- return ret;
- }
- bytes += ret;
- }
-
- if (ret)
- logger_->log_trace("Send data size %d over socket %d", size, fd);
- total_written_ += bytes;
- return bytes;
-}
-
-template<typename T>
-inline std::vector<uint8_t> Socket::readBuffer(const T& t) {
- std::vector<uint8_t> buf;
- buf.resize(sizeof t);
- readData(reinterpret_cast<uint8_t *>(&buf[0]), sizeof(t));
- return buf;
-}
-
-/**
- * Return the port for this socket
- * @returns port
- */
-uint16_t Socket::getPort() const {
- return port_;
-}
-
-int Socket::write(uint64_t base_value, bool is_little_endian) {
- return Serializable::write(base_value, this, is_little_endian);
-}
-
-int Socket::write(uint32_t base_value, bool is_little_endian) {
- return Serializable::write(base_value, this, is_little_endian);
-}
-
-int Socket::write(uint16_t base_value, bool is_little_endian) {
- return Serializable::write(base_value, this, is_little_endian);
-}
-
-int Socket::read(uint64_t &value, bool is_little_endian) {
- auto buf = readBuffer(value);
-
- if (is_little_endian) {
- value = ((uint64_t) buf[0] << 56) | ((uint64_t) (buf[1] & 255) << 48) | ((uint64_t) (buf[2] & 255) << 40) | ((uint64_t) (buf[3] & 255) << 32) | ((uint64_t) (buf[4] & 255) << 24)
- | ((uint64_t) (buf[5] & 255) << 16) | ((uint64_t) (buf[6] & 255) << 8) | ((uint64_t) (buf[7] & 255) << 0);
- } else {
- value = ((uint64_t) buf[0] << 0) | ((uint64_t) (buf[1] & 255) << 8) | ((uint64_t) (buf[2] & 255) << 16) | ((uint64_t) (buf[3] & 255) << 24) | ((uint64_t) (buf[4] & 255) << 32)
- | ((uint64_t) (buf[5] & 255) << 40) | ((uint64_t) (buf[6] & 255) << 48) | ((uint64_t) (buf[7] & 255) << 56);
- }
- return sizeof(value);
-}
-
-int Socket::read(uint32_t &value, bool is_little_endian) {
- auto buf = readBuffer(value);
-
- if (is_little_endian) {
- value = (buf[0] << 24) | (buf[1] << 16) | (buf[2] << 8) | buf[3];
- } else {
- value = buf[0] | buf[1] << 8 | buf[2] << 16 | buf[3] << 24;
- }
- return sizeof(value);
-}
-
-int Socket::read(uint16_t &value, bool is_little_endian) {
- auto buf = readBuffer(value);
-
- if (is_little_endian) {
- value = (buf[0] << 8) | buf[1];
- } else {
- value = buf[0] | buf[1] << 8;
- }
- return sizeof(value);
-}
-
-int Socket::readData(std::vector<uint8_t> &buf, int buflen, bool retrieve_all_bytes) {
- if (buflen < 0) {
- throw minifi::Exception{ExceptionType::GENERAL_EXCEPTION, "negative buflen"};
- }
-
- if (buf.size() < static_cast<size_t>(buflen)) {
- buf.resize(buflen);
- }
- return readData(buf.data(), buflen, retrieve_all_bytes);
-}
-
-int Socket::readData(uint8_t *buf, int buflen, bool retrieve_all_bytes) {
- int32_t total_read = 0;
- while (buflen) {
- int16_t fd = select_descriptor(1000);
- if (fd < 0) {
- if (listeners_ <= 0) {
- logger_->log_debug("fd %d close %i", fd, buflen);
- close(socket_file_descriptor_);
- }
- return -1;
- }
-#ifdef WIN32
- char *bfs = reinterpret_cast<char*>(buf);
- int bytes_read = recv(fd, bfs, buflen, 0);
-#else
- int bytes_read = recv(fd, buf, buflen, 0);
-#endif
- logger_->log_trace("Recv call %d", bytes_read);
- if (bytes_read <= 0) {
- if (bytes_read == 0) {
- logger_->log_debug("Other side hung up on %d", fd);
- } else {
- if (errno == EAGAIN || errno == EWOULDBLOCK) {
- // continue
- return -2;
- }
- logger_->log_error("Could not recv on %d, error: %s", fd, strerror(errno));
- }
- return -1;
- }
- buflen -= bytes_read;
- buf += bytes_read;
- total_read += bytes_read;
- if (!retrieve_all_bytes) {
- break;
- }
- }
- total_read_ += total_read;
- return total_read;
-}
-
-} /* namespace io */
-} /* namespace minifi */
-} /* namespace nifi */
-} /* namespace apache */
-} /* namespace org */
diff --git a/libminifi/test/RandomServerSocket.cpp b/libminifi/test/RandomServerSocket.cpp
index 2ff01f7..f912fbb 100644
--- a/libminifi/test/RandomServerSocket.cpp
+++ b/libminifi/test/RandomServerSocket.cpp
@@ -18,6 +18,10 @@
#include "./RandomServerSocket.h"
+#ifdef WIN32
+#pragma comment(lib, "Ws2_32.lib")
+#endif /* WIN32 */
+
#include <sstream>
#include <random>
#include <string>
diff --git a/libminifi/test/TestBase.h b/libminifi/test/TestBase.h
index a51b573..73d1430 100644
--- a/libminifi/test/TestBase.h
+++ b/libminifi/test/TestBase.h
@@ -164,7 +164,7 @@ class LogTestController {
}
modified_loggers.clear();
if (config)
- config = std::move(logging::LoggerConfiguration::newInstance());
+ config = logging::LoggerConfiguration::newInstance();
resetStream(log_output);
}
@@ -206,7 +206,7 @@ class LogTestController {
logging::LoggerConfiguration::getConfiguration().initialize(my_properties_);
logger_ = logging::LoggerConfiguration::getConfiguration().getLogger(core::getClassName<LogTestController>());
} else {
- config = std::move(logging::LoggerConfiguration::newInstance());
+ config = logging::LoggerConfiguration::newInstance();
// create for test purposes. most tests use the main logging factory, but this exists to test the logging
// framework itself.
config->initialize(my_properties_);
diff --git a/nanofi/include/cxx/CallbackProcessor.h b/nanofi/include/cxx/CallbackProcessor.h
index 62eaa68..7b6b78b 100644
--- a/nanofi/include/cxx/CallbackProcessor.h
+++ b/nanofi/include/cxx/CallbackProcessor.h
@@ -27,6 +27,7 @@
#include <thread>
#include <functional>
#include <iostream>
+#include <utility>
#include <sys/types.h>
#include "core/cstructs.h"
#include "io/BaseStream.h"
@@ -53,45 +54,38 @@ class CallbackProcessor : public core::Processor {
* Create a new processor
*/
CallbackProcessor(std::string name, utils::Identifier uuid = utils::Identifier())
- : Processor(name, uuid),
- ontrigger_callback_({}),
- onschedule_callback_({}),
- objref_(nullptr),
- logger_(logging::LoggerFactory<CallbackProcessor>::getLogger()) {
+ : Processor(std::move(name), uuid) {
}
// Destructor
- virtual ~CallbackProcessor() {
-
- }
+ ~CallbackProcessor() override = default;
// Processor Name
static constexpr char const* ProcessorName = "CallbackProcessor";
public:
-
void setCallback(void *obj, std::function<void(core::ProcessSession*, core::ProcessContext *context)> ontrigger_callback,
std::function<void(core::ProcessContext *context)> onschedule_callback = {}) {
objref_ = obj;
- ontrigger_callback_ = ontrigger_callback;
- onschedule_callback_ = onschedule_callback;
+ ontrigger_callback_ = std::move(ontrigger_callback);
+ onschedule_callback_ = std::move(onschedule_callback);
}
- virtual void onSchedule(core::ProcessContext *context, core::ProcessSessionFactory *sessionFactory);
+ void onSchedule(core::ProcessContext *context, core::ProcessSessionFactory *sessionFactory) override;
// OnTrigger method, implemented by MiNiFi CallbackProcessor
- virtual void onTrigger(core::ProcessContext *context, core::ProcessSession *session); // override;
+ void onTrigger(core::ProcessContext *context, core::ProcessSession *session) override; // override;
// Initialize, over write by NiFi CallbackProcessor
- virtual void initialize(); // override;
+ void initialize() override; // override;
- virtual bool supportsDynamicProperties() /*override*/ {
+ bool supportsDynamicProperties() override /*override*/ {
return true;
}
protected:
- void *objref_;
+ void *objref_{ nullptr };
std::function<void(core::ProcessSession*, core::ProcessContext *context)> ontrigger_callback_;
std::function<void(core::ProcessContext *context)> onschedule_callback_;
private:
// Logger
- std::shared_ptr<logging::Logger> logger_;
+ std::shared_ptr<logging::Logger> logger_{ logging::LoggerFactory<CallbackProcessor>::getLogger() };
};
diff --git a/nanofi/include/cxx/Instance.h b/nanofi/include/cxx/Instance.h
index 0325081..4684cd3 100644
--- a/nanofi/include/cxx/Instance.h
+++ b/nanofi/include/cxx/Instance.h
@@ -67,12 +67,9 @@ class Instance {
public:
explicit Instance(const std::string &url, const std::string &port, const std::string &repo_class_name = "")
- : configure_(std::make_shared<Configure>()),
+ : no_op_repo_(std::make_shared<minifi::core::Repository>()),
url_(url),
- agent_(nullptr),
- rpgInitialized_(false),
- listener_thread_pool_(1),
- no_op_repo_(std::make_shared<minifi::core::Repository>()) {
+ configure_(std::make_shared<Configure>()) {
if (repo_class_name == "filesystemrepository") {
content_repo_ = std::make_shared<minifi::core::repository::FileSystemRepository>();
@@ -171,12 +168,11 @@ class Instance {
std::shared_ptr<c2::C2CallbackAgent> agent_;
- std::atomic<bool> running_;
+ std::atomic<bool> running_{ false };
- bool rpgInitialized_;
+ bool rpgInitialized_{ false };
std::shared_ptr<minifi::core::Repository> no_op_repo_;
-
std::shared_ptr<minifi::core::ContentRepository> content_repo_;
std::shared_ptr<core::ProcessorNode> proc_node_;
@@ -185,7 +181,7 @@ class Instance {
std::string url_;
std::shared_ptr<Configure> configure_;
- utils::ThreadPool<state::Update> listener_thread_pool_;
+ utils::ThreadPool<state::Update> listener_thread_pool_{ 1 };
};
} /* namespace minifi */
diff --git a/nanofi/tests/CLogAggregatorTests.cpp b/nanofi/tests/CLogAggregatorTests.cpp
index edf62c9..31a9007 100644
--- a/nanofi/tests/CLogAggregatorTests.cpp
+++ b/nanofi/tests/CLogAggregatorTests.cpp
@@ -69,7 +69,7 @@ std::string join_strings(const std::vector<std::string>& strings, const std::str
*/
TEST_CASE("Test string tokenizer empty string", "[stringTokenizerEmptyString]") {
- std::string delimitedString = "";
+ std::string delimitedString;
char delim = '-';
struct token_list tokens = tokenize_string(delimitedString.c_str(), delim);
REQUIRE(tokens.size == 0);
@@ -254,7 +254,7 @@ TEST_CASE("Test tail file with less than 4096 delimited chars", "[testLogAggrega
TEST_CASE("Test tail file having 4096 bytes without delimiter", "[testLogAggregateFile4096Chars]") {
FileManager fm("test.txt");
- const std::string s = std::move(fm.WriteNChars(4096, 'a'));
+ const std::string s = fm.WriteNChars(4096, 'a');
const std::string filePath = fm.getFilePath();
fm.CloseStream();
@@ -271,8 +271,8 @@ TEST_CASE("Test tail file having 4096 bytes without delimiter", "[testLogAggrega
TEST_CASE("Test tail file having more than 4096 bytes without delimiter", "[testLogAggregarteFileMoreThan4096Chars]") {
FileManager fm("test.txt");
- const std::string s1 = std::move(fm.WriteNChars(4096, 'a'));
- const std::string s2 = std::move(fm.WriteNChars(4096, 'b'));
+ const std::string s1 = fm.WriteNChars(4096, 'a');
+ const std::string s2 = fm.WriteNChars(4096, 'b');
const std::string s3 = "helloworld";
fm.Write(s3);
fm.CloseStream();
@@ -295,9 +295,9 @@ TEST_CASE("Test tail file having more than 4096 bytes without delimiter", "[test
TEST_CASE("Test tail file having more than 4096 bytes with delimiter", "[testLogAggregateWithDelimitedString]") {
FileManager fm("test.txt");
- const std::string s1 = std::move(fm.WriteNChars(4096, 'a'));
- const std::string d1 = std::move(fm.WriteNChars(2, ';'));
- const std::string s2 = std::move(fm.WriteNChars(4096, 'b'));
+ const std::string s1 = fm.WriteNChars(4096, 'a');
+ const std::string d1 = fm.WriteNChars(2, ';');
+ const std::string s2 = fm.WriteNChars(4096, 'b');
fm.CloseStream();
TailFileTestResourceManager mgr("LogAggregator", on_trigger_logaggregator);
@@ -318,9 +318,9 @@ TEST_CASE("Test tail file having more than 4096 bytes with delimiter", "[testLog
TEST_CASE("Test tail file having more than 4096 bytes with delimiter and second chunk less than 4096", "[testLogAggregateDelimited]") {
FileManager fm("test.txt");
- const std::string s1 = std::move(fm.WriteNChars(4096, 'a'));
- const std::string d1 = std::move(fm.WriteNChars(2, ';'));
- const std::string s2 = std::move(fm.WriteNChars(4000, 'b'));
+ const std::string s1 = fm.WriteNChars(4096, 'a');
+ const std::string d1 = fm.WriteNChars(2, ';');
+ const std::string s2 = fm.WriteNChars(4000, 'b');
fm.CloseStream();
TailFileTestResourceManager mgr("LogAggregator", on_trigger_logaggregator);
@@ -342,9 +342,9 @@ TEST_CASE("Test tail file having more than 4096 bytes with delimiter and second
TEST_CASE("Test tail file having more than 4096 bytes with delimiter and second chunk more than 4096", "[testLogAggregateDelimited]") {
FileManager fm("test.txt");
- const std::string s1 = std::move(fm.WriteNChars(4096, 'a'));
- const std::string d1 = std::move(fm.WriteNChars(2, ';'));
- const std::string s2 = std::move(fm.WriteNChars(4098, 'b'));
+ const std::string s1 = fm.WriteNChars(4096, 'a');
+ const std::string d1 = fm.WriteNChars(2, ';');
+ const std::string s2 = fm.WriteNChars(4098, 'b');
fm.CloseStream();
TailFileTestResourceManager mgr("LogAggregator", on_trigger_logaggregator);