Coroutines Conversion: Migrate streams compression to coroutines (#866)

* Migrate streams compression to coroutines

* Simplify exception throw

Co-authored-by: Kenton Varda <kenton@cloudflare.com>

* Simplify if statement

Co-authored-by: James M Snell <jsnell@cloudflare.com>

---------

Co-authored-by: Kenton Varda <kenton@cloudflare.com>
Co-authored-by: James M Snell <jsnell@cloudflare.com>

Author: 3 people

src/workerd/api/streams/compression.c++

@@ -165,16 +165,15 @@ public:
   kj::Promise<void> write(kj::ArrayPtr<const kj::ArrayPtr<const kj::byte>> pieces) override {
     KJ_SWITCH_ONEOF(state) {
       KJ_CASE_ONEOF(ended, Ended) {
-        return JSG_KJ_EXCEPTION(FAILED, Error, "Write after close.");
-      }
-      KJ_CASE_ONEOF(exception, kj::Exception) {
-        return kj::cp(exception);
+        JSG_FAIL_REQUIRE(Error, "Write after close");
       }
+      KJ_CASE_ONEOF(exception, kj::Exception) { kj::throwFatalException(kj::cp(exception)); }
       KJ_CASE_ONEOF(open, Open) {
-        if (pieces.size() == 0) return kj::READY_NOW;
-        return write(pieces[0].begin(), pieces[0].size()).then([this, pieces]() {
-          return write(pieces.slice(1, pieces.size()));
-        });
+        if (pieces.size() != 0) {
+          co_await write(pieces[0].begin(), pieces[0].size());
+          co_await write(pieces.slice(1, pieces.size()));
+        }
+        co_return;
       }
     }
     KJ_UNREACHABLE;