More progesses.

examples/async_basic.cpp

@@ -9,25 +9,47 @@
 
 using namespace aedis;
 
-void receiver(command cmd, resp3::type type, std::queue<pipeline>& pipelines)
+net::awaitable<void>
+example(net::ip::tcp::socket& socket, std::queue<pipeline>& pipelines)
 {
-   std::cout << "Event: " << cmd << " (" << type << ")" << std::endl;
-   switch (cmd) {
-      case command::hello:
-      {
-	 prepare_queue(pipelines);
-	 pipelines.back().ping();
-	 pipelines.back().subscribe("some-channel");
-      } break;
-      case command::publish: break;
-      case command::quit: break;
-      case command::ping:
-      {
-	 prepare_queue(pipelines);
-	 pipelines.back().publish("some-channel", "Some message");
-	 pipelines.back().quit();
-      } break;
-      default: { }
+   pipelines.push({});
+   pipelines.back().hello("3");
+
+   std::string buffer;
+   response_buffers buffers;
+   response_adapters adapters{buffers};
+   consumer_state cs;
+
+   for (;;) {
+      auto const type =
+        co_await async_consume(
+            socket, buffer, pipelines, adapters, cs, net::use_awaitable);
+
+      if (type == resp3::type::push) {
+         std::cout << "Event: " << "(" << type << ")" << std::endl;
+         continue;
+      }
+
+      auto const cmd = pipelines.front().commands.front();
+
+      std::cout << "Event: " << cmd << " (" << type << ")" << std::endl;
+      switch (cmd) {
+         case command::hello:
+         {
+            prepare_queue(pipelines);
+            pipelines.back().ping();
+            pipelines.back().subscribe("some-channel");
+         } break;
+         case command::publish: break;
+         case command::quit: break;
+         case command::ping:
+         {
+            prepare_queue(pipelines);
+            pipelines.back().publish("some-channel", "Some message");
+            pipelines.back().quit();
+         } break;
+         default: { }
+      }
    }
 }
 
@@ -36,17 +58,11 @@ int main()
    net::io_context ioc;
    net::ip::tcp::resolver resolver{ioc};
    auto const res = resolver.resolve("127.0.0.1", "6379");
+
    net::ip::tcp::socket socket{ioc};
    net::connect(socket, res);
 
-   std::string buffer;
-   response_buffers buffers;
    std::queue<pipeline> pipelines;
-
-   pipelines.push({});
-   pipelines.back().hello("3");
-
-   auto f = [&](auto cmd, auto type) {receiver(cmd, type, pipelines);};
-   co_spawn(ioc, async_read(socket, buffer, buffers, pipelines, f), net::detached);
+   co_spawn(ioc, example(socket, pipelines), net::detached);
    ioc.run();
 }