puzzled : using observe_on(new_thread) & buffer in different order

[0xMonad]

Thank you for creating and maintaining this project! It's very helpful to my work. I have encountered some problems in the process of using, and hope to get your help. Thank you again for your hard work!

env:

gcc version 11.4.0 (Ubuntu 11.4.0-1ubuntu1~22.04) C++20

code

#include <exception>
#include <fmt/core.h>
#include <fmt/format.h>
#include <rpp/rpp.hpp>

using namespace rpp;

int main()
{
    auto obs = source::just(1, 2, 3, 4, 5, 6, 7);

    // clang-format off
    obs    
        | ops::observe_on(schedulers::new_thread{})
        | ops::buffer(2)
        | ops::subscribe(
            [](const std::vector<int>& v) {
                fmt::println("A size: {} val: {}", v.size() ,fmt::join(v, "-"));
            }, 
            [](const std::exception_ptr& e) {}, 
            []() { fmt::println(" complete "); });

    obs | ops::buffer(2)
        | ops::observe_on(schedulers::new_thread{}, std::chrono::milliseconds(100))
        | ops::subscribe(
            [](const std::vector<int>& v) {
                fmt::println("B size {} size val: {}", v.size(), fmt::join(v, "-"));
            }, 
            [](const std::exception_ptr& e) {}, 
            []() { fmt::println(" complete "); });

    // clang-format on
    pause();

    return 0;
}

output

A size: 2 val: 1-2
A size: 2 val: 3-4
A size: 2 val: 5-6
A size: 1 val: 7
 complete 
B size 2 size val: 1-2
B size 1 size val: 3
B size 1 size val: 4
B size 1 size val: 5
B size 1 size val: 6
B size 1 size val: 7
 complete 

question

I think case A/B should have the same output. but actually different.

Or is there something wrong with my usage.

[AlexInLog]

Hey @GitJQ !
Thanks for using and you are welcome =)

yeah, this case looks pretty interesting... i would expect they should be exactly the same.. I've reproduced it locally, have to debug a bit

[AlexInLog]

UPD: looks like i've found where issue is appeared, going to prepare fix today

[AlexInLog]

Fixed in v2 branch