More fun with ASIO

[geiseri]

I am not sure you can adequately answer this since I am not sure about your familiarity with boost::asio.

This is an evolution of #323 in that I would use the rpp::operators::subscribe(...) to create an async operation based on the
completion handler. So extending the example:

auto job1 = rpp::source::from_iterable<rpp::memory_model::use_shared>(series)
...
.subscribe(asio::as_future);
std::vector<std::string> result = job1.get();

With the asio completion handlers you can have different adapters. In the above case it would be a std::future<std::vector<std::string>> that would defer execution until get() is called.

I am not sure this makes sense so far, but I think the core question is what magic is needed to pass to subscribe that would allow me to get the std::future<> back?

[victimsnino]

Hey @geiseri !
Yeah, unfortunately i'm not really good at boost::asio.
Anyway, looks like you are doing something strange or at least not clear for me.
First of all rpp::source::from_iterable is observable which would emit values from series 1 by 1, so, final value can't be a vector at your example.
Second: it is not how rx works =) subscribe is only subscribing observer or callbacks to provided observable. So, in your case asio::as_future is actually (most probably) callback for on_next over emitted value (for example, std::string). So, it is how rx works: you subscribe observer/callbacks to emissions and just do nothing due to your callbacks/observer would handle emissions itself.

Actually, you should do your logic inside this callback instead. Actually i think you should do something like this instead

 rpp::source::from_iterable<rpp::memory_model::use_shared>(series)
...
.map(asio::as_future) // change value to future
.subscribe([](future<...> v){ v.get();});

If you still need to extract value and it is really what you want, you have to convert observable to blocking_observable to guarantee completion of observable till exit from subscribe call. Then you could do something like

std::future<std::vector<std::string>> job1{};

rpp::source::from_iterable<rpp::memory_model::use_shared>(series)
...
.map(asio::as_future)
.as_blocking()
.subscribe([&job1](std::future<std::vector<std::string>> v) { job1=v;});

std::vector<std::string> result = job1.get();

actually i have plans to extend blocking observable with kind of get for such an case but anyway it would looks like this

[geiseri]

You do bring up a good point about the actual flow:

Second: it is not how rx works =) subscribe is only subscribing observer or callbacks to provided observable. So, in your case asio::as_future is actually (most probably) callback for on_next over emitted value (for example, std::string). So, it is how rx works: you subscribe observer/callbacks to emissions and just do nothing due to your callbacks/observer would handle emissions itself.

I think I was too focused on an "implementation detail" that ignored the core desire. You are correct about the callback part and in my case I think I would like to be able to use a coroutine vs the callbacks. The code flow I have now would have a generator that will co_yield a connection, then a co_await to read the message, use transform/filter on the message, and finally co_await the write response. So at the end of the day the .subscribe is just to start the async operation of the pipeline and handle any sort of error. I guess similar to the rxjs from creates an observable from a Promise.

Is something like that even possible? I have gotten the impression that the actual coroutine task type is an implementation detail left to someone else. I assume any sort of implementation would need to be special to the scheduler used. Or is this something as simple as being able to handle a std::coroutine_handle's with .as_blocking()?

[victimsnino]

I think you are mixing two different worlds a bit =)
If i get your task right, you have something like

Generator source() {
   while(...) {
   ...
       co_yield socket.get_accepted_connection();
   }
}

Task main() {
   const auto gen = source();
   for (auto connection : gen) // gen does .resume/.next underhood
   {
      const auto request = co_await connection.handle();
      if (request == "Hello") {
            co_await connection.respond("world!");
      }
   }
}

and you want to wrap it to rx world. Could you extend/provide more details and provide imaginary usage example of what you are talking about?