如何在 rxcpp 中处理 request/response 流

Question

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

每次调用 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;
        })
        ;
}

如何在 rxcpp 中处理 request/response 流

How to handle a request/response stream in rxcpp

c++

reactive-programming

c++11

rxcpp