Stream throws undeclared exception

Summary: Test undeclared exceptions in the stream

Reviewed By: sazonovkirill

Differential Revision: D85378813

fbshipit-source-id: 3bf3eb8d877bc4e75c1445e7655051db7b62d32e

Author: evanjzou

third-party/thrift/src/thrift/lib/cpp2/async/tests/BiDiServiceE2ETest.cpp

@@ -344,6 +344,57 @@ CO_TEST_F(BiDiServiceE2ETest, StreamThrowsDeclaredException) {
   co_await folly::coro::collectAll(std::move(sinkTask), std::move(streamTask));
 }
 
+CO_TEST_F(BiDiServiceE2ETest, StreamThrowsUndeclaredException) {
+  constexpr int64_t kStreamItemsUntilThrow = 500;
+
+  struct Handler : public ServiceHandler<detail::test::TestBiDiService> {
+    folly::coro::Task<StreamTransformation<int64_t, int64_t>> co_canThrow()
+        override {
+      co_return StreamTransformation<int64_t, int64_t>{
+          [](folly::coro::AsyncGenerator<int64_t&&>)
+              -> folly::coro::AsyncGenerator<int64_t&&> {
+            for (std::size_t i = 0; i < kStreamItemsUntilThrow; ++i) {
+              co_yield int64_t(i);
+            }
+            throw std::runtime_error{"For test"};
+          }};
+    }
+  };
+
+  testConfig({std::make_shared<Handler>()});
+  auto client = makeClient<detail::test::TestBiDiService>();
+  BidirectionalStream<int64_t, int64_t> stream = co_await client->co_canThrow();
+
+  auto sinkGen =
+      folly::coro::co_invoke([]() -> folly::coro::AsyncGenerator<int64_t&&> {
+        for (int64_t i = 0; i < 1000; ++i) {
+          co_yield int64_t(i);
+        }
+      });
+  auto sinkTask = folly::coro::co_invoke(
+      [clientSink = std::move(stream.sink),
+       sinkGen = std::move(sinkGen)]() mutable -> folly::coro::Task<void> {
+        co_await std::move(clientSink).sink(std::move(sinkGen));
+      });
+
+  auto streamTask = folly::coro::co_invoke(
+      [&, streamGen = std::move(stream.stream).toAsyncGenerator()]() mutable
+      -> folly::coro::Task<void> {
+        for (int64_t i = 0; i < kStreamItemsUntilThrow; ++i) {
+          auto next = co_await streamGen.next();
+          if (!next) {
+            CO_FAIL() << fmt::format(
+                "Did not receive all stream elements, expected {} but got {}",
+                kStreamItemsUntilThrow,
+                i);
+          }
+        }
+        EXPECT_THROW(
+            co_await streamGen.next(), apache::thrift::TApplicationException);
+      });
+  co_await folly::coro::collectAll(std::move(sinkTask), std::move(streamTask));
+}
+
 CO_TEST_F(BiDiServiceE2ETest, IgnoreInputProduceOutput) {
   constexpr int64_t kTestLimit = 100;