boost/redis
2
0
Fork 0
mirror of https://github.com/boostorg/redis.git synced 2026-02-21 15:22:14 +00:00
Code Issues Projects Releases Wiki Activity

Improvements in the stream.

Browse Source
This commit is contained in:
Marcelo Zimbres
2021-10-31 22:05:38 +01:00
parent 4bf88b2424
commit 1be62b5d90
4 changed files with 85 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
examples/pubsub.cpp
View File

@@ -22,18 +22,14 @@ net::awaitable<void> publisher()
std::queue<resp3::request> requests;
requests.push({});
requests.back().push(command::hello, 3);
requests.back().push(command::publish, "channel1", "Message to channel1");
requests.back().push(command::publish, "channel2", "Message to channel2");
requests.back().push(command::quit);
resp3::stream s;
resp3::stream<tcp_socket> stream{std::move(socket)};
for (;;) {
resp3::response resp;
co_await s.async_consume(socket, requests, resp);
if (requests.front().commands.front() == command::hello) {
prepare_next(requests);
requests.back().push(command::publish, "channel1", "Message to channel1");
requests.back().push(command::publish, "channel2", "Message to channel2");
requests.back().push(command::quit);
}
co_await stream.async_consume(requests, resp);
}
}
@@ -47,21 +43,29 @@ net::awaitable<void> subscriber()
std::queue<resp3::request> requests;
requests.push({});
requests.back().push(command::hello, "3");
requests.back().push(command::subscribe, "channel1", "channel2");
resp3::stream s;
resp3::stream<tcp_socket> stream{std::move(socket)};
for (;;) {
resp3::response resp;
co_await s.async_consume(socket, requests, resp);
co_await stream.async_consume(requests, resp);
if (resp.get_type() == resp3::type::push) {
std::cout << "Subscriber " << id << ":\n" << resp << std::endl;
std::cout
<< "Subscriber " << id << "\n"
<< resp << std::endl;
continue;
}
if (requests.front().commands.front() == command::hello) {
id = resp.raw().at(8).data;
prepare_next(requests);
requests.back().push(command::subscribe, "channel1", "channel2");
auto const cmd = requests.front().commands.front();
switch (cmd) {
case command::hello:
id = resp.raw().at(8).data;
break;
default:
std::cout
<< cmd << "\n"
<< resp << std::endl;
}
}
}