Skip to content

Commit f12f155

Browse files
appkinsappkins
committed
xstreams parsing
1 parent 34387b8 commit f12f155

File tree

11 files changed

+349
-345
lines changed

11 files changed

+349
-345
lines changed

‎examples/cpp_redis_client.cpp‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
// SOFTWARE.
2222
#include <string>
2323
#include <cpp_redis/cpp_redis>
24+
#include <cpp_redis/misc/macro.hpp>
2425

2526
#define ENABLE_SESSION = 1
2627

@@ -83,8 +84,8 @@ main(void) {
8384
false, // no ack
8485
}, [](cpp_redis::reply &reply) {
8586
std::cout << "set hello 42: " << reply << std::endl;
86-
auto msg = reply.as_map();
87-
std::cout << "Mes: " << msg["message"] << std::endl;
87+
auto msg = reply.as_array();
88+
std::cout << "Mes: " << msg[0] << std::endl;
8889
// if (reply.is_string())
8990
// do_something_with_string(reply.as_string())
9091
});

‎examples/cpp_redis_consumer.cpp‎

Lines changed: 75 additions & 73 deletions
Original file line numberDiff line numberDiff line change
@@ -1,96 +1,98 @@
1-
/*
2-
*
3-
* Created by nick on 11/22/18.
4-
*
5-
* Copyright(c) 2018 Iris. All rights reserved.
6-
*
7-
* Use and copying of this software and preparation of derivative
8-
* works based upon this software are not permitted. Any distribution
9-
* of this software or derivative works must comply with all applicable
10-
* Canadian export control laws.
11-
*
12-
* THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED
13-
* WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
14-
* OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
15-
* DISCLAIMED. IN NO EVENT SHALL IRIS OR ITS CONTRIBUTORS
16-
* BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY,
17-
* OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
18-
* PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA,
19-
* OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
20-
* ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
21-
* OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
22-
* OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
23-
* SUCH DAMAGE.
24-
*
25-
*/
26-
1+
// The MIT License (MIT)
2+
//
3+
// Copyright (c) 2015-2017 Simon Ninon <simon.ninon@gmail.com>
4+
//
5+
// Permission is hereby granted, free of charge, to any person obtaining a copy
6+
// of this software and associated documentation files (the "Software"), to deal
7+
// in the Software without restriction, including without limitation the rights
8+
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
9+
// copies of the Software, and to permit persons to whom the Software is
10+
// furnished to do so, subject to the following conditions:
11+
//
12+
// The above copyright notice and this permission notice shall be included in all
13+
// copies or substantial portions of the Software.
14+
//
15+
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16+
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17+
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18+
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19+
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20+
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
21+
// SOFTWARE.
22+
#include <string>
2723
#include <cpp_redis/cpp_redis>
28-
#include <tacopie/tacopie>
29-
30-
#include <condition_variable>
31-
#include <iostream>
32-
#include <mutex>
33-
#include <signal.h>
3424

3525
#ifdef _WIN32
3626
#include <Winsock2.h>
3727
#endif /* _WIN32 */
3828

39-
std::condition_variable should_exit;
40-
41-
void
42-
sigint_handler(int) {
43-
should_exit.notify_all();
44-
}
45-
4629
int
4730
main(void) {
4831
#ifdef _WIN32
4932
//! Windows netword DLL init
50-
WORD version = MAKEWORD(2, 2);
51-
WSADATA data;
33+
WORD version = MAKEWORD(2, 2);
34+
WSADATA data;
5235

53-
if (WSAStartup(version, &data) != 0) {
54-
std::cerr << "WSAStartup() failure" << std::endl;
55-
return -1;
56-
}
36+
if (WSAStartup(version, &data) != 0) {
37+
std::cerr << "WSAStartup() failure" << std::endl;
38+
return -1;
39+
}
5740
#endif /* _WIN32 */
5841

5942
//! Enable logging
6043
cpp_redis::active_logger = std::unique_ptr<cpp_redis::logger>(new cpp_redis::logger);
6144

62-
cpp_redis::consumer_options_t opts;
63-
64-
opts.group_name = "groupone";
65-
opts.session_name = "sessone";
66-
67-
cpp_redis::consumer sub(opts);
68-
69-
70-
sub.connect("127.0.0.1", 6379, [](const std::string& host, std::size_t port, cpp_redis::consumer::connect_state status) {
71-
if (status == cpp_redis::consumer::connect_state::dropped) {
72-
std::cout << "client disconnected from " << host << ":" << port << std::endl;
73-
should_exit.notify_all();
74-
}
45+
cpp_redis::client client;
46+
47+
client.connect("127.0.0.1", 6379,
48+
[](const std::string &host, std::size_t port, cpp_redis::client::connect_state status) {
49+
if (status == cpp_redis::client::connect_state::dropped) {
50+
std::cout << "client disconnected from " << host << ":" << port << std::endl;
51+
}
52+
});
53+
54+
auto replcmd = [](cpp_redis::reply &reply) {
55+
std::cout << "set hello 42: " << reply << std::endl;
56+
// if (reply.is_string())
57+
// do_something_with_string(reply.as_string())
58+
};
59+
60+
const std::string group_name = "groupone";
61+
const std::string session_name = "sessone";
62+
const std::string consumer_name = "ABCD";
63+
64+
std::multimap<std::string, std::string> ins;
65+
ins.insert(std::pair<std::string, std::string>{"message", "hello"});
66+
67+
client.xadd(session_name, "*", ins, replcmd);
68+
client.xgroup_create(session_name, group_name, replcmd);
69+
70+
client.sync_commit();
71+
72+
client.xrange(session_name, {"-", "+", 10}, replcmd);
73+
74+
client.xreadgroup({group_name,
75+
consumer_name,
76+
{{session_name}, {">"}},
77+
1, // Count
78+
0, // block milli
79+
false, // no ack
80+
}, [](cpp_redis::reply &reply) {
81+
std::cout << "set hello 42: " << reply << std::endl;
82+
cpp_redis::xstream_reply msg(reply);
83+
std::cout << "Mes: " << msg[0].Messages[0].Values["message"] << std::endl;
84+
// if (reply.is_string())
85+
// do_something_with_string(reply.as_string())
7586
});
7687

77-
//! authentication if server-server requires it
78-
// sub.auth("some_password", [](const cpp_redis::reply& reply) {
79-
// if (reply.is_error()) { std::cerr << "Authentication failed: " << reply.as_string() << std::endl; }
80-
// else {
81-
// std::cout << "successful authentication" << std::endl;
82-
// }
83-
// });
88+
// commands are pipelined and only sent when client.commit() is called
89+
// client.commit();
8490

85-
sub.subscribe("sessone", "groupone", [](const std::string& chan, const std::map<std::string, std::string> & msg) {
86-
std::cout << "MESSAGE " << chan << ": " << msg.at("id") << std::endl;
87-
});
88-
sub.commit();
91+
// synchronous commit, no timeout
92+
client.sync_commit();
8993

90-
signal(SIGINT, &sigint_handler);
91-
std::mutex mtx;
92-
std::unique_lock<std::mutex> l(mtx);
93-
should_exit.wait(l);
94+
// synchronous commit, timeout
95+
// client.sync_commit(std::chrono::milliseconds(100));
9496

9597
#ifdef _WIN32
9698
WSACleanup();

‎includes/cpp_redis/builders/reply_builder.hpp‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -40,9 +40,9 @@ namespace builders {
4040
class reply_builder {
4141
public:
4242
//! ctor
43-
reply_builder(void);
43+
reply_builder();
4444
//! dtor
45-
~reply_builder(void) = default;
45+
~reply_builder() = default;
4646

4747
//! copy ctor
4848
reply_builder(const reply_builder&) = delete;
@@ -69,30 +69,30 @@ class reply_builder {
6969
//!
7070
//! \return the first available reply
7171
//!
72-
const reply& get_front(void) const;
72+
const reply& get_front() const;
7373

7474
//!
7575
//! pop the first available reply
7676
//!
77-
void pop_front(void);
77+
void pop_front();
7878

7979
//!
8080
//! \return whether a reply is available
8181
//!
82-
bool reply_available(void) const;
82+
bool reply_available() const;
8383

8484
//!
8585
//! reset the reply builder to its initial state (clear internal buffer and stages)
8686
//!
87-
void reset(void);
87+
void reset();
8888

8989
private:
9090
//!
9191
//! build reply using m_buffer content
9292
//!
9393
//! \return whether the reply has been fully built or not
9494
//!
95-
bool build_reply(void);
95+
bool build_reply();
9696

9797
private:
9898
//!

‎includes/cpp_redis/builders/simple_string_builder.hpp‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -37,9 +37,9 @@ namespace builders {
3737
class simple_string_builder : public builder_iface {
3838
public:
3939
//! ctor
40-
simple_string_builder(void);
40+
simple_string_builder();
4141
//! dtor
42-
~simple_string_builder(void) = default;
42+
~simple_string_builder() override = default;
4343

4444
//! copy ctor
4545
simple_string_builder(const simple_string_builder&) = delete;
@@ -54,22 +54,22 @@ class simple_string_builder : public builder_iface {
5454
//! \param data data to be consumed
5555
//! \return current instance
5656
//!
57-
builder_iface& operator<<(std::string& data);
57+
builder_iface& operator<<(std::string& data) override;
5858

5959
//!
6060
//! \return whether the reply could be built
6161
//!
62-
bool reply_ready(void) const;
62+
bool reply_ready() const override;
6363

6464
//!
6565
//! \return reply object
6666
//!
67-
reply get_reply(void) const;
67+
reply get_reply() const override;
6868

6969
//!
7070
//! \return the parsed simple string
7171
//!
72-
const std::string& get_simple_string(void) const;
72+
const std::string& get_simple_string() const;
7373

7474
private:
7575
//!

‎includes/cpp_redis/core/reply.hpp‎

Lines changed: 1 addition & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,6 @@
2525
#include <iostream>
2626
#include <string>
2727
#include <vector>
28-
#include <map>
2928

3029
#include <cstdint>
3130

@@ -43,7 +42,6 @@ namespace cpp_redis {
4342
#define __CPP_REDIS_REPLY_NULL 3
4443
#define __CPP_REDIS_REPLY_INT 4
4544
#define __CPP_REDIS_REPLY_ARRAY 5
46-
#define __CPP_REDIS_REPLY_MAP 6
4745

4846
//!
4947
//! type of reply, based on redis server standard replies
@@ -54,8 +52,7 @@ namespace cpp_redis {
5452
simple_string = __CPP_REDIS_REPLY_SIMPLE,
5553
null = __CPP_REDIS_REPLY_NULL,
5654
integer = __CPP_REDIS_REPLY_INT,
57-
array = __CPP_REDIS_REPLY_ARRAY,
58-
map = __CPP_REDIS_REPLY_MAP
55+
array = __CPP_REDIS_REPLY_ARRAY
5956
};
6057

6158
//!
@@ -174,14 +171,6 @@ namespace cpp_redis {
174171
//!
175172
const std::vector<reply> &as_array() const;
176173

177-
//! \brief converts to map of strings
178-
//! \return map of strings
179-
const std::map<std::string, reply> as_map() const;
180-
181-
//! \brief converts to map of strings
182-
//! \return map of strings
183-
const std::map<std::string, std::string> as_str_map() const;
184-
185174
//!
186175
//! \return the underlying string
187176
//!
@@ -220,13 +209,6 @@ namespace cpp_redis {
220209
//!
221210
void set(const std::vector<reply> &rows);
222211

223-
//!
224-
//! set a map reply
225-
//!
226-
//! \param hash map reply
227-
//!
228-
void set(const std::map<std::string, reply> &hash);
229-
230212
//!
231213
//! for array replies, add a new row to the reply
232214
//!
@@ -244,8 +226,6 @@ namespace cpp_redis {
244226
private:
245227
type m_type;
246228
std::vector<cpp_redis::reply> m_rows;
247-
mutable std::map<std::string, cpp_redis::reply> m_hash;
248-
mutable std::map<std::string, std::string> m_str_hash;
249229
std::string m_strval;
250230
int64_t m_intval;
251231
};

‎includes/cpp_redis/core/types.hpp‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
#include <vector>
2828
#include <map>
2929
#include <chrono>
30+
#include <cpp_redis/core/reply.hpp>
3031

3132

3233
namespace cpp_redis {
@@ -54,6 +55,29 @@ namespace cpp_redis {
5455
std::int32_t Count;
5556
} range_type_t;
5657

58+
class xmessage {
59+
public:
60+
explicit xmessage(reply data);
61+
std::string Id;
62+
std::map<std::string, std::string> Values;
63+
};
64+
65+
typedef xmessage xmessage_t;
66+
67+
class xstream {
68+
public:
69+
explicit xstream(reply data);
70+
std::string Stream;
71+
std::vector<xmessage_t> Messages;
72+
};
73+
74+
typedef xstream xstream_t;
75+
76+
class xstream_reply : public std::vector<xstream_t> {
77+
public:
78+
explicit xstream_reply(reply data);
79+
};
80+
5781
class range {
5882
public:
5983
enum class range_state {

‎includes/cpp_redis/misc/macro.hpp‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,3 +27,5 @@
2727
#else /* _WIN32 */
2828
#define __CPP_REDIS_LENGTH(size) size // for Unix, keep size as `size_t`
2929
#endif /* _WIN32 */
30+
31+
#define __CPP_REDIS_PRINT(...) printf(__VA_ARGS__)

0 commit comments

Comments
 (0)