如何rxcpp

Question

处理请求/响应流，我需要在rxcpp实现相机采样系统..我想象这是传递一个requestStream为PARAM和接收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

Answer 1

s是本地堆栈变量，但lambdas正在捕获对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; 
     }) 
     ; 
}