src/RSocketResponder.cpp
/// Handles a new Channel requested by the other end.
Handles a new Channel requested by the other end.
Handles a new Channel requested by the other end .
yarpl::Reference<yarpl::flowable::Subscriber<Payload>>
RSocketResponder::handleRequestChannelCore(
    Payload request,
    StreamId streamId,
    const yarpl::Reference<yarpl::flowable::Subscriber<Payload>>&
    response) noexcept {
  class EagerSubscriberBridge
      : public yarpl::flowable::Subscriber<rsocket::Payload> {
   public:
    void onSubscribe(
        yarpl::Reference<yarpl::flowable::Subscription> subscription) noexcept {
      CHECK(!subscription_);
      subscription_ = std::move(subscription);
      if (inner_) {
        inner_->onSubscribe(subscription_);
      }
    }
    void onNext(rsocket::Payload element) noexcept {
      DCHECK(inner_);
      inner_->onNext(std::move(element));
    }
    void onComplete() noexcept {
      DCHECK(inner_);
      inner_->onComplete();
      inner_.reset();
      subscription_.reset();
    }
    void onError(std::exception_ptr ex) noexcept {
      DCHECK(inner_);
      inner_->onError(std::move(ex));
      inner_.reset();
      subscription_.reset();
    }
    void subscribe(
        yarpl::Reference<yarpl::flowable::Subscriber<rsocket::Payload>> inner) {
      CHECK(!inner_); 
      CHECK(inner);
      inner_ = std::move(inner);
      if (subscription_) {
        inner_->onSubscribe(subscription_);
      }
    }
   private:
    yarpl::Reference<yarpl::flowable::Subscriber<rsocket::Payload>> inner_;
    yarpl::Reference<yarpl::flowable::Subscription> subscription_;
  };
  auto eagerSubscriber = yarpl::make_ref<EagerSubscriberBridge>();
  auto flowable = handleRequestChannel(
      std::move(request),
      yarpl::flowable::Flowables::fromPublisher<Payload>([eagerSubscriber](
          yarpl::Reference<yarpl::flowable::Subscriber<Payload>> subscriber) {
        eagerSubscriber->subscribe(subscriber);
      }),
      std::move(streamId));
  flowable->subscribe(std::move(response));
  return eagerSubscriber;
}
yarpl :: Reference < yarpl :: flowable :: Subscriber < Payload > > RSocketResponder :: handleRequestChannelCore ( Payload request , StreamId streamId , const yarpl :: Reference < yarpl :: flowable :: Subscriber < Payload > > & response ) noexcept { class EagerSubscriberBridge : public yarpl :: flowable :: Subscriber < rsocket :: Payload > { public : void onSubscribe ( yarpl :: Reference < yarpl :: flowable :: Subscription > subscription ) noexcept { CHECK ( ! subscription_ ) ; subscription_ = std :: move ( subscription ) ; if ( inner_ ) { inner_ -> onSubscribe ( subscription_ ) ; } } void onNext ( rsocket :: Payload element ) noexcept { DCHECK ( inner_ ) ; inner_ -> onNext ( std :: move ( element ) ) ; } void onComplete ( ) noexcept { DCHECK ( inner_ ) ; inner_ -> onComplete ( ) ; inner_ . reset ( ) ; subscription_ . reset ( ) ; } void onError ( std :: exception_ptr ex ) noexcept { DCHECK ( inner_ ) ; inner_ -> onError ( std :: move ( ex ) ) ; inner_ . reset ( ) ; subscription_ . reset ( ) ; } void subscribe ( yarpl :: Reference < yarpl :: flowable :: Subscriber < rsocket :: Payload > > inner ) { CHECK ( ! inner_ ) ; CHECK ( inner ) ; inner_ = std :: move ( inner ) ; if ( subscription_ ) { inner_ -> onSubscribe ( subscription_ ) ; } } private : yarpl :: Reference < yarpl :: flowable :: Subscriber < rsocket :: Payload > > inner_ ; yarpl :: Reference < yarpl :: flowable :: Subscription > subscription_ ; } ; auto eagerSubscriber = yarpl :: make_ref < EagerSubscriberBridge > ( ) ; auto flowable = handleRequestChannel ( std :: move ( request ) , yarpl :: flowable :: Flowables :: fromPublisher < Payload > ( [ eagerSubscriber ] ( yarpl :: Reference < yarpl :: flowable :: Subscriber < Payload > > subscriber ) { eagerSubscriber -> subscribe ( subscriber ) ; } ) , std :: move ( streamId ) ) ; flowable -> subscribe ( std :: move ( response ) ) ; return eagerSubscriber ; }
Handles a new Channel requested by the other end.
Handles a new Channel requested by the other end .
// only one call to subscribe is supported // bridge from the existing eager RequestHandler and old Subscriber type // to the lazy Flowable and new Subscriber type