Как обрабатывать поток запросов / ответов в rxcpp
Мне нужно реализовать систему выборки камеры в rxcpp. То, как я себе это представлял, это передача requestStream в качестве параметра и получение responseStream.
Каждый раз, когда вызывается requestSample, создается новый сеанс камеры, а когда вызывается on_complete () для requestStream, сеанс камеры завершается.
observable<ImageSample> requestSampleStream(observable<ImageRequest> requestStream$) {
auto response$ = rxcpp::observable<>::create<ImageSample>(
[&](rxcpp::subscriber<ImageSample> s){
auto request_next = [&](ImageRequest imageRequest) {
cout << "image request next" << endl;
SampleImage vsi;
s.on_next(vsi);
};
auto request_completed = [&] {
cout << "no more samples needed.. close camera" << endl;
s.on_completed();
};
auto request_error = [&](std::exception_ptr e) {
try { rethrow_exception(e); }
catch (const exception &ex) {
cerr << "error happened on request stream.. close the camera and send error on return stream" << endl << ex.what() << endl;
}
s.on_error(e);
};
requestStream$.subscribe(request_next,
request_error,
request_completed
);
});
return response$;
}
Проблема в том, что к тому времени requestStream$.subscribe(request_next.. называется, я получаю EXC_BAD_ACCESS
Решение
s является локальной переменной стека, но лямбда-выражения захватывают ссылку на s, Эта ссылка недействительна ко времени вызова лямбды. менять [&] в [=] и это должно работать!
Другой вариант заключается в использовании существующих map алго.
observable<ImageSample> requestSampleStream(observable<ImageRequest> requestStream$) {
return requestStream$ |
map([](ImageRequest imageRequest){
cout << "image request next" << endl;
SampleImage vsi;
s.on_next(vsi);
})
// optional
| tap([&](std::exception_ptr e) {
try { rethrow_exception(e); }
catch (const exception &ex) {
cerr << "error happened on request stream.. close the camera and send error on return stream" << endl << ex.what() << endl;
}
},
[](){
cout << "no more samples needed.. close camera" << endl;
})
;
}
Другие решения
Других решений пока нет …