Skip to content

Commit 64b479a

Browse files
nairisnairis
committed
fixed consumer queue - working
1 parent 43d116c commit 64b479a

22 files changed

+594
-1330
lines changed

‎examples/CMakeLists.txt‎

Lines changed: 26 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -39,50 +39,29 @@ include_directories(${CPP_REDIS_INCLUDES})
3939
###
4040
link_directories(${DEPS_LIBRARIES})
4141

42-
43-
###
44-
# executable
45-
###
46-
add_executable(cpp_redis_client cpp_redis_client.cpp)
47-
target_link_libraries(cpp_redis_client cpp_redis)
48-
49-
add_executable(cpp_redis_consumer cpp_redis_consumer.cpp)
50-
target_link_libraries(cpp_redis_consumer cpp_redis)
51-
52-
add_executable(cpp_redis_future_client cpp_redis_future_client.cpp)
53-
target_link_libraries(cpp_redis_future_client cpp_redis)
54-
55-
add_executable(cpp_redis_subscriber cpp_redis_subscriber.cpp)
56-
target_link_libraries(cpp_redis_subscriber cpp_redis)
57-
58-
add_executable(cpp_redis_logger cpp_redis_logger.cpp)
59-
target_link_libraries(cpp_redis_logger cpp_redis)
60-
61-
add_executable(cpp_redis_kill cpp_redis_kill.cpp)
62-
target_link_libraries(cpp_redis_kill cpp_redis)
63-
64-
add_executable(cpp_redis_streams_client cpp_redis_streams_client.cpp)
65-
target_link_libraries(cpp_redis_streams_client cpp_redis)
66-
67-
add_executable(cpp_redis_high_availability_client cpp_redis_high_availability_client.cpp)
68-
target_link_libraries(cpp_redis_high_availability_client cpp_redis)
69-
70-
71-
###
72-
# link libs
73-
###
74-
if(WIN32)
75-
target_link_libraries(cpp_redis_client ws2_32)
76-
target_link_libraries(cpp_redis_future_client ws2_32)
77-
target_link_libraries(cpp_redis_subscriber ws2_32)
78-
target_link_libraries(cpp_redis_logger ws2_32)
79-
target_link_libraries(cpp_redis_kill ws2_32)
80-
target_link_libraries(cpp_redis_high_availability_client ws2_32)
81-
else()
82-
target_link_libraries(cpp_redis_client pthread)
83-
target_link_libraries(cpp_redis_future_client pthread)
84-
target_link_libraries(cpp_redis_subscriber pthread)
85-
target_link_libraries(cpp_redis_logger pthread)
86-
target_link_libraries(cpp_redis_kill pthread)
87-
target_link_libraries(cpp_redis_high_availability_client pthread)
88-
endif(WIN32)
42+
set(EXAMPLES cpp_redis_client
43+
cpp_redis_consumer
44+
cpp_redis_future_client
45+
cpp_redis_subscriber
46+
cpp_redis_logger
47+
cpp_redis_kill
48+
cpp_redis_streams_client
49+
cpp_redis_high_availability_client
50+
)
51+
52+
foreach(EXAMPLE IN ITEMS ${EXAMPLES})
53+
###
54+
# executable
55+
###
56+
add_executable(${EXAMPLE} ${EXAMPLE}.cpp)
57+
target_link_libraries(${EXAMPLE} cpp_redis)
58+
59+
###
60+
# link libs
61+
###
62+
if(WIN32)
63+
target_link_libraries(${EXAMPLE} ws2_32)
64+
else()
65+
target_link_libraries(${EXAMPLE} pthread)
66+
endif(WIN32)
67+
endforeach(EXAMPLE)

‎includes/cpp_redis/builders/builder_iface.hpp‎

Lines changed: 16 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -39,26 +39,26 @@ namespace cpp_redis {
3939
public:
4040
virtual ~builder_iface() = default;
4141

42-
/**
43-
* take data as parameter which is consumed to build the reply
44-
* every bytes used to build the reply must be removed from the buffer passed as parameter
45-
*
46-
* @param data data to be consumed
47-
* @return current instance
48-
*
49-
*/
42+
/**
43+
* take data as parameter which is consumed to build the reply
44+
* every bytes used to build the reply must be removed from the buffer passed as parameter
45+
*
46+
* @param data data to be consumed
47+
* @return current instance
48+
*
49+
*/
5050
virtual builder_iface &operator<<(std::string &data) = 0;
5151

52-
/**
53-
* @return whether the reply could be built
54-
*
55-
*/
52+
/**
53+
* @return whether the reply could be built
54+
*
55+
*/
5656
virtual bool reply_ready() const = 0;
5757

58-
/**
59-
* @return reply object
60-
*
61-
*/
58+
/**
59+
* @return reply object
60+
*
61+
*/
6262
virtual reply get_reply() const = 0;
6363
};
6464

‎includes/cpp_redis/builders/builders_factory.hpp‎

Lines changed: 16 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -28,22 +28,21 @@
2828

2929
namespace cpp_redis {
3030

31-
namespace builders {
32-
33-
/**
34-
* create a builder corresponding to the given id
35-
* * + for simple strings
36-
* * - for errors
37-
* * : for integers
38-
* * $ for bulk strings
39-
* * * for arrays
40-
*
41-
* @param id char that determines which builder to return
42-
* @return new builder instance depending on id value
43-
*/
44-
//!
45-
std::unique_ptr<builder_iface> create_builder(char id);
46-
47-
} // namespace builders
31+
namespace builders {
32+
33+
/**
34+
* create a builder corresponding to the given id
35+
* * + for simple strings
36+
* * - for errors
37+
* * : for integers
38+
* * $ for bulk strings
39+
* * * for arrays
40+
*
41+
* @param id char that determines which builder to return
42+
* @return new builder instance depending on id value
43+
*/
44+
std::unique_ptr<builder_iface> create_builder(char id);
45+
46+
} // namespace builders
4847

4948
} // namespace cpp_redis

‎includes/cpp_redis/builders/bulk_string_builder.hpp‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -36,9 +36,9 @@ namespace builders {
3636
class bulk_string_builder : public builder_iface {
3737
public:
3838
//! ctor
39-
bulk_string_builder(void);
39+
bulk_string_builder();
4040
//! dtor
41-
~bulk_string_builder(void) = default;
41+
~bulk_string_builder() override = default;
4242

4343
//! copy ctor
4444
bulk_string_builder(const bulk_string_builder&) = delete;
@@ -53,30 +53,30 @@ class bulk_string_builder : public builder_iface {
5353
//! @param data data to be consumed
5454
//! @return current instance
5555
//!
56-
builder_iface& operator<<(std::string& data);
56+
builder_iface& operator<<(std::string& data) override;
5757

5858
//!
5959
//! @return whether the reply could be built
6060
//!
61-
bool reply_ready(void) const;
61+
bool reply_ready() const override;
6262

6363
//!
6464
//! @return reply object
6565
//!
66-
reply get_reply(void) const;
66+
reply get_reply() const override;
6767

6868
//!
6969
//! @return the parsed bulk string
7070
//!
71-
const std::string& get_bulk_string(void) const;
71+
const std::string& get_bulk_string() const;
7272

7373
//!
7474
//! @return whether the bulk string is null
7575
//!
76-
bool is_null(void) const;
76+
bool is_null() const;
7777

7878
private:
79-
void build_reply(void);
79+
void build_reply();
8080
bool fetch_size(std::string& str);
8181
void fetch_str(std::string& str);
8282

‎includes/cpp_redis/builders/error_builder.hpp‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -40,12 +40,12 @@ class error_builder : public builder_iface {
4040
* ctor
4141
*
4242
*/
43-
error_builder(void) = default;
43+
error_builder() = default;
4444
/**
4545
* dtor
4646
*
4747
*/
48-
~error_builder(void) = default;
48+
~error_builder() override = default;
4949

5050
/**
5151
* copy ctor
@@ -67,25 +67,25 @@ class error_builder : public builder_iface {
6767
* @return current instance
6868
*
6969
*/
70-
builder_iface& operator<<(std::string& data);
70+
builder_iface& operator<<(std::string& data) override;
7171

7272
/**
7373
* @return whether the reply could be built
7474
*
7575
*/
76-
bool reply_ready(void) const;
76+
bool reply_ready() const override;
7777

7878
/**
7979
* @return reply object
8080
*
8181
*/
82-
reply get_reply(void) const;
82+
reply get_reply() const override;
8383

8484
/**
8585
* @return the parsed error
8686
*
8787
*/
88-
const std::string& get_error(void) const;
88+
const std::string& get_error() const;
8989

9090
private:
9191
/**

‎includes/cpp_redis/builders/integer_builder.hpp‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@
2525
#include <cpp_redis/builders/builder_iface.hpp>
2626
#include <cpp_redis/core/reply.hpp>
2727

28-
#include <stdint.h>
28+
#include <cstdint>
2929

3030
namespace cpp_redis {
3131

@@ -41,12 +41,12 @@ class integer_builder : public builder_iface {
4141
* ctor
4242
*
4343
*/
44-
integer_builder(void);
44+
integer_builder();
4545
/**
4646
* dtor
4747
*
4848
*/
49-
~integer_builder(void) = default;
49+
~integer_builder() override = default;
5050

5151
/**
5252
* copy ctor
@@ -68,25 +68,25 @@ class integer_builder : public builder_iface {
6868
* @return current instance
6969
*
7070
*/
71-
builder_iface& operator<<(std::string& data);
71+
builder_iface& operator<<(std::string& data) override;
7272

7373
/**
7474
* @return whether the reply could be built
7575
*
7676
*/
77-
bool reply_ready(void) const;
77+
bool reply_ready() const override;
7878

7979
/**
8080
* @return reply object
8181
*
8282
*/
83-
reply get_reply(void) const;
83+
reply get_reply() const override;
8484

8585
/**
8686
* @return the parsed integer
8787
*
8888
*/
89-
int64_t get_integer(void) const;
89+
int64_t get_integer() const;
9090

9191
private:
9292
/**

‎includes/cpp_redis/builders/simple_string_builder.hpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,7 @@ namespace cpp_redis {
3232
namespace builders {
3333

3434
/**
35-
* builder to build redis simplestring replies
35+
* builder to build redis simple string replies
3636
*
3737
*/
3838
class simple_string_builder : public builder_iface {

‎includes/cpp_redis/core/consumer.hpp‎

Lines changed: 8 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,11 @@ namespace cpp_redis {
4242
acknowledgement_callback_t acknowledgement_callback;
4343
} consumer_callback_container_t;
4444

45+
typedef struct consumer_reply {
46+
std::string group_id;
47+
xstream_reply_t reply;
48+
} consumer_reply_t;
49+
4550
class consumer_client_container {
4651
public:
4752
consumer_client_container();
@@ -67,10 +72,6 @@ namespace cpp_redis {
6772
const consumer_callback_t &consumer_callback,
6873
const acknowledgement_callback_t &acknowledgement_callback = nullptr);
6974

70-
void process();
71-
72-
void poll();
73-
7475
//! @brief Connect to redis server
7576
//! @param host host to be connected to
7677
//! @param port port to be connected to
@@ -86,14 +87,6 @@ namespace cpp_redis {
8687
std::int32_t max_reconnects = 0,
8788
std::uint32_t reconnect_interval_ms = 0);
8889

89-
void read_group_handler(const xreadgroup_options_t &a);
90-
91-
void push_reply(const reply_t &reply);
92-
93-
reply_t pop_reply();
94-
95-
void stream_reply_handler(const xstream_reply& streams) __attribute__((optimize(0)));
96-
9790
//!
9891
//! commit pipelined transaction
9992
//! that is, send to the network all commands pipelined by calling send() / subscribe() / ...
@@ -104,6 +97,9 @@ namespace cpp_redis {
10497

10598
void dispatch_changed_handler(size_t size);
10699

100+
private:
101+
void poll();
102+
107103
private:
108104
std::string m_stream;
109105
std::string m_name;
@@ -119,16 +115,6 @@ namespace cpp_redis {
119115
std::condition_variable dispatch_queue_changed;
120116
std::mutex dispatch_queue_changed_mutex;
121117

122-
std::mutex m_replies_mutex;
123-
std::queue<reply_t> m_replies;
124-
125-
//! Whether there are new replies on the queue
126-
std::atomic_bool replies_empty{true};
127-
128-
//! Used to notify the processing queue
129-
std::condition_variable replies_changed;
130-
std::mutex replies_changed_mutex;
131-
132118
bool is_ready = false;
133119
std::atomic_bool m_should_read_pending{true};
134120
};

‎includes/cpp_redis/core/reply.hpp‎

Lines changed: 0 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -270,12 +270,6 @@ namespace cpp_redis {
270270
*/
271271
type get_type() const;
272272

273-
/**
274-
* support for output
275-
*
276-
*/
277-
friend std::ostream &operator<<(std::ostream &os, const reply &reply);
278-
279273
private:
280274
type m_type;
281275
std::vector<cpp_redis::reply> m_rows;

0 commit comments

Comments
 (0)