Fix memory leaks in OldNewBridge.h (#371)

Places where we weren't releasing the yarpl Reference types.

Adds an operator bool() to Reference for convenience, most smart pointer types
are expected to have them.

Running a local stream-hello-world example server and client no longer reports
any leaks when built with LSAN.

experimental/rsocket/OldNewBridge.h

@@ -24,11 +24,16 @@ class NewToOldSubscription : public yarpl::flowable::Subscription {
   ~NewToOldSubscription() = default;
 
   void request(int64_t n) override {
-    inner_->request(n);
+    if (inner_) {
+      inner_->request(n);
+    }
   }
 
   void cancel() override {
     inner_->cancel();
+
+    inner_.reset();
+    release();
   }
 
  private:
@@ -55,10 +60,16 @@ class OldToNewSubscriber
 
   void onComplete() noexcept {
     inner_->onComplete();
+
+    inner_.reset();
+    bridge_.reset();
   }
 
   void onError(folly::exception_wrapper ex) noexcept {
     inner_->onError(ex.to_exception_ptr());
+
+    inner_.reset();
+    bridge_.reset();
   }
 
  private:
@@ -74,24 +85,24 @@ class OldToNewSubscription : public reactivesocket::Subscription {
       : inner_{inner} {}
 
   void request(size_t n) noexcept override {
-    if (!terminated_) {
+    if (inner_) {
       inner_->request(n);
     }
   }
 
   void cancel() noexcept override {
-    if (!terminated_) {
+    if (inner_) {
       inner_->cancel();
     }
+    inner_.reset();
   }
 
   void terminate() {
-    terminated_ = true;
+    inner_.reset();
   }
 
  private:
   yarpl::Reference<yarpl::flowable::Subscription> inner_{nullptr};
-  bool terminated_{false};
 };
 
 class NewToOldSubscriber : public yarpl::flowable::Subscriber<reactivesocket::Payload> {
@@ -120,13 +131,19 @@ class NewToOldSubscriber : public yarpl::flowable::Subscriber<reactivesocket::Pa
       bridge_->terminate();
     }
     inner_->onComplete();
+
+    inner_.reset();
+    bridge_.reset();
   }
 
   void onError(std::exception_ptr eptr) override {
     if (bridge_) {
       bridge_->terminate();
     }
     inner_->onError(folly::exception_wrapper(std::move(eptr)));
+
+    inner_.reset();
+    bridge_.reset();
   }
 
  private:

experimental/yarpl/include/yarpl/Refcounted.h

@@ -117,6 +117,10 @@ class Reference {
     Reference().swap(*this);
   }
 
+  explicit operator bool() const {
+    return pointer_;
+  }
+
  private:
   void swap(Reference& other) {
     T* temp = pointer_;