如何在 rxcpp 中处理请求/响应流

Question

我需要在rxcpp中实现一个相机采样系统。我想象的方式是将requestStream作为参数传递并接收响应流

每次调用 requestSample 时，都会创建一个新的相机会话，当为 requestStream 调用 on_complete(( 时，相机会话将终止

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

Answer 1

s是一个局部堆栈变量，但 lambda 正在捕获对s的引用。在调用 lambda 时，此引用无效。将[&]更改为[=]，这应该有效！

另一种选择是使用现有的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;
})
;
}