Add folly::coro::accumulate algorithm

Summary: The `accumulate` algorithm will behave similarly to `std::accumulate`, but instead returns a `Task` of the accumulated result from the values in the `AsyncGenerator`

Reviewed By: kirkshoop

Differential Revision: D18714284

fbshipit-source-id: 66625cd1347d767467d98a421b78b6f45d70260c

folly/experimental/coro/Accumulate-inl.h

+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <folly/experimental/coro/Accumulate.h>
+
+namespace folly {
+namespace coro {
+
+template <typename Reference, typename Value, typename Output>
+Task<Output> accumulate(
+    AsyncGenerator<Reference, Value> generator,
+    Output init) {
+  return accumulate(std::move(generator), std::move(init), std::plus{});
+}
+
+template <
+    typename Reference,
+    typename Value,
+    typename Output,
+    typename BinaryOp>
+Task<Output> accumulate(
+    AsyncGenerator<Reference, Value> generator,
+    Output init,
+    BinaryOp op) {
+  while (auto next = co_await generator.next()) {
+    init = op(std::move(init), std::move(next).value());
+  }
+  co_return init;
+}
+} // namespace coro
+} // namespace folly

folly/experimental/coro/Accumulate.h

+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <folly/experimental/coro/AsyncGenerator.h>
+#include <folly/experimental/coro/Task.h>
+
+namespace folly {
+namespace coro {
+
+// Accumulate the values from an input stream into a single value given
+// an optional binary accumulation operation, similar to std::accumulate.
+//
+// The input is a stream of values.
+//
+// The output is a Task containing the result of the accumulation
+//
+// Example:
+//   AsyncGenerator<int> stream();
+//
+//   Task<void> consumer() {
+//     auto sum = accumulate(stream(), 0, std::plus{});
+//   }
+template <typename Reference, typename Value, typename Output>
+Task<Output> accumulate(
+    AsyncGenerator<Reference, Value> generator,
+    Output init);
+
+template <
+    typename Reference,
+    typename Value,
+    typename Output,
+    typename BinaryOp>
+Task<Output> accumulate(
+    AsyncGenerator<Reference, Value> generator,
+    Output init,
+    BinaryOp op);
+
+} // namespace coro
+} // namespace folly
+
+#include <folly/experimental/coro/Accumulate-inl.h>

folly/experimental/coro/test/AccumulateTest.cpp

+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <folly/Portability.h>
+
+#if FOLLY_HAS_COROUTINES
+
+#include <folly/experimental/coro/Accumulate.h>
+#include <folly/experimental/coro/BlockingWait.h>
+
+#include <folly/portability/GTest.h>
+
+using namespace folly::coro;
+
+AsyncGenerator<int> generateInts(int begin, int end) {
+  for (int i = begin; i < end; i++) {
+    co_await co_reschedule_on_current_executor;
+    co_yield i;
+  }
+}
+
+TEST(AccumulateTest, NoOperationProvided) {
+  auto result = blockingWait(accumulate(generateInts(0, 5), 0));
+  auto expected = 0 + 1 + 2 + 3 + 4;
+
+  EXPECT_EQ(result, expected);
+}
+
+TEST(AccumulateTest, OperationProvided) {
+  auto result =
+      blockingWait(accumulate(generateInts(1, 5), 1, std::multiplies{}));
+  auto expected = 1 * 2 * 3 * 4;
+
+  EXPECT_EQ(result, expected);
+}
+
+#endif