RXcpp поток данных из функции непрерывно

Question

Я учусь реактивно для C++, и я ищу некоторые рекомендации. Я создал функцию, чтобы ждать события, а затем вернуть его. Я хочу поймать все события, которые происходят с реактивной асинхронизацией, и обрабатывать их по мере их появления. Здесь не то, что я до сих пор:

int Engine::Initialize() 
{ 
    al_init(); 

    display = al_create_display(1200, 800); 

    eventQueue = al_create_event_queue(); 

    al_install_mouse(); 
    al_install_keyboard(); 

    al_register_event_source(eventQueue, al_get_keyboard_event_source()); 
    al_register_event_source(eventQueue, al_get_mouse_event_source()); 

//test: wait for 2 events to happen 
    auto events = rxcpp::observable<>::create([](rxcpp::subscriber<ALLEGRO_EVENT> e) 
    { 
     e.on_next(WaitForEvent); 
     e.on_next(WaitForEvent); 
     e.on_completed(); 
    }); 

    events.subscribe([](ALLEGRO_EVENT e) 
    { 
     printf("We have an Event: %d \n", e.type); 
    }, 

    []() 
    { 
     printf("Done\n"); 
    }); 

    return 0; 
} 

ALLEGRO_EVENT Engine::WaitForEvent() 
{ 
    ALLEGRO_EVENT event; 

    al_wait_for_event(eventQueue, &event); 

    return event; 
} 

Кажется, я получаю ошибку: не экземпляр шаблона функции «rxcpp :: наблюдаемым :: создать» совпадает со списком аргументов. Нужно ли мне создавать свой собственный шаблон или что-то, чтобы иметь возможность наблюдать за ALLEGRO_EVENT?

Answer 1

От the docs он выглядит так: create использует явный параметр шаблона для возвращаемого типа, поэтому вам необходимо его предоставить. Ваш слушатель ничего не возвращает, поэтому void будет работать.

auto events = rxcpp::observable<>::create<void>(... 

Answer 2

@tim имеет правильную идею.

Я отправляю ответ от наших discussion

create<>() необходимо знать тип, который будет передан в on_next(). Кроме того, WaitForEvent является функцией, но on_next() ожидает ALLEGRO_EVENT поэтому убедитесь, что позвонить WaitForEvent() и передать результат в on_next()

auto events = rxcpp::observable<>::create<ALLEGRO_EVENT>([this](rxcpp::subscriber<ALLEGRO_EVENT> e) 
{ 
    e.on_next(this->WaitForEvent()); 
    e.on_next(this->WaitForEvent()); 
    e.on_completed(); 
});