fixed consumer queue - working

Author: nairis

examples/CMakeLists.txt

@@ -39,50 +39,29 @@ include_directories(${CPP_REDIS_INCLUDES})
 ###
 link_directories(${DEPS_LIBRARIES})
 
-
-###
-# executable
-###
-add_executable(cpp_redis_client cpp_redis_client.cpp)
-target_link_libraries(cpp_redis_client cpp_redis)
-
-add_executable(cpp_redis_consumer cpp_redis_consumer.cpp)
-target_link_libraries(cpp_redis_consumer cpp_redis)
-
-add_executable(cpp_redis_future_client cpp_redis_future_client.cpp)
-target_link_libraries(cpp_redis_future_client cpp_redis)
-
-add_executable(cpp_redis_subscriber cpp_redis_subscriber.cpp)
-target_link_libraries(cpp_redis_subscriber cpp_redis)
-
-add_executable(cpp_redis_logger cpp_redis_logger.cpp)
-target_link_libraries(cpp_redis_logger cpp_redis)
-
-add_executable(cpp_redis_kill cpp_redis_kill.cpp)
-target_link_libraries(cpp_redis_kill cpp_redis)
-
-add_executable(cpp_redis_streams_client cpp_redis_streams_client.cpp)
-target_link_libraries(cpp_redis_streams_client cpp_redis)
-
-add_executable(cpp_redis_high_availability_client cpp_redis_high_availability_client.cpp)
-target_link_libraries(cpp_redis_high_availability_client cpp_redis)
-
-
-###
-# link libs
-###
-if(WIN32)
-  target_link_libraries(cpp_redis_client ws2_32)
-  target_link_libraries(cpp_redis_future_client ws2_32)
-  target_link_libraries(cpp_redis_subscriber ws2_32)
-  target_link_libraries(cpp_redis_logger ws2_32)
-  target_link_libraries(cpp_redis_kill ws2_32)
-  target_link_libraries(cpp_redis_high_availability_client ws2_32)
-else()
-  target_link_libraries(cpp_redis_client pthread)
-  target_link_libraries(cpp_redis_future_client pthread)
-  target_link_libraries(cpp_redis_subscriber pthread)
-  target_link_libraries(cpp_redis_logger pthread)
-  target_link_libraries(cpp_redis_kill pthread)
-  target_link_libraries(cpp_redis_high_availability_client pthread)
-endif(WIN32)
+set(EXAMPLES cpp_redis_client
+              cpp_redis_consumer
+              cpp_redis_future_client
+              cpp_redis_subscriber
+              cpp_redis_logger
+              cpp_redis_kill
+              cpp_redis_streams_client
+              cpp_redis_high_availability_client
+    )
+
+foreach(EXAMPLE IN ITEMS ${EXAMPLES})
+  ###
+  # executable
+  ###
+  add_executable(${EXAMPLE} ${EXAMPLE}.cpp)
+  target_link_libraries(${EXAMPLE} cpp_redis)
+
+  ###
+  # link libs
+  ###
+  if(WIN32)
+    target_link_libraries(${EXAMPLE} ws2_32)
+  else()
+    target_link_libraries(${EXAMPLE} pthread)
+  endif(WIN32)
+endforeach(EXAMPLE)

includes/cpp_redis/builders/builder_iface.hpp

@@ -39,26 +39,26 @@ namespace cpp_redis {
 		public:
 				virtual ~builder_iface() = default;
 
-/**
- * take data as parameter which is consumed to build the reply
- * every bytes used to build the reply must be removed from the buffer passed as parameter
- *
- * @param data data to be consumed
- * @return current instance
- *
- */
+				/**
+				 * take data as parameter which is consumed to build the reply
+				 * every bytes used to build the reply must be removed from the buffer passed as parameter
+				 *
+				 * @param data data to be consumed
+				 * @return current instance
+				 *
+				 */
 				virtual builder_iface &operator<<(std::string &data) = 0;
 
-/**
- * @return whether the reply could be built
- *
- */
+				/**
+				 * @return whether the reply could be built
+				 *
+				 */
 				virtual bool reply_ready() const = 0;
 
-/**
- * @return reply object
- *
- */
+				/**
+				 * @return reply object
+				 *
+				 */
 				virtual reply get_reply() const = 0;
 		};
 

includes/cpp_redis/builders/builders_factory.hpp

@@ -28,22 +28,21 @@
 
 namespace cpp_redis {
 
-namespace builders {
-
-/**
- * create a builder corresponding to the given id
- *  * + for simple strings
- *  * - for errors
- *  * : for integers
- *  * $ for bulk strings
- *  * * for arrays
- *
- * @param id char that determines which builder to return
- * @return new builder instance depending on id value
- */
-//!
-std::unique_ptr<builder_iface> create_builder(char id);
-
-} // namespace builders
+	namespace builders {
+
+		/**
+		 * create a builder corresponding to the given id
+		 *  * + for simple strings
+		 *  * - for errors
+		 *  * : for integers
+		 *  * $ for bulk strings
+		 *  * * for arrays
+		 *
+		 * @param id char that determines which builder to return
+		 * @return new builder instance depending on id value
+		 */
+		std::unique_ptr<builder_iface> create_builder(char id);
+
+	} // namespace builders
 
 } // namespace cpp_redis

includes/cpp_redis/builders/bulk_string_builder.hpp

@@ -36,9 +36,9 @@ namespace builders {
 class bulk_string_builder : public builder_iface {
 public:
 //! ctor
-  bulk_string_builder(void);
+  bulk_string_builder();
 //! dtor
-  ~bulk_string_builder(void) = default;
+  ~bulk_string_builder() override = default;
 
 //! copy ctor
   bulk_string_builder(const bulk_string_builder&) = delete;
@@ -53,30 +53,30 @@ class bulk_string_builder : public builder_iface {
 //! @param data data to be consumed
 //! @return current instance
 //!
-  builder_iface& operator<<(std::string& data);
+  builder_iface& operator<<(std::string& data) override;
 
 //!
 //! @return whether the reply could be built
 //!
-  bool reply_ready(void) const;
+  bool reply_ready() const override;
 
 //!
 //! @return reply object
 //!
-  reply get_reply(void) const;
+  reply get_reply() const override;
 
 //!
 //! @return the parsed bulk string
 //!
-  const std::string& get_bulk_string(void) const;
+  const std::string& get_bulk_string() const;
 
 //!
 //! @return whether the bulk string is null
 //!
-  bool is_null(void) const;
+  bool is_null() const;
 
 private:
-  void build_reply(void);
+  void build_reply();
   bool fetch_size(std::string& str);
   void fetch_str(std::string& str);
 

includes/cpp_redis/builders/error_builder.hpp

@@ -40,12 +40,12 @@ class error_builder : public builder_iface {
  * ctor
  *
  */
-  error_builder(void) = default;
+  error_builder() = default;
 /**
  * dtor
  *
  */
-  ~error_builder(void) = default;
+  ~error_builder() override = default;
 
 /**
  * copy ctor
@@ -67,25 +67,25 @@ class error_builder : public builder_iface {
  * @return current instance
  *
  */
-  builder_iface& operator<<(std::string& data);
+  builder_iface& operator<<(std::string& data) override;
 
 /**
  * @return whether the reply could be built
  *
  */
-  bool reply_ready(void) const;
+  bool reply_ready() const override;
 
 /**
  * @return reply object
  *
  */
-  reply get_reply(void) const;
+  reply get_reply() const override;
 
 /**
  * @return the parsed error
  *
  */
-  const std::string& get_error(void) const;
+  const std::string& get_error() const;
 
 private:
 /**

includes/cpp_redis/builders/integer_builder.hpp

@@ -25,7 +25,7 @@
 #include <cpp_redis/builders/builder_iface.hpp>
 #include <cpp_redis/core/reply.hpp>
 
-#include <stdint.h>
+#include <cstdint>
 
 namespace cpp_redis {
 
@@ -41,12 +41,12 @@ class integer_builder : public builder_iface {
  * ctor
  *
  */
-  integer_builder(void);
+  integer_builder();
 /**
  * dtor
  *
  */
-  ~integer_builder(void) = default;
+  ~integer_builder() override = default;
 
 /**
  * copy ctor
@@ -68,25 +68,25 @@ class integer_builder : public builder_iface {
  * @return current instance
  *
  */
-  builder_iface& operator<<(std::string& data);
+  builder_iface& operator<<(std::string& data) override;
 
 /**
  * @return whether the reply could be built
  *
  */
-  bool reply_ready(void) const;
+  bool reply_ready() const override;
 
 /**
  * @return reply object
  *
  */
-  reply get_reply(void) const;
+  reply get_reply() const override;
 
 /**
  * @return the parsed integer
  *
  */
-  int64_t get_integer(void) const;
+  int64_t get_integer() const;
 
 private:
 /**

includes/cpp_redis/builders/simple_string_builder.hpp

@@ -32,7 +32,7 @@ namespace cpp_redis {
 namespace builders {
 
 /**
- * builder to build redis simplestring replies
+ * builder to build redis simple string replies
  *
  */
 class simple_string_builder : public builder_iface {

includes/cpp_redis/core/consumer.hpp

@@ -42,6 +42,11 @@ namespace cpp_redis {
 			acknowledgement_callback_t acknowledgement_callback;
 	} consumer_callback_container_t;
 
+	typedef struct consumer_reply {
+			std::string group_id;
+			xstream_reply_t reply;
+	} consumer_reply_t;
+
 	class consumer_client_container {
 	public:
 			consumer_client_container();
@@ -67,10 +72,6 @@ namespace cpp_redis {
 			                    const consumer_callback_t &consumer_callback,
 			                    const acknowledgement_callback_t &acknowledgement_callback = nullptr);
 
-			void process();
-
-			void poll();
-
 //! @brief Connect to redis server
 //! @param host host to be connected to
 //! @param port port to be connected to
@@ -86,14 +87,6 @@ namespace cpp_redis {
 					std::int32_t max_reconnects = 0,
 					std::uint32_t reconnect_interval_ms = 0);
 
-			void read_group_handler(const xreadgroup_options_t &a);
-
-			void push_reply(const reply_t &reply);
-
-			reply_t pop_reply();
-
-			void stream_reply_handler(const xstream_reply& streams) __attribute__((optimize(0)));
-
 //!
 //! commit pipelined transaction
 //! that is, send to the network all commands pipelined by calling send() / subscribe() / ...
@@ -104,6 +97,9 @@ namespace cpp_redis {
 
 			void dispatch_changed_handler(size_t size);
 
+	private:
+			void poll();
+
 	private:
 			std::string m_stream;
 			std::string m_name;
@@ -119,16 +115,6 @@ namespace cpp_redis {
 			std::condition_variable dispatch_queue_changed;
 			std::mutex dispatch_queue_changed_mutex;
 
-			std::mutex m_replies_mutex;
-			std::queue<reply_t> m_replies;
-
-//! Whether there are new replies on the queue
-			std::atomic_bool replies_empty{true};
-
-//! Used to notify the processing queue
-			std::condition_variable replies_changed;
-			std::mutex replies_changed_mutex;
-
 			bool is_ready = false;
 			std::atomic_bool m_should_read_pending{true};
 	};

includes/cpp_redis/core/reply.hpp

@@ -270,12 +270,6 @@ namespace cpp_redis {
  */
 			type get_type() const;
 
-/**
- * support for output
- *
- */
-			friend std::ostream &operator<<(std::ostream &os, const reply &reply);
-
 	private:
 			type m_type;
 			std::vector<cpp_redis::reply> m_rows;