Add support for in-place creation of NotificationQueue::Consumer

Summary: as title, this comes handy when multiple consumers are needed
within the same class.

Reviewed By: has

Differential Revision: D2530249

fb-gh-sync-id: 942761782a100b2d3fe54d94a7c1b0e03b95a847

folly/io/async/NotificationQueue.h

@@ -77,6 +77,11 @@ class NotificationQueue {
         destroyedFlagPtr_(nullptr),
         maxReadAtOnce_(kDefaultMaxReadAtOnce) {}
 
+    // create a consumer in-place, without the need to build new class
+    template <typename TCallback>
+    static std::unique_ptr<Consumer, DelayedDestruction::Destructor> make(
+        TCallback&& callback);
+
     /**
      * messageAvailable() will be invoked whenever a new
      * message is available from the pipe.
@@ -798,4 +803,49 @@ bool NotificationQueue<MessageT>::Consumer::consumeUntilDrained(
   return true;
 }
 
+/**
+ * Creates a NotificationQueue::Consumer wrapping a function object
+ * Modeled after AsyncTimeout::make
+ *
+ */
+
+namespace detail {
+
+template <typename MessageT, typename TCallback>
+struct notification_queue_consumer_wrapper
+    : public NotificationQueue<MessageT>::Consumer {
+
+  template <typename UCallback>
+  explicit notification_queue_consumer_wrapper(UCallback&& callback)
+      : callback_(std::forward<UCallback>(callback)) {}
+
+  // we are being stricter here and requiring noexcept for callback
+  void messageAvailable(MessageT&& message) override {
+    static_assert(
+      noexcept(std::declval<TCallback>()(std::forward<MessageT>(message))),
+      "callback must be declared noexcept, e.g.: `[]() noexcept {}`"
+    );
+
+    callback_(std::forward<MessageT>(message));
+  }
+
+ private:
+  TCallback callback_;
+};
+
+} // namespace detail
+
+template <typename MessageT>
+template <typename TCallback>
+std::unique_ptr<typename NotificationQueue<MessageT>::Consumer,
+                DelayedDestruction::Destructor>
+NotificationQueue<MessageT>::Consumer::make(TCallback&& callback) {
+  return std::unique_ptr<NotificationQueue<MessageT>::Consumer,
+                         DelayedDestruction::Destructor>(
+      new detail::notification_queue_consumer_wrapper<
+          MessageT,
+          typename std::decay<TCallback>::type>(
+          std::forward<TCallback>(callback)));
+}
+
 } // folly

folly/io/async/test/NotificationQueueTest.cpp

@@ -644,3 +644,21 @@ TEST(NotificationQueueTest, UseAfterFork) {
   EXPECT_EQ(5678, consumer.messages.front());
   consumer.messages.pop_front();
 }
+
+TEST(NotificationQueueConsumer, make) {
+  int value = 0;
+  EventBase evb;
+  NotificationQueue<int> queue(32);
+
+  auto consumer = decltype(queue)::Consumer::make([&](
+      int&& msg) noexcept { value = msg; });
+
+  consumer->startConsuming(&evb, &queue);
+
+  int const newValue = 10;
+  queue.tryPutMessage(newValue);
+
+  evb.loopOnce();
+
+  EXPECT_EQ(newValue, value);
+}