support for move-only types

Summary: universal references and perfect forwarding to the rescue

Test Plan: added demonstrative unit tests

Reviewed By: hans@fb.com

Subscribers: trunkagent, fugalh, njormrod, bmatheny

FB internal diff: D1592032

Tasks: 5283342

folly/experimental/wangle/rx/Observer.h

@@ -31,7 +31,7 @@ template <class T> class FunctionObserver;
 template <class T>
 struct Observer {
   // These are what it means to be an Observer.
-  virtual void onNext(T) = 0;
+  virtual void onNext(const T&) = 0;
   virtual void onError(Error) = 0;
   virtual void onCompleted() = 0;
 
@@ -79,7 +79,7 @@ struct Observer {
 /// make one of these directly - instead use the Observer::create() methods.
 template <class T>
 struct FunctionObserver : public Observer<T> {
-  typedef std::function<void(T)> OnNext;
+  typedef std::function<void(const T&)> OnNext;
   typedef std::function<void(Error)> OnError;
   typedef std::function<void()> OnCompleted;
 
@@ -92,7 +92,7 @@ struct FunctionObserver : public Observer<T> {
       onCompleted_(std::forward<C>(c))
   {}
 
-  void onNext(T val) override {
+  void onNext(const T& val) override {
     if (onNext_) onNext_(val);
   }
 

folly/experimental/wangle/rx/Subject.h

@@ -27,7 +27,7 @@ namespace folly { namespace wangle {
 template <class T>
 struct Subject : public Observable<T>, public Observer<T> {
   typedef typename Observable<T>::ObserversGuard ObserversGuard;
-  void onNext(T val) override {
+  void onNext(const T& val) override {
     ObserversGuard guard(this);
     for (auto& kv : Observable<T>::getObservers()) {
       kv.second->onNext(val);

folly/experimental/wangle/rx/test/RxTest.cpp

@@ -105,3 +105,43 @@ TEST(RxTest, SubscribeUnsubscribeDuringCallback) {
   EXPECT_EQ(2, outerCount);
   EXPECT_EQ(0, innerCount);
 }
+
+// Move only type
+typedef std::unique_ptr<int> MO;
+static MO makeMO() { return folly::make_unique<int>(1); }
+template <typename T>
+static ObserverPtr<T> makeMOObserver() {
+  return Observer<T>::create([](const T& mo) {
+    EXPECT_EQ(1, *mo);
+  });
+}
+
+TEST(RxTest, MoveOnlyRvalue) {
+  Subject<MO> subject;
+  auto s1 = subject.subscribe(makeMOObserver<MO>());
+  auto s2 = subject.subscribe(makeMOObserver<MO>());
+  auto mo = makeMO();
+  // Can't bind lvalues to rvalue references
+  // subject.onNext(mo);
+  subject.onNext(std::move(mo));
+  subject.onNext(makeMO());
+}
+
+// Copy only type
+struct CO {
+  CO() = default;
+  CO(const CO&) = default;
+  CO(CO&&) = delete;
+};
+
+template <typename T>
+static ObserverPtr<T> makeCOObserver() {
+  return Observer<T>::create([](const T& mo) {});
+}
+
+TEST(RxTest, CopyOnly) {
+  Subject<CO> subject;
+  auto s1 = subject.subscribe(makeCOObserver<CO>());
+  CO co;
+  subject.onNext(co);
+}