operators vs asio::completion_tokens

[geiseri]

TL;DR: want to use a function that returns the result as a callback, and then return that value to the next operator. I am pretty sure this won't work because the asio token is lazy and the rpp operator is eager. Or is it?

I have a callback interface from a class that is invoked every time a message arrives on a socket. I used the code below to hook the event emitter to the observable.

    auto emitter = ...
    auto obs = rpp::source::create<std::vector<uint8_t>>([emitter](const auto& sub) {
            emitter->watch([](const std::vector<uint8_t> msg) {
                sub.on_next(msg);
           });  
    });

This is working just fine for the case of:

   auto obs = (...)
           .map([](auto msg) { return std::make_shared<Decoder>(msg); })
          .subscribe(
                     [](std::shared_ptr<Decoder> res) {  std::cout << res->value(); }, // Print out the parsed value
                     [](std::exception_ptr err) {
                         try {
                             std::rethrow_exception(err);
                         } catch (std::exception &exp) {
                            std::cerr <<  exp.what();
                         }
                         exit();
                     });

What I would like to do next is dispatch this to an async call that returns an asio completion token type. I would like to extract that result and then pass it on to the next stage of the pipeline. Something like:

   auto asio_request = ...
   auto obs = (...)
           .map([](auto msg) { return std::make_shared<Decoder>(std::move(msg)); })
           .map([asio_request](auto msg) { 
                   return asio_request->request(msg->value(), asio::deferred); } //  return an asio completion token. 
           })
           ...???... // on the resolution of the completion token emit the value and/or error 
          .subscribe(
                     [](auto res) {  std::cout << res; }, // Print out the message returned from the completion token
                     [](std::exception_ptr err) {
                         try {
                             std::rethrow_exception(err);
                         } catch (std::exception &exp) {
                            std::cerr <<  exp.what();
                         }
                         exit();
                     });

[AlexInLog]

Sorry, i don't have a lot of experience with asio especially new things related to tokens and etc. Could you provide more detailed or abstract example?
If i get correct, do you want to do blocking operation in async way? In this case you can use scheduler to transfer blocking operation to another thread. SOmething like this

auto obs = ...;
obs.observer_on(rpp::scheduler::new_thread{}).
   .map([](auto msg) { .... std::this_thread::sleep_for(std::chrono::seconds{10}); return msg.get_value(); })
   .subscribe....

[geiseri]

I actually want the opposite, I want to perform an async action in a blocking way. It would be exactly like you had above, save for the issue that their generic completion token uses a callback and not a .get()

What I ended up doing was a little janky, but so far it's like this:

std::vector<int> series{2,5,0,1,6,4,8,7,9,3}
asio::io_context io_context;
auto work = asio::make_work_guard(io_context);

// Create a single asio process per item in the series.
auto job1 = rpp::source::from_iterable<rpp::memory_model::use_shared>(series)
    .flat_map([](auto time) {
        // Create a simple asio timer for testing the completion token
        auto delay_timer = std::make_shared<asio::steady_timer>(io_context, std::chrono::seconds(10 - time));
        std::cout << "Job " << time <<" started" << std::endl;

        // create an observable that will call on_next when the asio process completes
        return rpp::source::create<std::string>([time, delay_timer](const auto &sub) {
            std::cout << "subscriber created " << time;
            delay_timer->async_wait(asio::deferred([time](boost::system::error_code ec) mutable {
                // This is just a tap, and isn't needed, the following callback is used.
                std::cout << "Job " << time << " finished" << std::endl;
                return asio::deferred.values(ec);
            }))([sub, delay_timer, time](boost::system::error_code ec) {
                // keep the asio object in scope until the on_next is called. 
                (void)delay_timer;
                (void)ec;
                std::cout << "on next " <<  time << std::endl;
                // Send the value on to the next in the chain
                sub.on_next(fmt::format("job {}", time));
                // Needed to show that the observable is complete.
                sub.on_completed();
            });
        });
    })
    // Collect the results of all the async tasks into a single vector
    .reduce(std::vector<std::string>{}, [](std::vector<std::string> &&seed, std::string new_value) {
        std::cout << "adding value " << new_value << std::endl;
        seed.push_back(new_value);
        return std::move(seed);
    })
    .subscribe([work](std::vector<std::string> res) {
        // Print the result
        std::cout << "set " << fmt::join(res, ",") << std::endl;
        // keep the io_context.run() from returning until all the jobs have been resolved. 
        (void) work;
    });
std::cout << "lets go!" << std::endl;
io_context.run();

While this works it violates the "Observables must issue notifications to observers serial" contract. The output produced is:

series 2,5,0,1,6,4,8,7,9,3
Job 2 started
subscriber created 2
Job 5 started
subscriber created 5
Job 0 started
subscriber created 0
Job 1 started
subscriber created 1
Job 6 started
subscriber created 6
Job 4 started
subscriber created 4
Job 8 started
subscriber created 8
Job 7 started
subscriber created 7
Job 9 started
subscriber created 9
Job 3 started
subscriber created 3
lets go!
Job 9 finished
on next 9
adding value job 9
Job 8 finished
on next 8
adding value job 8
Job 7 finished
on next 7
adding value job 7
Job 6 finished
on next 6
adding value job 6
Job 5 finished
on next 5
adding value job 5
Job 4 finished
on next 4
adding value job 4
Job 3 finished
on next 3
adding value job 3
Job 2 finished
on next 2
adding value job 2
Job 1 finished
on next 1
adding value job 1
Job 0 finished
on next 0
adding value job 0
set job 9,job 8,job 7,job 6,job 5,job 4,job 3,job 2,job 1,job 0

So while they are created in the right order since the timers all resolve at different times they end up reordering based on the time.

ASIO has a notion of a strand so that you can complete things in a series, so I was going to look into that next. I was also thinking about providing a scheduler like in Qt. That would allow for serialization.

So I guess the question is at this point "am I doing it wrong?" before I go down this road.

[AlexInLog]

While this works it violates the "Observables must issue notifications to observers serial" contract

There is no violation of this rule. Yeah, actually you are doing async operations, but without violation of this rule. flat_map operator is actually something like map()+merge().

So, when you are create observables inside flat_map, actually you are create MULTIPLE observables and each of them doesn't violations this rule : you are emitting values in async way, but in serial way - only from one callback which triggered once. You can violate rule there if would start multiple callbacks, or callback can be called multiple time NON serial.

Only one possible issue - merge (possibly) obtains values from different threads in async way, but it is ready for it.

You can check, that there is no any violations via adding sleep in subscribe - while you are inside "sleep" on some value, there is no way to obtain any other value, due to merge emitting values in serial way.

[geiseri]

Ah, so I might be "doing it wrong" for my end goal. It makes sense that the multiple observers from what you are saying. Since they can call on_next at any time they will reach the reduce in that order.

I would like to keep the order of the items in the reduce so they match the order in the order they were in when passed to the flat_map. I am not sure how the subscriber would handle this though, since it would need to be able to track the on_next breaking the "pipeline" approach. It really would need to be controlled by the scheduler for each asio operation's executor so that they resolve in the order they were submitted.

That being said, I wonder if a custom rpp scheduler could be used to coordinate the execution. This would dovetail nicely into the Unified Executors Proposal for C++ (P0443) on the off chance it ever arrives.

[geiseri]

Oh, I think I know what I want. It looks like I want. map(...).concat() vs the flat_map. Does that sound correct?

[AlexInLog]

Ohhhh, i've finally catch your idea =) yeah, you need map+concat. Concat is starting subscribe to "next" observable ONLY when previous one is COMPLETED. So yeah, if you are starting "timer" inside on_subscribe, then yes, when subscriber subscribes to first one, then timer started and and concat waits till observable send on_completed. Only after this one it would start subscribing to second one.

actually, flat_map is map+merge, merge is subscribing to everything "immediately" while concat keeps order and keep only 1 subscription at the same time.

[geiseri]

TL;DR: map+concat is the answer =)