Add response_body function to r2::ObjectBody. (#462)

This commit adds a new function to `r2::ObjectBody` to allow
the underlying `ReadableStream` for the body of an R2 object to
be used to create a `ResponseBody`, via the `ResponseBody::Stream`
variant.

This adds support for cases where workers can hand a reference to
the R2 object to the workers runtime for streaming to the client,
unmodified. Importantly, this allows the worker to avoid CPU usage
while streaming the object.

Before this commit, `ObjectBody::Stream` could be used but that
would result in unexpected CPU usage during the streaming process
leading to the working hitting the CPU limit for large (> 150 MB)
R2 objects.

Closes #389

Author: mcnulty

worker/src/r2/mod.rs

@@ -12,7 +12,9 @@ use worker_sys::{
     R2UploadedPart as EdgeR2UploadedPart,
 };
 
-use crate::{env::EnvBinding, ByteStream, Date, Error, FixedLengthStream, Headers, Result};
+use crate::{
+    env::EnvBinding, ByteStream, Date, Error, FixedLengthStream, Headers, ResponseBody, Result,
+};
 
 mod builder;
 
@@ -282,6 +284,19 @@ impl<'body> ObjectBody<'body> {
         })
     }
 
+    /// Returns a [ResponseBody] containing the data in the [Object].
+    ///
+    /// This function can be used to hand off the [Object] data to the workers runtime for streaming
+    /// to the client in a [crate::Response]. This ensures that the worker does not consume CPU time
+    /// while the streaming occurs, which can be significant if instead [ObjectBody::stream] is used.
+    pub fn response_body(self) -> Result<ResponseBody> {
+        if self.inner.body_used() {
+            return Err(Error::BodyUsed);
+        }
+
+        Ok(ResponseBody::Stream(self.inner.body()))
+    }
+
     pub async fn bytes(self) -> Result<Vec<u8>> {
         let js_buffer = JsFuture::from(self.inner.array_buffer()).await?;
         let js_buffer = Uint8Array::new(&js_buffer);