[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #864: MINIFICPP-1319 - Stream refactor
adamdebreceni commented on a change in pull request #864:
URL: https://github.com/apache/nifi-minifi-cpp/pull/864#discussion_r486382255
##
File path: libminifi/src/io/InputStream.cpp
##
@@ -0,0 +1,83 @@
+/**
+ *
+ * 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
+#include
+#include
+#include
+#include
+#include "io/InputStream.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace io {
+
+int InputStream::read(std::vector& buffer, int len) {
+ if (buffer.size() < len) {
+buffer.resize(len);
+ }
+ int ret = read(buffer.data(), len);
+ buffer.resize((std::max)(ret, 0));
+ return ret;
+}
+
+int InputStream::read(bool ) {
+ uint8_t buf = 0;
+
+ if (read(, 1) != 1) {
+return -1;
+ }
+ value = buf;
+ return 1;
+}
+
+int InputStream::read(std::string , bool widen) {
Review comment:
it seems like there is actual need for `widen` as some parts call it
with `true` while others with `false`, further investigation is necessary but I
wouldn't change this in this PR
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #864: MINIFICPP-1319 - Stream refactor
adamdebreceni commented on a change in pull request #864:
URL: https://github.com/apache/nifi-minifi-cpp/pull/864#discussion_r486373948
##
File path: libminifi/include/io/CRCStream.h
##
@@ -34,307 +35,111 @@
#endif
#include "BaseStream.h"
#include "Exception.h"
-#include "Serializable.h"
namespace org {
namespace apache {
namespace nifi {
namespace minifi {
namespace io {
+namespace internal {
-template
-class CRCStream : public BaseStream {
+template
+class CRCStreamBase : public virtual Stream {
public:
- /**
- * Raw pointer because the caller guarantees that
- * it will exceed our lifetime.
- */
- explicit CRCStream(T *child_stream);
- CRCStream(T *child_stream, uint64_t initial_crc);
-
- CRCStream(CRCStream&&) noexcept;
-
- ~CRCStream() override = default;
-
- T *getstream() const {
+ StreamType *getstream() const {
return child_stream_;
}
- void disableEncoding() {
-disable_encoding_ = true;
- }
-
- /**
- * Reads data and places it into buf
- * @param buf buffer in which we extract data
- * @param buflen
- */
- int readData(std::vector , int buflen) override;
- /**
- * Reads data and places it into buf
- * @param buf buffer in which we extract data
- * @param 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 , int buflen);
-
- /**
- * writes value to stream
- * @param value value to write
- * @param size size of value
- */
- int writeData(uint8_t *value, int size) override;
-
- using BaseStream::write;
-
- /**
- * write 4 bytes to stream
- * @param base_value non encoded value
- * @param stream output stream
- * @param is_little_endian endianness determination
- * @return resulting write size
- **/
- 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
- * @param stream output stream
- * @param is_little_endian endianness determination
- * @return resulting write size
- **/
- int write(uint16_t base_value, bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
-
- /**
- * write 8 bytes to stream
- * @param base_value non encoded value
- * @param stream output stream
- * @param is_little_endian endianness determination
- * @return resulting write size
- **/
- int write(uint64_t base_value, bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
+ void close() override { child_stream_->close(); }
- /**
- * Reads a system word
- * @param value value to write
- */
- int read(uint64_t , bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
-
- /**
- * Reads a uint32_t
- * @param value value to write
- */
- int read(uint32_t , bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
-
- /**
- * Reads a system short
- * @param value value to write
- */
- int read(uint16_t , bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
-
- const size_t getSize() const override { return child_stream_->getSize(); }
-
- void closeStream() override { child_stream_->closeStream(); }
-
- short initialize() override { // NOLINT
+ int initialize() override {
child_stream_->initialize();
reset();
return 0;
}
- void updateCRC(uint8_t *buffer, uint32_t length);
+ void updateCRC(uint8_t *buffer, uint32_t length) {
+crc_ = crc32(crc_, buffer, length);
+ }
uint64_t getCRC() {
-return crc_;
+return gsl::narrow(crc_);
}
- void reset();
-
- protected:
- /**
- * Creates a vector and returns the vector using the provided
- * type name.
- * @param t incoming object
- * @returns vector.
- */
- template
- std::vector readBuffer(const K& t) {
-std::vector buf;
-readBuffer(buf, t);
-return buf;
+ void reset() {
+crc_ = crc32(0L, Z_NULL, 0);
}
- /**
- * Populates the vector using the provided type name.
- * @param buf output buffer
- * @param t incoming object
- * @returns number of bytes read.
- */
- template
- int readBuffer(std::vector& buf, const K& t) {
-buf.resize(sizeof t);
-return readData(reinterpret_cast(buf.data()), sizeof(t));
+ protected:
+ explicit CRCStreamBase(gsl::not_null child_stream) :
child_stream_(child_stream) {}
+ // this implementation is here to make MSVC happy, declaration would be
enough
+ // as it will never get called (and should not be called)
+ CRCStreamBase() : child_stream_(gsl::make_not_null(nullptr)) {
+assert(false);
Review comment:
in light of recent changes (i.e. making the field nullable) this whole
questionable stuff is unnecessary, and has been removed
This is an automated message from the Apache Git Service.
To
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #864: MINIFICPP-1319 - Stream refactor
adamdebreceni commented on a change in pull request #864:
URL: https://github.com/apache/nifi-minifi-cpp/pull/864#discussion_r486372528
##
File path: libminifi/test/unit/SiteToSiteHelper.h
##
@@ -19,132 +19,46 @@
#define LIBMINIFI_TEST_UNIT_SITE2SITE_HELPER_H_
#include
-#include "io/BaseStream.h"
+#include "io/BufferStream.h"
#include "io/EndianCheck.h"
#include "core/Core.h"
/**
* Test repository
*/
class SiteToSiteResponder : public minifi::io::BaseStream {
private:
- std::queue server_responses_;
+ minifi::io::BufferStream server_responses_;
std::queue client_responses_;
public:
SiteToSiteResponder() = default;
// initialize
- virtual short initialize() {
+ int initialize() override {
return 1;
}
- void push_response(std::string resp) {
-server_responses_.push(resp);
+ void push_response(const std::string& resp) {
+server_responses_.write((const uint8_t*)resp.data(), resp.length());
Review comment:
done
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #864: MINIFICPP-1319 - Stream refactor
adamdebreceni commented on a change in pull request #864:
URL: https://github.com/apache/nifi-minifi-cpp/pull/864#discussion_r486371593
##
File path: libminifi/include/io/CRCStream.h
##
@@ -34,307 +35,111 @@
#endif
#include "BaseStream.h"
#include "Exception.h"
-#include "Serializable.h"
namespace org {
namespace apache {
namespace nifi {
namespace minifi {
namespace io {
+namespace internal {
-template
-class CRCStream : public BaseStream {
+template
+class CRCStreamBase : public virtual Stream {
public:
- /**
- * Raw pointer because the caller guarantees that
- * it will exceed our lifetime.
- */
- explicit CRCStream(T *child_stream);
- CRCStream(T *child_stream, uint64_t initial_crc);
-
- CRCStream(CRCStream&&) noexcept;
-
- ~CRCStream() override = default;
-
- T *getstream() const {
+ StreamType *getstream() const {
return child_stream_;
}
- void disableEncoding() {
-disable_encoding_ = true;
- }
-
- /**
- * Reads data and places it into buf
- * @param buf buffer in which we extract data
- * @param buflen
- */
- int readData(std::vector , int buflen) override;
- /**
- * Reads data and places it into buf
- * @param buf buffer in which we extract data
- * @param 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 , int buflen);
-
- /**
- * writes value to stream
- * @param value value to write
- * @param size size of value
- */
- int writeData(uint8_t *value, int size) override;
-
- using BaseStream::write;
-
- /**
- * write 4 bytes to stream
- * @param base_value non encoded value
- * @param stream output stream
- * @param is_little_endian endianness determination
- * @return resulting write size
- **/
- 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
- * @param stream output stream
- * @param is_little_endian endianness determination
- * @return resulting write size
- **/
- int write(uint16_t base_value, bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
-
- /**
- * write 8 bytes to stream
- * @param base_value non encoded value
- * @param stream output stream
- * @param is_little_endian endianness determination
- * @return resulting write size
- **/
- int write(uint64_t base_value, bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
+ void close() override { child_stream_->close(); }
- /**
- * Reads a system word
- * @param value value to write
- */
- int read(uint64_t , bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
-
- /**
- * Reads a uint32_t
- * @param value value to write
- */
- int read(uint32_t , bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
-
- /**
- * Reads a system short
- * @param value value to write
- */
- int read(uint16_t , bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
-
- const size_t getSize() const override { return child_stream_->getSize(); }
-
- void closeStream() override { child_stream_->closeStream(); }
-
- short initialize() override { // NOLINT
+ int initialize() override {
child_stream_->initialize();
reset();
return 0;
}
- void updateCRC(uint8_t *buffer, uint32_t length);
+ void updateCRC(uint8_t *buffer, uint32_t length) {
+crc_ = crc32(crc_, buffer, length);
+ }
uint64_t getCRC() {
-return crc_;
+return gsl::narrow(crc_);
}
- void reset();
-
- protected:
- /**
- * Creates a vector and returns the vector using the provided
- * type name.
- * @param t incoming object
- * @returns vector.
- */
- template
- std::vector readBuffer(const K& t) {
-std::vector buf;
-readBuffer(buf, t);
-return buf;
+ void reset() {
+crc_ = crc32(0L, Z_NULL, 0);
}
- /**
- * Populates the vector using the provided type name.
- * @param buf output buffer
- * @param t incoming object
- * @returns number of bytes read.
- */
- template
- int readBuffer(std::vector& buf, const K& t) {
-buf.resize(sizeof t);
-return readData(reinterpret_cast(buf.data()), sizeof(t));
+ protected:
+ explicit CRCStreamBase(gsl::not_null child_stream) :
child_stream_(child_stream) {}
+ // this implementation is here to make MSVC happy, declaration would be
enough
+ // as it will never get called (and should not be called)
+ CRCStreamBase() : child_stream_(gsl::make_not_null(nullptr)) {
+assert(false);
}
- uLong crc_;
- T *child_stream_;
- bool disable_encoding_;
+ uLong crc_ = 0;
+ gsl::not_null child_stream_;
};
-template
-CRCStream::CRCStream(T *child_stream)
-: child_stream_(child_stream),
- disable_encoding_(false) {
- crc_ = crc32(0L, Z_NULL, 0);
-}
-
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #864: MINIFICPP-1319 - Stream refactor
adamdebreceni commented on a change in pull request #864:
URL: https://github.com/apache/nifi-minifi-cpp/pull/864#discussion_r486372290
##
File path: libminifi/test/unit/FileStreamTests.cpp
##
@@ -153,21 +153,21 @@ TEST_CASE("TestFileBadArgumentNoChange3", "[TestLoader]")
{
minifi::io::FileStream stream(path, 0, true);
std::vector readBuffer;
- REQUIRE(stream.readData(readBuffer, stream.getSize()) == stream.getSize());
+ REQUIRE(stream.read(readBuffer, stream.size()) == stream.size());
uint8_t* data = readBuffer.data();
REQUIRE(std::string(reinterpret_cast(data), readBuffer.size()) ==
"tempFile");
stream.seek(4);
- stream.write(nullptr, 0);
+ stream.write((const uint8_t*)nullptr, 0);
Review comment:
done, I don't remember why the cast was needed at all
##
File path: libminifi/test/unit/FileStreamTests.cpp
##
@@ -115,21 +115,21 @@ TEST_CASE("TestFileBadArgumentNoChange2", "[TestLoader]")
{
minifi::io::FileStream stream(path, 0, true);
std::vector readBuffer;
- REQUIRE(stream.readData(readBuffer, stream.getSize()) == stream.getSize());
+ REQUIRE(stream.read(readBuffer, stream.size()) == stream.size());
uint8_t* data = readBuffer.data();
REQUIRE(std::string(reinterpret_cast(data), readBuffer.size()) ==
"tempFile");
stream.seek(4);
- stream.write(nullptr, 0);
+ stream.write((const uint8_t*)nullptr, 0);
Review comment:
done
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #864: MINIFICPP-1319 - Stream refactor
adamdebreceni commented on a change in pull request #864:
URL: https://github.com/apache/nifi-minifi-cpp/pull/864#discussion_r486371426
##
File path: libminifi/include/sitetosite/Peer.h
##
@@ -345,7 +310,7 @@ class SiteToSitePeer : public
org::apache::nifi::minifi::io::BaseStream {
* Move assignment operator.
*/
SiteToSitePeer& operator=(SiteToSitePeer&& other) {
-stream_ =
std::unique_ptr(other.stream_.release());
+stream_ =
std::unique_ptr(other.stream_.release());
Review comment:
done
##
File path: libminifi/include/io/CRCStream.h
##
@@ -34,307 +35,111 @@
#endif
#include "BaseStream.h"
#include "Exception.h"
-#include "Serializable.h"
namespace org {
namespace apache {
namespace nifi {
namespace minifi {
namespace io {
+namespace internal {
-template
-class CRCStream : public BaseStream {
+template
+class CRCStreamBase : public virtual Stream {
public:
- /**
- * Raw pointer because the caller guarantees that
- * it will exceed our lifetime.
- */
- explicit CRCStream(T *child_stream);
- CRCStream(T *child_stream, uint64_t initial_crc);
-
- CRCStream(CRCStream&&) noexcept;
-
- ~CRCStream() override = default;
-
- T *getstream() const {
+ StreamType *getstream() const {
return child_stream_;
}
- void disableEncoding() {
-disable_encoding_ = true;
- }
-
- /**
- * Reads data and places it into buf
- * @param buf buffer in which we extract data
- * @param buflen
- */
- int readData(std::vector , int buflen) override;
- /**
- * Reads data and places it into buf
- * @param buf buffer in which we extract data
- * @param 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 , int buflen);
-
- /**
- * writes value to stream
- * @param value value to write
- * @param size size of value
- */
- int writeData(uint8_t *value, int size) override;
-
- using BaseStream::write;
-
- /**
- * write 4 bytes to stream
- * @param base_value non encoded value
- * @param stream output stream
- * @param is_little_endian endianness determination
- * @return resulting write size
- **/
- 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
- * @param stream output stream
- * @param is_little_endian endianness determination
- * @return resulting write size
- **/
- int write(uint16_t base_value, bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
-
- /**
- * write 8 bytes to stream
- * @param base_value non encoded value
- * @param stream output stream
- * @param is_little_endian endianness determination
- * @return resulting write size
- **/
- int write(uint64_t base_value, bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
+ void close() override { child_stream_->close(); }
- /**
- * Reads a system word
- * @param value value to write
- */
- int read(uint64_t , bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
-
- /**
- * Reads a uint32_t
- * @param value value to write
- */
- int read(uint32_t , bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
-
- /**
- * Reads a system short
- * @param value value to write
- */
- int read(uint16_t , bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
-
- const size_t getSize() const override { return child_stream_->getSize(); }
-
- void closeStream() override { child_stream_->closeStream(); }
-
- short initialize() override { // NOLINT
+ int initialize() override {
child_stream_->initialize();
reset();
return 0;
}
- void updateCRC(uint8_t *buffer, uint32_t length);
+ void updateCRC(uint8_t *buffer, uint32_t length) {
+crc_ = crc32(crc_, buffer, length);
+ }
uint64_t getCRC() {
-return crc_;
+return gsl::narrow(crc_);
}
- void reset();
-
- protected:
- /**
- * Creates a vector and returns the vector using the provided
- * type name.
- * @param t incoming object
- * @returns vector.
- */
- template
- std::vector readBuffer(const K& t) {
-std::vector buf;
-readBuffer(buf, t);
-return buf;
+ void reset() {
+crc_ = crc32(0L, Z_NULL, 0);
}
- /**
- * Populates the vector using the provided type name.
- * @param buf output buffer
- * @param t incoming object
- * @returns number of bytes read.
- */
- template
- int readBuffer(std::vector& buf, const K& t) {
-buf.resize(sizeof t);
-return readData(reinterpret_cast(buf.data()), sizeof(t));
+ protected:
+ explicit CRCStreamBase(gsl::not_null child_stream) :
child_stream_(child_stream) {}
+ // this implementation is here to make MSVC happy, declaration would be
enough
+ // as it will never get
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #864: MINIFICPP-1319 - Stream refactor
adamdebreceni commented on a change in pull request #864:
URL: https://github.com/apache/nifi-minifi-cpp/pull/864#discussion_r486277619
##
File path: libminifi/src/io/BufferStream.cpp
##
@@ -0,0 +1,52 @@
+/**
+ *
+ * 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
+#include
+#include "io/BufferStream.h"
+#include "utils/StreamUtils.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace io {
+
+int BufferStream::write(const uint8_t *value, int size) {
+ utils::internal::ensureNonNegativeWrite(size);
Review comment:
switched to `gsl_Expects` everywhere
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #864: MINIFICPP-1319 - Stream refactor
adamdebreceni commented on a change in pull request #864:
URL: https://github.com/apache/nifi-minifi-cpp/pull/864#discussion_r486277381
##
File path: libminifi/include/io/OutputStream.h
##
@@ -0,0 +1,94 @@
+/**
+ *
+ * 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.
+ */
+
+#pragma once
+
+#include
+#include
+#include
+#include "Stream.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace io {
+
+/**
+ * Serializable instances provide base functionality to
+ * write certain objects/primitives to a data stream.
+ *
+ */
+class OutputStream : public virtual Stream {
+ public:
+ /**
+ * write valueto stream
+ * @param value non encoded value
+ * @param len length of value
+ * @return resulting write size
+ **/
+ virtual int write(const uint8_t *value, int len) = 0;
+
+ int write(const std::vector& buffer, int len);
+
+ /**
+ * write bool to stream
+ * @param value non encoded value
+ * @return resulting write size
+ **/
+ int write(bool value);
+
+ /**
+ * write string to stream
+ * @param str string to write
+ * @return resulting write size
+ **/
+ int write(const std::string& str, bool widen = false);
+
+ /**
+ * write string to stream
+ * @param str string to write
+ * @return resulting write size
+ **/
+ int write(const char* str, bool widen = false);
+
+ /**
+ * writes sizeof(Integral) bytes to the stream
+ * @param value to write
+ * @return resulting write size
+ **/
+ template::value && !std::is_same::value>>
+ int write(Integral value) {
+uint8_t buffer[sizeof(Integral)]{};
+
+for (std::size_t byteIdx = 0; byteIdx < sizeof(Integral); ++byteIdx) {
+ buffer[byteIdx] = static_cast(value >> (8*(sizeof(Integral) -
1) - 8*byteIdx));
Review comment:
done
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #864: MINIFICPP-1319 - Stream refactor
adamdebreceni commented on a change in pull request #864:
URL: https://github.com/apache/nifi-minifi-cpp/pull/864#discussion_r486275648
##
File path: libminifi/include/io/CRCStream.h
##
@@ -34,307 +35,111 @@
#endif
#include "BaseStream.h"
#include "Exception.h"
-#include "Serializable.h"
namespace org {
namespace apache {
namespace nifi {
namespace minifi {
namespace io {
+namespace internal {
-template
-class CRCStream : public BaseStream {
+template
+class CRCStreamBase : public virtual Stream {
public:
- /**
- * Raw pointer because the caller guarantees that
- * it will exceed our lifetime.
- */
- explicit CRCStream(T *child_stream);
- CRCStream(T *child_stream, uint64_t initial_crc);
-
- CRCStream(CRCStream&&) noexcept;
-
- ~CRCStream() override = default;
-
- T *getstream() const {
+ StreamType *getstream() const {
return child_stream_;
}
- void disableEncoding() {
-disable_encoding_ = true;
- }
-
- /**
- * Reads data and places it into buf
- * @param buf buffer in which we extract data
- * @param buflen
- */
- int readData(std::vector , int buflen) override;
- /**
- * Reads data and places it into buf
- * @param buf buffer in which we extract data
- * @param 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 , int buflen);
-
- /**
- * writes value to stream
- * @param value value to write
- * @param size size of value
- */
- int writeData(uint8_t *value, int size) override;
-
- using BaseStream::write;
-
- /**
- * write 4 bytes to stream
- * @param base_value non encoded value
- * @param stream output stream
- * @param is_little_endian endianness determination
- * @return resulting write size
- **/
- 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
- * @param stream output stream
- * @param is_little_endian endianness determination
- * @return resulting write size
- **/
- int write(uint16_t base_value, bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
-
- /**
- * write 8 bytes to stream
- * @param base_value non encoded value
- * @param stream output stream
- * @param is_little_endian endianness determination
- * @return resulting write size
- **/
- int write(uint64_t base_value, bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
+ void close() override { child_stream_->close(); }
- /**
- * Reads a system word
- * @param value value to write
- */
- int read(uint64_t , bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
-
- /**
- * Reads a uint32_t
- * @param value value to write
- */
- int read(uint32_t , bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
-
- /**
- * Reads a system short
- * @param value value to write
- */
- int read(uint16_t , bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
-
- const size_t getSize() const override { return child_stream_->getSize(); }
-
- void closeStream() override { child_stream_->closeStream(); }
-
- short initialize() override { // NOLINT
+ int initialize() override {
child_stream_->initialize();
reset();
return 0;
}
- void updateCRC(uint8_t *buffer, uint32_t length);
+ void updateCRC(uint8_t *buffer, uint32_t length) {
+crc_ = crc32(crc_, buffer, length);
+ }
uint64_t getCRC() {
-return crc_;
+return gsl::narrow(crc_);
}
- void reset();
-
- protected:
- /**
- * Creates a vector and returns the vector using the provided
- * type name.
- * @param t incoming object
- * @returns vector.
- */
- template
- std::vector readBuffer(const K& t) {
-std::vector buf;
-readBuffer(buf, t);
-return buf;
+ void reset() {
+crc_ = crc32(0L, Z_NULL, 0);
}
- /**
- * Populates the vector using the provided type name.
- * @param buf output buffer
- * @param t incoming object
- * @returns number of bytes read.
- */
- template
- int readBuffer(std::vector& buf, const K& t) {
-buf.resize(sizeof t);
-return readData(reinterpret_cast(buf.data()), sizeof(t));
+ protected:
+ explicit CRCStreamBase(gsl::not_null child_stream) :
child_stream_(child_stream) {}
+ // this implementation is here to make MSVC happy, declaration would be
enough
+ // as it will never get called (and should not be called)
+ CRCStreamBase() : child_stream_(gsl::make_not_null(nullptr)) {
+assert(false);
Review comment:
I am not sure what the standard-compliant behavior is here, but gcc and
clang both accept the following, whereas msvc complained for the undefined
reference `A::A()`
```
struct A {
A(); // is a declaration enough?
A(int x) : x(x) {}
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #864: MINIFICPP-1319 - Stream refactor
adamdebreceni commented on a change in pull request #864:
URL: https://github.com/apache/nifi-minifi-cpp/pull/864#discussion_r486275648
##
File path: libminifi/include/io/CRCStream.h
##
@@ -34,307 +35,111 @@
#endif
#include "BaseStream.h"
#include "Exception.h"
-#include "Serializable.h"
namespace org {
namespace apache {
namespace nifi {
namespace minifi {
namespace io {
+namespace internal {
-template
-class CRCStream : public BaseStream {
+template
+class CRCStreamBase : public virtual Stream {
public:
- /**
- * Raw pointer because the caller guarantees that
- * it will exceed our lifetime.
- */
- explicit CRCStream(T *child_stream);
- CRCStream(T *child_stream, uint64_t initial_crc);
-
- CRCStream(CRCStream&&) noexcept;
-
- ~CRCStream() override = default;
-
- T *getstream() const {
+ StreamType *getstream() const {
return child_stream_;
}
- void disableEncoding() {
-disable_encoding_ = true;
- }
-
- /**
- * Reads data and places it into buf
- * @param buf buffer in which we extract data
- * @param buflen
- */
- int readData(std::vector , int buflen) override;
- /**
- * Reads data and places it into buf
- * @param buf buffer in which we extract data
- * @param 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 , int buflen);
-
- /**
- * writes value to stream
- * @param value value to write
- * @param size size of value
- */
- int writeData(uint8_t *value, int size) override;
-
- using BaseStream::write;
-
- /**
- * write 4 bytes to stream
- * @param base_value non encoded value
- * @param stream output stream
- * @param is_little_endian endianness determination
- * @return resulting write size
- **/
- 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
- * @param stream output stream
- * @param is_little_endian endianness determination
- * @return resulting write size
- **/
- int write(uint16_t base_value, bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
-
- /**
- * write 8 bytes to stream
- * @param base_value non encoded value
- * @param stream output stream
- * @param is_little_endian endianness determination
- * @return resulting write size
- **/
- int write(uint64_t base_value, bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
+ void close() override { child_stream_->close(); }
- /**
- * Reads a system word
- * @param value value to write
- */
- int read(uint64_t , bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
-
- /**
- * Reads a uint32_t
- * @param value value to write
- */
- int read(uint32_t , bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
-
- /**
- * Reads a system short
- * @param value value to write
- */
- int read(uint16_t , bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
-
- const size_t getSize() const override { return child_stream_->getSize(); }
-
- void closeStream() override { child_stream_->closeStream(); }
-
- short initialize() override { // NOLINT
+ int initialize() override {
child_stream_->initialize();
reset();
return 0;
}
- void updateCRC(uint8_t *buffer, uint32_t length);
+ void updateCRC(uint8_t *buffer, uint32_t length) {
+crc_ = crc32(crc_, buffer, length);
+ }
uint64_t getCRC() {
-return crc_;
+return gsl::narrow(crc_);
}
- void reset();
-
- protected:
- /**
- * Creates a vector and returns the vector using the provided
- * type name.
- * @param t incoming object
- * @returns vector.
- */
- template
- std::vector readBuffer(const K& t) {
-std::vector buf;
-readBuffer(buf, t);
-return buf;
+ void reset() {
+crc_ = crc32(0L, Z_NULL, 0);
}
- /**
- * Populates the vector using the provided type name.
- * @param buf output buffer
- * @param t incoming object
- * @returns number of bytes read.
- */
- template
- int readBuffer(std::vector& buf, const K& t) {
-buf.resize(sizeof t);
-return readData(reinterpret_cast(buf.data()), sizeof(t));
+ protected:
+ explicit CRCStreamBase(gsl::not_null child_stream) :
child_stream_(child_stream) {}
+ // this implementation is here to make MSVC happy, declaration would be
enough
+ // as it will never get called (and should not be called)
+ CRCStreamBase() : child_stream_(gsl::make_not_null(nullptr)) {
+assert(false);
Review comment:
I am not sure what the standard-compliant behavior is here, but gcc and
clang both accept the following, whereas msvc complained for the undefined
reference `A::A()`
```
struct A {
A(); // is a declaration enough?
A(int x) : x(x) {}
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #864: MINIFICPP-1319 - Stream refactor
adamdebreceni commented on a change in pull request #864:
URL: https://github.com/apache/nifi-minifi-cpp/pull/864#discussion_r486275648
##
File path: libminifi/include/io/CRCStream.h
##
@@ -34,307 +35,111 @@
#endif
#include "BaseStream.h"
#include "Exception.h"
-#include "Serializable.h"
namespace org {
namespace apache {
namespace nifi {
namespace minifi {
namespace io {
+namespace internal {
-template
-class CRCStream : public BaseStream {
+template
+class CRCStreamBase : public virtual Stream {
public:
- /**
- * Raw pointer because the caller guarantees that
- * it will exceed our lifetime.
- */
- explicit CRCStream(T *child_stream);
- CRCStream(T *child_stream, uint64_t initial_crc);
-
- CRCStream(CRCStream&&) noexcept;
-
- ~CRCStream() override = default;
-
- T *getstream() const {
+ StreamType *getstream() const {
return child_stream_;
}
- void disableEncoding() {
-disable_encoding_ = true;
- }
-
- /**
- * Reads data and places it into buf
- * @param buf buffer in which we extract data
- * @param buflen
- */
- int readData(std::vector , int buflen) override;
- /**
- * Reads data and places it into buf
- * @param buf buffer in which we extract data
- * @param 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 , int buflen);
-
- /**
- * writes value to stream
- * @param value value to write
- * @param size size of value
- */
- int writeData(uint8_t *value, int size) override;
-
- using BaseStream::write;
-
- /**
- * write 4 bytes to stream
- * @param base_value non encoded value
- * @param stream output stream
- * @param is_little_endian endianness determination
- * @return resulting write size
- **/
- 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
- * @param stream output stream
- * @param is_little_endian endianness determination
- * @return resulting write size
- **/
- int write(uint16_t base_value, bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
-
- /**
- * write 8 bytes to stream
- * @param base_value non encoded value
- * @param stream output stream
- * @param is_little_endian endianness determination
- * @return resulting write size
- **/
- int write(uint64_t base_value, bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
+ void close() override { child_stream_->close(); }
- /**
- * Reads a system word
- * @param value value to write
- */
- int read(uint64_t , bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
-
- /**
- * Reads a uint32_t
- * @param value value to write
- */
- int read(uint32_t , bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
-
- /**
- * Reads a system short
- * @param value value to write
- */
- int read(uint16_t , bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
-
- const size_t getSize() const override { return child_stream_->getSize(); }
-
- void closeStream() override { child_stream_->closeStream(); }
-
- short initialize() override { // NOLINT
+ int initialize() override {
child_stream_->initialize();
reset();
return 0;
}
- void updateCRC(uint8_t *buffer, uint32_t length);
+ void updateCRC(uint8_t *buffer, uint32_t length) {
+crc_ = crc32(crc_, buffer, length);
+ }
uint64_t getCRC() {
-return crc_;
+return gsl::narrow(crc_);
}
- void reset();
-
- protected:
- /**
- * Creates a vector and returns the vector using the provided
- * type name.
- * @param t incoming object
- * @returns vector.
- */
- template
- std::vector readBuffer(const K& t) {
-std::vector buf;
-readBuffer(buf, t);
-return buf;
+ void reset() {
+crc_ = crc32(0L, Z_NULL, 0);
}
- /**
- * Populates the vector using the provided type name.
- * @param buf output buffer
- * @param t incoming object
- * @returns number of bytes read.
- */
- template
- int readBuffer(std::vector& buf, const K& t) {
-buf.resize(sizeof t);
-return readData(reinterpret_cast(buf.data()), sizeof(t));
+ protected:
+ explicit CRCStreamBase(gsl::not_null child_stream) :
child_stream_(child_stream) {}
+ // this implementation is here to make MSVC happy, declaration would be
enough
+ // as it will never get called (and should not be called)
+ CRCStreamBase() : child_stream_(gsl::make_not_null(nullptr)) {
+assert(false);
Review comment:
I am not sure what the standard-compliant behavior is here, but gcc and
clang both accept the following, but msvc complained for the undefined
reference `A::A()`
```
struct A {
A(); // is a declaration enough?
A(int x) : x(x) {}
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #864: MINIFICPP-1319 - Stream refactor
adamdebreceni commented on a change in pull request #864:
URL: https://github.com/apache/nifi-minifi-cpp/pull/864#discussion_r486275648
##
File path: libminifi/include/io/CRCStream.h
##
@@ -34,307 +35,111 @@
#endif
#include "BaseStream.h"
#include "Exception.h"
-#include "Serializable.h"
namespace org {
namespace apache {
namespace nifi {
namespace minifi {
namespace io {
+namespace internal {
-template
-class CRCStream : public BaseStream {
+template
+class CRCStreamBase : public virtual Stream {
public:
- /**
- * Raw pointer because the caller guarantees that
- * it will exceed our lifetime.
- */
- explicit CRCStream(T *child_stream);
- CRCStream(T *child_stream, uint64_t initial_crc);
-
- CRCStream(CRCStream&&) noexcept;
-
- ~CRCStream() override = default;
-
- T *getstream() const {
+ StreamType *getstream() const {
return child_stream_;
}
- void disableEncoding() {
-disable_encoding_ = true;
- }
-
- /**
- * Reads data and places it into buf
- * @param buf buffer in which we extract data
- * @param buflen
- */
- int readData(std::vector , int buflen) override;
- /**
- * Reads data and places it into buf
- * @param buf buffer in which we extract data
- * @param 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 , int buflen);
-
- /**
- * writes value to stream
- * @param value value to write
- * @param size size of value
- */
- int writeData(uint8_t *value, int size) override;
-
- using BaseStream::write;
-
- /**
- * write 4 bytes to stream
- * @param base_value non encoded value
- * @param stream output stream
- * @param is_little_endian endianness determination
- * @return resulting write size
- **/
- 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
- * @param stream output stream
- * @param is_little_endian endianness determination
- * @return resulting write size
- **/
- int write(uint16_t base_value, bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
-
- /**
- * write 8 bytes to stream
- * @param base_value non encoded value
- * @param stream output stream
- * @param is_little_endian endianness determination
- * @return resulting write size
- **/
- int write(uint64_t base_value, bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
+ void close() override { child_stream_->close(); }
- /**
- * Reads a system word
- * @param value value to write
- */
- int read(uint64_t , bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
-
- /**
- * Reads a uint32_t
- * @param value value to write
- */
- int read(uint32_t , bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
-
- /**
- * Reads a system short
- * @param value value to write
- */
- int read(uint16_t , bool is_little_endian =
EndiannessCheck::IS_LITTLE) override;
-
- const size_t getSize() const override { return child_stream_->getSize(); }
-
- void closeStream() override { child_stream_->closeStream(); }
-
- short initialize() override { // NOLINT
+ int initialize() override {
child_stream_->initialize();
reset();
return 0;
}
- void updateCRC(uint8_t *buffer, uint32_t length);
+ void updateCRC(uint8_t *buffer, uint32_t length) {
+crc_ = crc32(crc_, buffer, length);
+ }
uint64_t getCRC() {
-return crc_;
+return gsl::narrow(crc_);
}
- void reset();
-
- protected:
- /**
- * Creates a vector and returns the vector using the provided
- * type name.
- * @param t incoming object
- * @returns vector.
- */
- template
- std::vector readBuffer(const K& t) {
-std::vector buf;
-readBuffer(buf, t);
-return buf;
+ void reset() {
+crc_ = crc32(0L, Z_NULL, 0);
}
- /**
- * Populates the vector using the provided type name.
- * @param buf output buffer
- * @param t incoming object
- * @returns number of bytes read.
- */
- template
- int readBuffer(std::vector& buf, const K& t) {
-buf.resize(sizeof t);
-return readData(reinterpret_cast(buf.data()), sizeof(t));
+ protected:
+ explicit CRCStreamBase(gsl::not_null child_stream) :
child_stream_(child_stream) {}
+ // this implementation is here to make MSVC happy, declaration would be
enough
+ // as it will never get called (and should not be called)
+ CRCStreamBase() : child_stream_(gsl::make_not_null(nullptr)) {
+assert(false);
Review comment:
I am not sure what is the standard-compliant behavior here, but gcc and
clang both accept the following, but msvc complained for the undefined
reference `A::A()`
```
struct A {
A(); // is a declaration enough?
A(int x) : x(x) {}
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #864: MINIFICPP-1319 - Stream refactor
adamdebreceni commented on a change in pull request #864: URL: https://github.com/apache/nifi-minifi-cpp/pull/864#discussion_r485646889 ## File path: libminifi/src/sitetosite/SiteToSiteClient.cpp ## @@ -499,8 +499,7 @@ int16_t SiteToSiteClient::send(std::string transactionID, DataPacket *packet, co return -1; } -ret = transaction->getStream().writeData(reinterpret_cast(const_cast(packet->payload_.c_str())), - gsl::narrow(len)); +ret = transaction->getStream().write(reinterpret_cast(const_cast(packet->payload_.c_str())), len); Review comment: this was possibly caused by an unsuccessful rebase This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #864: MINIFICPP-1319 - Stream refactor
adamdebreceni commented on a change in pull request #864:
URL: https://github.com/apache/nifi-minifi-cpp/pull/864#discussion_r470474756
##
File path: libminifi/include/io/ZlibStream.h
##
@@ -62,12 +61,12 @@ class ZlibBaseStream : public BaseStream {
ZlibStreamState state_{ZlibStreamState::UNINITIALIZED};
z_stream strm_{};
std::vector outputBuffer_;
+ OutputStream* output_;
Review comment:
did it both for the ZlibStream and CRCStream
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #864: MINIFICPP-1319 - Stream refactor
adamdebreceni commented on a change in pull request #864:
URL: https://github.com/apache/nifi-minifi-cpp/pull/864#discussion_r470474756
##
File path: libminifi/include/io/ZlibStream.h
##
@@ -62,12 +61,12 @@ class ZlibBaseStream : public BaseStream {
ZlibStreamState state_{ZlibStreamState::UNINITIALIZED};
z_stream strm_{};
std::vector outputBuffer_;
+ OutputStream* output_;
Review comment:
did it for both the ZlibStream and CRCStream
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #864: MINIFICPP-1319 - Stream refactor
adamdebreceni commented on a change in pull request #864:
URL: https://github.com/apache/nifi-minifi-cpp/pull/864#discussion_r470035889
##
File path: libminifi/include/io/Stream.h
##
@@ -0,0 +1,52 @@
+/**
+ *
+ * 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.
+ */
+
+#pragma once
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace io {
+
+/**
+ * All streams serialize/deserialize in big-endian
+ */
+class Stream {
+ public:
+ virtual void close() {}
+
+ virtual void seek(uint64_t offset) {
+throw std::runtime_error("Seek is not supported");
+ }
+
+ virtual int initialize() {
+return 1;
+ }
+
+ virtual const uint8_t* getBuffer() const {
+throw std::runtime_error("Not a buffered stream");
+ }
+ virtual ~Stream() = default;
+};
Review comment:
I'm not entirely sure where we should put them, for example
`FileOutputCallback::process` takes a `BaseStream` and then calls `getBuffer`
(which only makes sense for a `BufferStream`), if we would like to change the
method, that it can only take a `BufferStream`, then we are up for a (not
necessary infeasible) ride, I moved these methods back and forth between this
Stream class and the BaseStream, `ZlibCompressStream` is only an `OutputStream`
but it does implement a non-trivial `close` method, that's why they settled in
`Stream`, also in the future `content_repo->write(claim)` should only return an
`OutputStream`, not a `BaseStream`, and as it stands we do call `seek` on the
result
I think we should investigate further with your proposed "granularization"
in a separate PR
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #864: MINIFICPP-1319 - Stream refactor
adamdebreceni commented on a change in pull request #864:
URL: https://github.com/apache/nifi-minifi-cpp/pull/864#discussion_r469966242
##
File path: libminifi/include/io/OutputStream.h
##
@@ -0,0 +1,107 @@
+/**
+ *
+ * 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.
+ */
+
+#pragma once
+
+#include
+#include
+#include
+#include "Stream.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace io {
+
+/**
+ * Serializable instances provide base functionality to
+ * write certain objects/primitives to a data stream.
+ *
+ */
+class OutputStream : public virtual Stream {
+ public:
+ /**
+ * write valueto stream
+ * @param value non encoded value
+ * @param len length of value
+ * @return resulting write size
+ **/
+ virtual int write(const uint8_t *value, int len) {
+throw std::runtime_error("Stream is not writable");
Review comment:
made it pure virtual (InputStream::read as well)
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #864: MINIFICPP-1319 - Stream refactor
adamdebreceni commented on a change in pull request #864:
URL: https://github.com/apache/nifi-minifi-cpp/pull/864#discussion_r469928037
##
File path: extensions/coap/tests/CoapC2VerifyHeartbeat.cpp
##
@@ -132,18 +132,18 @@ class VerifyCoAPServer : public CoapIntegrationBase {
{
// should result in valid operation
- minifi::io::BaseStream stream;
+ minifi::io::BufferStream stream;
uint16_t version = 0, size = 1;
uint8_t operation = 1;
stream.write(version);
stream.write(size);
stream.write(, 1);
- stream.writeUTF("id");
- stream.writeUTF("operand");
+ stream.write("id");
+ stream.write("operand");
- uint8_t *data = new uint8_t[stream.getSize()];
- memcpy(data, stream.getBuffer(), stream.getSize());
- minifi::coap::CoapResponse response(205, std::unique_ptr(data),
stream.getSize());
+ uint8_t *data = new uint8_t[stream.size()];
Review comment:
yep it does not, but `unique_ptr` and move works, just taking an array
as `std::unique_ptr` is... not great, so I will look into it
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #864: MINIFICPP-1319 - Stream refactor
adamdebreceni commented on a change in pull request #864:
URL: https://github.com/apache/nifi-minifi-cpp/pull/864#discussion_r469903259
##
File path: libminifi/src/io/BufferStream.cpp
##
@@ -0,0 +1,56 @@
+/**
+ *
+ * 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
+#include
+#include
+#include
Review comment:
done
##
File path: libminifi/include/io/OutputStream.h
##
@@ -0,0 +1,107 @@
+/**
+ *
+ * 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.
+ */
+
+#pragma once
+
+#include
+#include
+#include
+#include "Stream.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace io {
+
+/**
+ * Serializable instances provide base functionality to
+ * write certain objects/primitives to a data stream.
+ *
+ */
+class OutputStream : public virtual Stream {
+ public:
+ /**
+ * write valueto stream
+ * @param value non encoded value
+ * @param len length of value
+ * @return resulting write size
+ **/
+ virtual int write(const uint8_t *value, int len) {
+throw std::runtime_error("Stream is not writable");
Review comment:
done
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #864: MINIFICPP-1319 - Stream refactor
adamdebreceni commented on a change in pull request #864:
URL: https://github.com/apache/nifi-minifi-cpp/pull/864#discussion_r469902822
##
File path: libminifi/src/io/InputStream.cpp
##
@@ -0,0 +1,141 @@
+/**
+ *
+ * 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
+#include
+#include
+#include
+#include
+#include "io/InputStream.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace io {
+
+int InputStream::read(std::vector& buffer, int len) {
+ if (buffer.size() < len) {
+buffer.resize(len);
+ }
+ int ret = read(buffer.data(), len);
+ buffer.resize((std::max)(ret, 0));
+ return ret;
+}
+
+
+int InputStream::read(uint8_t ) {
+ uint8_t buf = 0;
+
+ if (read(, 1) != 1) {
+return -1;
+ }
+ value = buf;
+ return 1;
+}
+
+int InputStream::read(bool ) {
+ uint8_t buf = 0;
+
+ if (read(, 1) != 1) {
+return -1;
+ }
+ value = buf;
+ return 1;
+}
+
+int InputStream::read(std::string , bool widen) {
+ uint32_t len = 0;
+ int ret = 0;
+ if (!widen) {
+uint16_t shortLength = 0;
+ret = read(shortLength);
+len = shortLength;
+ } else {
+ret = read(len);
+ }
+
+ if (ret <= 0) {
+return ret;
+ }
+
+ if (len == 0) {
+str = "";
+return ret;
+ }
+
+ std::vector buffer(len);
+ if (read(buffer.data(), len) != len) {
+return -1;
+ }
+
+ str = std::string(reinterpret_cast(buffer.data()), len);
+ return ret + len;
+}
+
+int InputStream::read(uint64_t ) {
+ uint8_t buf[8]{};
+ if (read(buf, 8) != 8) {
+return -1;
+ }
+
+ value = 0;
+ value |= (uint64_t) buf[0] << 56;
Review comment:
done, (and put back the previous template for reading/writing unsigned
integrals)
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #864: MINIFICPP-1319 - Stream refactor
adamdebreceni commented on a change in pull request #864:
URL: https://github.com/apache/nifi-minifi-cpp/pull/864#discussion_r469902504
##
File path: libminifi/src/io/OutputStream.cpp
##
@@ -0,0 +1,118 @@
+/**
+ *
+ * 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
+#include
+#include
+#include
+#include
+#include
+#include "io/OutputStream.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace io {
+
+int OutputStream::write(const std::vector& buffer, int len) {
+ if (buffer.size() < len) {
+return -1;
+ }
+ return write(buffer.data(), len);
+}
+
+int OutputStream::write(uint8_t value) {
+ return write(, 1);
+}
+
+int OutputStream::write(bool value) {
+ uint8_t temp = value;
+ return write(, 1);
+}
+
+int OutputStream::write(const std::string& str, bool widen) {
+ return write_str(str.c_str(), str.length(), widen);
+}
+
+int OutputStream::write(const char* str, bool widen) {
+ return write_str(str, std::strlen(str), widen);
+}
+
+int OutputStream::write_str(const char* str, uint32_t len, bool widen) {
+ int ret = 0;
+ if (!widen) {
+uint16_t shortLen = len;
+if (len != shortLen) {
+ return -1;
+}
+ret = write(shortLen);
+ } else {
+ret = write(len);
+ }
+
+ if (ret <= 0) {
+return ret;
+ }
+
+ if (len == 0) {
+return ret;
+ }
+
+ return ret + write(reinterpret_cast(str), len);
+}
+
+int OutputStream::write(uint16_t value) {
+ uint8_t buffer[2]{};
+
+ buffer[0] = value >> 8;
+ buffer[1] = value;
+
+ return write(buffer, 2);
+}
+
+int OutputStream::write(uint32_t value) {
+ uint8_t buffer[4]{};
+
+ buffer[0] = value >> 24;
+ buffer[1] = value >> 16;
+ buffer[2] = value >> 8;
+ buffer[3] = value;
+
+ return write(buffer, 4);
+}
+
+int OutputStream::write(uint64_t value) {
+ uint8_t buffer[8]{};
+
+ buffer[0] = value >> 56;
Review comment:
didn't find the warnings, but added explicit cast
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #864: MINIFICPP-1319 - Stream refactor
adamdebreceni commented on a change in pull request #864:
URL: https://github.com/apache/nifi-minifi-cpp/pull/864#discussion_r469901701
##
File path: libminifi/test/unit/FileStreamTests.cpp
##
@@ -235,12 +235,12 @@ TEST_CASE("TestFileExceedSize", "[TestLoader]") {
std::vector verifybuffer;
for (int i = 0; i < 10; i++)
-REQUIRE(stream.readData(verifybuffer, 8192) == 8192);
- REQUIRE(stream.readData(verifybuffer, 8192) == 0);
+REQUIRE(stream.read(verifybuffer, 8192) == 8192);
Review comment:
done
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
