diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt
index 1c7e81c..d2d865d 100644
--- a/examples/CMakeLists.txt
+++ b/examples/CMakeLists.txt
@@ -35,7 +35,13 @@ set(EXAMPLES
 )
 
 if(${HAS_POSTGRES})
-    list(APPEND EXAMPLES postgres postgres-talk populate-postgres)
+    list(APPEND EXAMPLES
+        postgres
+        postgres-talk
+        postgres-talk-accuonsea
+        postgres-talk-c++now
+        populate-postgres
+    )
 endif()
 
 foreach(EXAMPLE ${EXAMPLES})
diff --git a/examples/postgres-talk-accuonsea.cpp b/examples/postgres-talk-accuonsea.cpp
new file mode 100644
index 0000000..a7e6b4b
--- /dev/null
+++ b/examples/postgres-talk-accuonsea.cpp
@@ -0,0 +1,214 @@
+// examples/postgres-talk.cpp                                         -*-C++-*-
+// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
+
+#include <beman/net/net.hpp>
+#include <beman/execution/execution.hpp>
+#include <libpq-fe.h>
+#include "demo_algorithm.hpp"
+#include "demo_http.hpp"
+#include <chrono>
+#include <iostream>
+#include <optional>
+#include <ranges>
+#include <string>
+#include <thread>
+#include <vector>
+
+namespace ex  = beman::execution;
+namespace net = beman::net;
+using namespace std::chrono_literals;
+
+// PQconnectdb(const char* connstr) -> PGconn*
+// PQfinish(PGconn*) - clean-up
+// PQexec(const PGconn *conn, const char* query) -> PGresult* - query the database
+// PQsendQuery(const PGconn *conn, const char* query) - send a query
+// PQconsumeInput(const PGconn *conn) - consume available input, clear socket stat
+// PQgetResult(const PGconn *conn) -> PGresult* - get result, return nullptr if no more results, or would block
+// PQclear(PGresult) - clean-up
+// PQsetnonblocking(const PGconn *conn, int arg) - set non-blocking mode to avoid write blocks
+// PQsocket(const PGconn *conn) - get socket
+// PQflush(const PGconn *conn) - flush output buffer, return 1 if still pending data
+// PQisBusy(const PGconn *conn) -> int - PQgetResult() would block
+// PQsetSingleRowMode(PGconn *conn) - set single row mode, return 0 on failure
+// PQsetChunkedMode(PGconn *conn, int arg) - set chunked mode, return 0 on failure
+
+namespace {
+const std::string connection_string("user=sruser dbname=sruser");
+const std::string query("select *, pg_sleep(0.5) from messages where 0 < key and key < 5;");
+const std::string query2("select *, pg_sleep(0.5) from messages where 4 < key and key < 8;");
+
+inline constexpr auto print_result [[maybe_unused]]{[](const PGresult* result) noexcept {
+    std::cout << "n=" << PQntuples(result) << ", m=" << PQnfields(result) << "\n";
+    for (int i = 0, n = PQntuples(result); i < n; ++i) {
+        for (int j = 0, m = PQnfields(result); j < m; ++j == m || std::cout << ", ") {
+            std::cout << PQgetvalue(result, i, j);
+        }
+        std::cout << "\n";
+    }
+}};
+} // namespace
+
+namespace pg {
+struct conn {
+    std::unique_ptr<PGconn, decltype([](auto c) { PQfinish(c); })> con;
+    net::ip::tcp::socket                                           socket;
+    conn(net::io_context& io, PGconn* c) : con(c), socket(io, io.make_socket(PQsocket(con.get()))) {}
+    operator PGconn*() const { return con.get(); }
+};
+
+struct result {
+    std::unique_ptr<PGresult, decltype([](auto r) { PQclear(r); })> res;
+    result(PGresult* r) : res(r) {}
+    operator PGresult*() const { return res.get(); }
+};
+using results = std::vector<pg::result>;
+ex::task<results> exec(auto& conn, std::string query) {
+    if (!PQsendQuery(conn, query.c_str())) {
+        std::cout << "ERROR: " << PQerrorMessage(conn) << "\n";
+        throw std::runtime_error("ERROR");
+    }
+    while (PQflush(conn)) {
+        co_await net::async_poll(conn.socket, net::event_type::out);
+    }
+    pg::results res;
+    std::cout << "sent query\n";
+
+    while (true) {
+        while (PQisBusy(conn)) {
+            co_await ex::unstoppable(net::async_poll(conn.socket, net::event_type::in));
+            if (!PQconsumeInput(conn)) {
+                // error handling
+            }
+        }
+        std::cout << "got a result\n";
+        if (!res.emplace_back(PQgetResult(conn))) {
+            res.pop_back();
+            break;
+        }
+    }
+    co_return res;
+}
+
+template <typename Object>
+struct mutex {
+    struct state_base {
+        state_base*  next{};
+        virtual void run() = 0;
+    };
+
+    Object      obj;
+    bool        is_busy{false};
+    state_base* waiting{};
+
+    template <typename... A>
+    mutex(A&&... a) : obj(std::forward<A>(a)...) {}
+
+    template <typename Fun>
+    using sender_t = decltype(std::declval<Fun>()(std::declval<Object&>()));
+
+    template <typename Rcvr, typename Fun>
+    struct state : state_base {
+        using operation_state_concept = ex::operation_state_tag;
+        std::remove_cvref_t<Rcvr> rcvr;
+        mutex*                    mut;
+
+        struct receiver {
+            using receiver_concept = ex::receiver_tag;
+            state* s;
+            auto   get_env() const noexcept { return ex::get_env(s->rcvr); }
+            template <typename... A>
+            void set_value(A&&... a) noexcept {
+                s->complete();
+                ex::set_value(std::move(s->rcvr), std::forward<A>(a)...);
+            }
+            template <typename A>
+            void set_error(A&& a) noexcept {
+                s->complete();
+                ex::set_error(std::move(s->rcvr), std::forward<A>(a));
+            }
+            void set_stopped() noexcept {
+                s->complete();
+                ex::set_stopped(std::move(s->rcvr));
+            }
+        };
+
+        using inner_state_t = ex::connect_result_t<sender_t<Fun>, receiver>;
+
+        inner_state_t inner_state;
+        state(Rcvr&& r, Fun f, mutex* m)
+            : rcvr(std::forward<Rcvr>(r)), mut(m), inner_state(ex::connect(std::move(f)(mut->obj), receiver{this})) {}
+        void start() noexcept {
+            if (!std::exchange(this->mut->is_busy, true)) {
+                run();
+            } else {
+                this->next = std::exchange(mut->waiting, this);
+            }
+        }
+        void complete() {
+            if (mut->waiting) {
+                std::exchange(mut->waiting, mut->waiting->next)->run();
+            } else {
+                mut->is_busy = false;
+            }
+        }
+        void run() { ex::start(inner_state); }
+    };
+
+    template <typename Fun>
+    struct sender {
+        using sender_concept = ex::sender_tag;
+        template <typename, typename... Env>
+        static consteval auto get_completion_signatures() {
+            return ex::get_completion_signatures<sender_t<Fun>, Env...>();
+        }
+
+        Fun    fun;
+        mutex* mut;
+
+        template <ex::receiver Rcvr>
+        auto connect(Rcvr&& r) && {
+            return state<Rcvr, Fun>(std::forward<Rcvr>(r), std::move(fun), mut);
+        }
+    };
+
+    template <typename Fun>
+    auto run(Fun fun) {
+        return sender<Fun>(std::move(fun), this);
+    }
+};
+
+auto exec(pg::mutex<pg::conn>& conn, std::string query) {
+    return conn.run([=](pg::conn& obj) { return exec(obj, query); });
+}
+} // namespace pg
+
+// ----------------------------------------------------------------------------
+
+auto main() -> int {
+    std::cout << std::unitbuf << "Postgres Example\n";
+    net::io_context     io;
+    pg::mutex<pg::conn> conn(io, PQconnectdb(connection_string.c_str()));
+
+    auto clock = [](auto sched) -> ex::task<> {
+        while (true) {
+            std::cout << "time=" << std::chrono::system_clock::now() << "\n";
+            // std::this_thread::sleep_for(1s);
+            co_await net::resume_after(sched, 1s);
+        }
+        co_return;
+    }(io.get_scheduler());
+    std::cout << "do more!\n";
+
+    auto query = [](auto& conn) -> ex::task<> {
+        auto res = co_await pg::exec(conn, ::query);
+        std::ranges::for_each(res, ::print_result);
+        co_return;
+    }(conn);
+    auto query2 = [](auto& conn) -> ex::task<> {
+        auto res = co_await pg::exec(conn, ::query2);
+        std::ranges::for_each(res, ::print_result);
+        co_return;
+    }(conn);
+
+    ex::sync_wait(demo::when_any(io.async_run(), std::move(clock), ex::when_all(std::move(query), std::move(query2))));
+}
diff --git a/examples/postgres-talk-c++now.cpp b/examples/postgres-talk-c++now.cpp
new file mode 100644
index 0000000..eb33a0f
--- /dev/null
+++ b/examples/postgres-talk-c++now.cpp
@@ -0,0 +1,217 @@
+// examples/postgres-talk.cpp                                         -*-C++-*-
+// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
+
+#include <beman/net/net.hpp>
+#include <beman/execution/execution.hpp>
+#include <libpq-fe.h>
+#include "demo_algorithm.hpp"
+#include "demo_http.hpp"
+#include <chrono>
+#include <iostream>
+#include <optional>
+#include <ranges>
+#include <string>
+#include <thread>
+#include <vector>
+
+namespace ex  = beman::execution;
+namespace net = beman::net;
+using namespace std::chrono_literals;
+
+// PQconnectdb(const char* connstr) -> PGconn*
+// PQfinish(PGconn*) - clean-up
+// PQexec(const PGconn *conn, const char* query) -> PGresult* - query the database
+// PQsendQuery(const PGconn *conn, const char* query) - send a query
+// PQconsumeInput(const PGconn *conn) - consume available input, clear socket stat
+// PQgetResult(const PGconn *conn) -> PGresult* - get result, return nullptr if no more results, or would block
+// PQclear(PGresult) - clean-up
+// PQsetnonblocking(const PGconn *conn, int arg) - set non-blocking mode to avoid write blocks
+// PQsocket(const PGconn *conn) - get socket
+// PQflush(const PGconn *conn) - flush output buffer, return 1 if still pending data
+// PQisBusy(const PGconn *conn) -> int - PQgetResult() would block
+// PQsetSingleRowMode(PGconn *conn) - set single row mode, return 0 on failure
+// PQsetChunkedMode(PGconn *conn, int arg) - set chunked mode, return 0 on failure
+
+namespace {
+const std::string connection_string("user=sruser dbname=sruser");
+const std::string query("select *, pg_sleep(0.5) from messages where 0 < key and key < 5;");
+const std::string query2("select *, pg_sleep(0.5) from messages where 4 < key and key < 8;");
+
+inline constexpr auto print_result{[](const PGresult* result) noexcept {
+    std::cout << "n=" << PQntuples(result) << ", m=" << PQnfields(result) << "\n";
+    for (int i = 0, n = PQntuples(result); i < n; ++i) {
+        for (int j = 0, m = PQnfields(result); j < m; ++j == m || std::cout << ", ") {
+            std::cout << PQgetvalue(result, i, j);
+        }
+        std::cout << "\n";
+    }
+}};
+} // namespace
+
+namespace pg {
+
+struct conn {
+    std::unique_ptr<PGconn, decltype([](auto c) { PQfinish(c); })> cn;
+    net::ip::tcp::socket                                           socket;
+    conn(net::io_context& io, PGconn* c) : cn(c), socket(io, io.make_socket(PQsocket(c))) {}
+    operator PGconn*() const { return cn.get(); }
+};
+
+struct result {
+    std::unique_ptr<PGresult, decltype([](auto r) { PQclear(r); })> res;
+    result(auto r) : res(r) {}
+    operator PGresult*() const { return res.get(); }
+};
+using results = std::vector<pg::result>;
+struct error {
+    std::string message;
+    error(std::string m) : message(m) { std::cout << "error:" << m << "\n"; }
+};
+
+struct env {
+    using error_types = ex::completion_signatures<ex::set_error_t(pg::error)>;
+};
+
+ex::task<pg::results, pg::env> exec(auto& conn, std::string query) {
+    if (!PQsendQuery(conn, query.c_str())) {
+        co_yield ex::with_error(pg::error(PQerrorMessage(conn)));
+    }
+    while (PQflush(conn)) {
+        co_await ex::unstoppable(net::async_poll(conn.socket, net::event_type::out));
+    }
+    pg::results res;
+    while (true) {
+        while (PQisBusy(conn)) {
+            co_await ex::unstoppable(net::async_poll(conn.socket, net::event_type::in));
+            if (!PQconsumeInput(conn)) {
+                co_yield ex::with_error(pg::error(PQerrorMessage(conn)));
+            }
+        }
+        if (!res.emplace_back(pg::result(PQgetResult(conn)))) {
+            res.pop_back();
+            break;
+        }
+    }
+    co_return std::move(res);
+}
+
+template <typename Object>
+class mutex {
+    struct state_base {
+        state_base*  next{};
+        virtual void run() = 0;
+    };
+    Object      obj;
+    bool        is_busy{};
+    state_base* waiting{};
+
+    template <typename Fun>
+    using sender_t = decltype(std::declval<Fun>()(std::declval<Object&>()));
+
+    template <ex::receiver Rcvr, typename Fun>
+    struct state : state_base {
+        using operation_state_concept = ex::operation_state_tag;
+
+        std::remove_cvref_t<Rcvr> rcvr;
+        mutex*                    self;
+
+        struct receiver {
+            using receiver_concept = ex::receiver_tag;
+            state* st;
+
+            auto get_env() const noexcept { return ex::get_env(st->rcvr); }
+            template <typename... A>
+            void set_value(A&&... a) && noexcept {
+                st->complete();
+                ex::set_value(std::move(st->rcvr), std::forward<A>(a)...);
+            }
+            template <typename E>
+            void set_error(E&& e) && noexcept {
+                st->complete();
+                ex::set_error(std::move(st->rcvr), std::forward<E>(e));
+            }
+            void set_stopped() && noexcept {
+                st->complete();
+                ex::set_stopped(std::move(st->rcvr));
+            }
+        };
+        using inner_state_t = ex::connect_result_t<sender_t<Fun>, receiver>;
+        inner_state_t inner_state;
+
+        state(Rcvr&& r, mutex* s, Fun f)
+            : rcvr(std::forward<Rcvr>(r)),
+              self(s),
+              inner_state(ex::connect(std::move(f)(self->obj), receiver(this))) {}
+        void start() & noexcept {
+            if (std::exchange(self->is_busy, true)) {
+                this->next = std::exchange(self->waiting, this);
+            } else {
+                run();
+            }
+        }
+        void run() override { ex::start(inner_state); }
+        void complete() {
+            if (self->waiting) {
+                std::exchange(self->waiting, self->waiting->next)->run();
+            } else {
+                self->is_busy = false;
+            }
+        }
+    };
+    template <typename Fun>
+    struct sender {
+        using sender_concept = ex::sender_tag;
+        template <typename, typename... Env>
+        static consteval auto get_completion_signatures() {
+            return ex::get_completion_signatures<sender_t<Fun>, Env...>();
+        }
+        mutex* self;
+        Fun    fun;
+
+        template <ex::receiver Rcvr>
+        auto connect(Rcvr&& rcvr) && {
+            return state<Rcvr, Fun>(std::forward<Rcvr>(rcvr), self, std::move(fun));
+        }
+    };
+
+  public:
+    template <typename... A>
+    mutex(A&&... a) : obj(std::forward<A>(a)...) {}
+    template <typename Fun>
+    auto run(Fun fun) {
+        return sender<Fun>(this, std::move(fun));
+    }
+};
+
+auto exec(pg::mutex<pg::conn>& conn, std::string query) {
+    return conn.run([query = std::move(query)](pg::conn& conn) { return pg::exec(conn, query); });
+}
+} // namespace pg
+
+// ----------------------------------------------------------------------------
+
+auto main() -> int {
+    std::cout << std::unitbuf << "Postgres Example\n";
+    net::io_context     io;
+    pg::mutex<pg::conn> conn(io, PQconnectdb(connection_string.c_str()));
+
+    auto clock{[](auto sched) -> ex::task<> {
+        while (true) {
+            std::cout << "time=" << std::chrono::system_clock::now() << "\n";
+            co_await net::resume_after(sched, 1s);
+        }
+    }(io.get_scheduler())};
+
+    auto query1{[](auto sched, auto& conn) -> ex::task<> {
+        auto res(co_await pg::exec(conn, query));
+        std::ranges::for_each(res, print_result);
+        co_await net::resume_after(sched, 2s);
+    }(io.get_scheduler(), conn)};
+    auto query2(pg::exec(conn, ::query2) | ex::then([](auto res) { std::ranges::for_each(res, print_result); }));
+
+    // ex::sync_wait(demo::when_any());
+    // ex::sync_wait(ex::when_all());
+
+    ex::sync_wait(
+        demo::when_any(io.async_run(), std::move(clock), ex::when_all(std::move(query1), std::move(query2))));
+}
diff --git a/examples/postgres-talk.cpp b/examples/postgres-talk.cpp
index eb33a0f..a7e6b4b 100644
--- a/examples/postgres-talk.cpp
+++ b/examples/postgres-talk.cpp
@@ -37,7 +37,7 @@ const std::string connection_string("user=sruser dbname=sruser");
 const std::string query("select *, pg_sleep(0.5) from messages where 0 < key and key < 5;");
 const std::string query2("select *, pg_sleep(0.5) from messages where 4 < key and key < 8;");
 
-inline constexpr auto print_result{[](const PGresult* result) noexcept {
+inline constexpr auto print_result [[maybe_unused]]{[](const PGresult* result) noexcept {
     std::cout << "n=" << PQntuples(result) << ", m=" << PQnfields(result) << "\n";
     for (int i = 0, n = PQntuples(result); i < n; ++i) {
         for (int j = 0, m = PQnfields(result); j < m; ++j == m || std::cout << ", ") {
@@ -49,115 +49,111 @@ inline constexpr auto print_result{[](const PGresult* result) noexcept {
 } // namespace
 
 namespace pg {
-
 struct conn {
-    std::unique_ptr<PGconn, decltype([](auto c) { PQfinish(c); })> cn;
+    std::unique_ptr<PGconn, decltype([](auto c) { PQfinish(c); })> con;
     net::ip::tcp::socket                                           socket;
-    conn(net::io_context& io, PGconn* c) : cn(c), socket(io, io.make_socket(PQsocket(c))) {}
-    operator PGconn*() const { return cn.get(); }
+    conn(net::io_context& io, PGconn* c) : con(c), socket(io, io.make_socket(PQsocket(con.get()))) {}
+    operator PGconn*() const { return con.get(); }
 };
 
 struct result {
     std::unique_ptr<PGresult, decltype([](auto r) { PQclear(r); })> res;
-    result(auto r) : res(r) {}
+    result(PGresult* r) : res(r) {}
     operator PGresult*() const { return res.get(); }
 };
 using results = std::vector<pg::result>;
-struct error {
-    std::string message;
-    error(std::string m) : message(m) { std::cout << "error:" << m << "\n"; }
-};
-
-struct env {
-    using error_types = ex::completion_signatures<ex::set_error_t(pg::error)>;
-};
-
-ex::task<pg::results, pg::env> exec(auto& conn, std::string query) {
+ex::task<results> exec(auto& conn, std::string query) {
     if (!PQsendQuery(conn, query.c_str())) {
-        co_yield ex::with_error(pg::error(PQerrorMessage(conn)));
+        std::cout << "ERROR: " << PQerrorMessage(conn) << "\n";
+        throw std::runtime_error("ERROR");
     }
     while (PQflush(conn)) {
-        co_await ex::unstoppable(net::async_poll(conn.socket, net::event_type::out));
+        co_await net::async_poll(conn.socket, net::event_type::out);
     }
     pg::results res;
+    std::cout << "sent query\n";
+
     while (true) {
         while (PQisBusy(conn)) {
             co_await ex::unstoppable(net::async_poll(conn.socket, net::event_type::in));
             if (!PQconsumeInput(conn)) {
-                co_yield ex::with_error(pg::error(PQerrorMessage(conn)));
+                // error handling
             }
         }
-        if (!res.emplace_back(pg::result(PQgetResult(conn)))) {
+        std::cout << "got a result\n";
+        if (!res.emplace_back(PQgetResult(conn))) {
             res.pop_back();
             break;
         }
     }
-    co_return std::move(res);
+    co_return res;
 }
 
 template <typename Object>
-class mutex {
+struct mutex {
     struct state_base {
         state_base*  next{};
         virtual void run() = 0;
     };
+
     Object      obj;
-    bool        is_busy{};
+    bool        is_busy{false};
     state_base* waiting{};
 
+    template <typename... A>
+    mutex(A&&... a) : obj(std::forward<A>(a)...) {}
+
     template <typename Fun>
     using sender_t = decltype(std::declval<Fun>()(std::declval<Object&>()));
 
-    template <ex::receiver Rcvr, typename Fun>
+    template <typename Rcvr, typename Fun>
     struct state : state_base {
         using operation_state_concept = ex::operation_state_tag;
-
         std::remove_cvref_t<Rcvr> rcvr;
-        mutex*                    self;
+        mutex*                    mut;
 
         struct receiver {
             using receiver_concept = ex::receiver_tag;
-            state* st;
-
-            auto get_env() const noexcept { return ex::get_env(st->rcvr); }
+            state* s;
+            auto   get_env() const noexcept { return ex::get_env(s->rcvr); }
             template <typename... A>
-            void set_value(A&&... a) && noexcept {
-                st->complete();
-                ex::set_value(std::move(st->rcvr), std::forward<A>(a)...);
+            void set_value(A&&... a) noexcept {
+                s->complete();
+                ex::set_value(std::move(s->rcvr), std::forward<A>(a)...);
             }
-            template <typename E>
-            void set_error(E&& e) && noexcept {
-                st->complete();
-                ex::set_error(std::move(st->rcvr), std::forward<E>(e));
+            template <typename A>
+            void set_error(A&& a) noexcept {
+                s->complete();
+                ex::set_error(std::move(s->rcvr), std::forward<A>(a));
             }
-            void set_stopped() && noexcept {
-                st->complete();
-                ex::set_stopped(std::move(st->rcvr));
+            void set_stopped() noexcept {
+                s->complete();
+                ex::set_stopped(std::move(s->rcvr));
             }
         };
+
         using inner_state_t = ex::connect_result_t<sender_t<Fun>, receiver>;
-        inner_state_t inner_state;
 
-        state(Rcvr&& r, mutex* s, Fun f)
-            : rcvr(std::forward<Rcvr>(r)),
-              self(s),
-              inner_state(ex::connect(std::move(f)(self->obj), receiver(this))) {}
-        void start() & noexcept {
-            if (std::exchange(self->is_busy, true)) {
-                this->next = std::exchange(self->waiting, this);
-            } else {
+        inner_state_t inner_state;
+        state(Rcvr&& r, Fun f, mutex* m)
+            : rcvr(std::forward<Rcvr>(r)), mut(m), inner_state(ex::connect(std::move(f)(mut->obj), receiver{this})) {}
+        void start() noexcept {
+            if (!std::exchange(this->mut->is_busy, true)) {
                 run();
+            } else {
+                this->next = std::exchange(mut->waiting, this);
             }
         }
-        void run() override { ex::start(inner_state); }
         void complete() {
-            if (self->waiting) {
-                std::exchange(self->waiting, self->waiting->next)->run();
+            if (mut->waiting) {
+                std::exchange(mut->waiting, mut->waiting->next)->run();
             } else {
-                self->is_busy = false;
+                mut->is_busy = false;
             }
         }
+        void run() { ex::start(inner_state); }
     };
+
     template <typename Fun>
     struct sender {
         using sender_concept = ex::sender_tag;
@@ -165,26 +161,24 @@ class mutex {
         static consteval auto get_completion_signatures() {
             return ex::get_completion_signatures<sender_t<Fun>, Env...>();
         }
-        mutex* self;
+
         Fun    fun;
+        mutex* mut;
 
         template <ex::receiver Rcvr>
-        auto connect(Rcvr&& rcvr) && {
-            return state<Rcvr, Fun>(std::forward<Rcvr>(rcvr), self, std::move(fun));
+        auto connect(Rcvr&& r) && {
+            return state<Rcvr, Fun>(std::forward<Rcvr>(r), std::move(fun), mut);
         }
     };
 
-  public:
-    template <typename... A>
-    mutex(A&&... a) : obj(std::forward<A>(a)...) {}
     template <typename Fun>
     auto run(Fun fun) {
-        return sender<Fun>(this, std::move(fun));
+        return sender<Fun>(std::move(fun), this);
     }
 };
 
 auto exec(pg::mutex<pg::conn>& conn, std::string query) {
-    return conn.run([query = std::move(query)](pg::conn& conn) { return pg::exec(conn, query); });
+    return conn.run([=](pg::conn& obj) { return exec(obj, query); });
 }
 } // namespace pg
 
@@ -195,23 +189,26 @@ auto main() -> int {
     net::io_context     io;
     pg::mutex<pg::conn> conn(io, PQconnectdb(connection_string.c_str()));
 
-    auto clock{[](auto sched) -> ex::task<> {
+    auto clock = [](auto sched) -> ex::task<> {
         while (true) {
             std::cout << "time=" << std::chrono::system_clock::now() << "\n";
+            // std::this_thread::sleep_for(1s);
             co_await net::resume_after(sched, 1s);
         }
-    }(io.get_scheduler())};
-
-    auto query1{[](auto sched, auto& conn) -> ex::task<> {
-        auto res(co_await pg::exec(conn, query));
-        std::ranges::for_each(res, print_result);
-        co_await net::resume_after(sched, 2s);
-    }(io.get_scheduler(), conn)};
-    auto query2(pg::exec(conn, ::query2) | ex::then([](auto res) { std::ranges::for_each(res, print_result); }));
-
-    // ex::sync_wait(demo::when_any());
-    // ex::sync_wait(ex::when_all());
-
-    ex::sync_wait(
-        demo::when_any(io.async_run(), std::move(clock), ex::when_all(std::move(query1), std::move(query2))));
+        co_return;
+    }(io.get_scheduler());
+    std::cout << "do more!\n";
+
+    auto query = [](auto& conn) -> ex::task<> {
+        auto res = co_await pg::exec(conn, ::query);
+        std::ranges::for_each(res, ::print_result);
+        co_return;
+    }(conn);
+    auto query2 = [](auto& conn) -> ex::task<> {
+        auto res = co_await pg::exec(conn, ::query2);
+        std::ranges::for_each(res, ::print_result);
+        co_return;
+    }(conn);
+
+    ex::sync_wait(demo::when_any(io.async_run(), std::move(clock), ex::when_all(std::move(query), std::move(query2))));
 }