Merge pull request #864 from cloudflare/jsnell/coroutines-conversion-blobinputstream

Author: jasnell

src/workerd/api/blob.c++

@@ -144,13 +144,23 @@ public:
         blob(kj::mv(blob)) {}
 
   kj::Promise<size_t> tryRead(void* buffer, size_t minBytes, size_t maxBytes) override {
+    // Attempt to read a maximum of maxBytes from the remaining unread content of the blob
+    // into the given buffer. It is the caller's responsibility to ensure that buffer has
+    // enough capacity for at least maxBytes.
+    // The minBytes argument is ignored in this implementation of tryRead.
+    // The buffer must be kept alive by the caller until the returned promise is fulfilled.
+    // The returned promise is fulfilled with the actual number of bytes read.
     size_t amount = kj::min(maxBytes, unread.size());
-    memcpy(buffer, unread.begin(), amount);
-    unread = unread.slice(amount, unread.size());
+    if (amount > 0) {
+      memcpy(buffer, unread.begin(), amount);
+      unread = unread.slice(amount, unread.size());
+    }
     return amount;
   }
 
   kj::Maybe<uint64_t> tryGetLength(StreamEncoding encoding) override {
+    // Returns the number of bytes remaining to be read for the given encoding if that
+    // encoding is supported. This implementation only supports StreamEncoding::IDENTITY.
     if (encoding == StreamEncoding::IDENTITY) {
       return unread.size();
     } else {
@@ -159,20 +169,22 @@ public:
   }
 
   kj::Promise<DeferredProxy<void>> pumpTo(WritableStreamSink& output, bool end) override {
-    if (unread.size() == 0) {
-      return addNoopDeferredProxy(kj::READY_NOW);
-    }
+    // Write all of the remaining unread content of the blob to output.
+    // If end is true, output.end() will be called once the write has been completed.
+    // Importantly, the WritableStreamSink must be kept alive by the caller until the
+    // returned promise is fulfilled.
+    if (unread.size() != 0) {
+      auto promise = output.write(unread.begin(), unread.size());
+      unread = nullptr;
 
-    auto promise = output.write(unread.begin(), unread.size());
-    unread = nullptr;
+      co_await promise;
 
-    if (end) {
-      promise = promise.then([&output]() { return output.end(); });
+      if (end) co_await output.end();
     }
 
     // We can't defer the write to the proxy stage since it depends on `blob` which lives in the
     // isolate.
-    return addNoopDeferredProxy(kj::mv(promise));
+    co_return newNoopDeferredProxy();
   }
 
 private:

src/workerd/api/util.h

@@ -132,15 +132,24 @@ struct DeferredProxy {
   kj::Promise<T> proxyTask;
 };
 
+inline DeferredProxy<void> newNoopDeferredProxy() {
+  return DeferredProxy<void> { kj::READY_NOW };
+}
+
+template <typename T>
+inline DeferredProxy<T> newNoopDeferredProxy(T&& value) {
+  return DeferredProxy<T> { kj::mv(value) };
+}
+
 template <typename T>
 inline kj::Promise<DeferredProxy<T>> addNoopDeferredProxy(kj::Promise<T> promise) {
   // Helper method to use when you need to return `Promise<DeferredProxy<T>>` but no part of the
   // operation you are returning is eligible to be deferred past the IoContext lifetime.
-
-  return promise.then([](T&& value) { return DeferredProxy<T> { kj::mv(value) }; });
+  co_return newNoopDeferredProxy(co_await promise);
 }
 inline kj::Promise<DeferredProxy<void>> addNoopDeferredProxy(kj::Promise<void> promise) {
-  return promise.then([]() { return DeferredProxy<void> { kj::READY_NOW }; });
+  co_await promise;
+  co_return newNoopDeferredProxy();
 }
 
 kj::Maybe<jsg::V8Ref<v8::Object>> cloneRequestCf(