Merge pull request #268 from palantir/boxed-body

[stacked] Box streaming request bodies

changelog/@unreleased/pr-268.v2.yml

@@ -0,0 +1,5 @@
+type: break
+break:
+  description: Streaming request bodies are now passed by value to the client.
+  links:
+  - https://github.com/palantir/conjure-rust/pull/268

conjure-codegen/src/clients.rs

@@ -124,7 +124,7 @@ fn generate_endpoint(
     let where_ = where_(ctx, style, body_arg);
 
     let request = quote!(request_);
-    let setup_request = setup_request(ctx, body_arg, style, &request);
+    let setup_request = setup_request(ctx, def, body_arg, style, &request);
 
     let method = endpoint
         .http_method()
@@ -236,22 +236,22 @@ fn return_type_name(ctx: &Context, def: &ServiceDefinition, ty: &ReturnType<'_>)
 
 fn setup_request(
     ctx: &Context,
+    def: &ServiceDefinition,
     body_arg: Option<&ArgumentDefinition>,
     style: Style,
     request: &TokenStream,
 ) -> TokenStream {
     match body_arg {
         Some(body_arg) => {
             let name = ctx.field_name(body_arg.arg_name());
+            let box_ = ctx.box_ident(def.service_name());
             if ctx.is_binary(body_arg.type_()) {
                 match style {
                     Style::Sync => quote! {
-                        let mut #name = #name;
-                        let mut #request = conjure_http::private::encode_binary_request(&mut #name as _);
+                        let mut #request = conjure_http::private::encode_binary_request(#box_::new(#name));
                     },
                     Style::Async => quote! {
-                        conjure_http::private::pin_mut!(#name);
-                        let mut #request = conjure_http::private::async_encode_binary_request(#name as _);
+                        let mut #request = conjure_http::private::async_encode_binary_request(#box_::pin(#name));
                     },
                 }
             } else {

conjure-http/src/client.rs

@@ -109,7 +109,7 @@ pub enum RequestBody<'a, W> {
     /// A body already buffered in memory.
     Fixed(Bytes),
     /// A streaming body.
-    Streaming(&'a mut dyn WriteBody<W>),
+    Streaming(Box<dyn WriteBody<W> + 'a>),
 }
 
 /// The body of an async Conjure request.
@@ -119,7 +119,7 @@ pub enum AsyncRequestBody<'a, W> {
     /// A body already buffered in memory.
     Fixed(Bytes),
     /// A streaming body.
-    Streaming(Pin<&'a mut (dyn AsyncWriteBody<W> + Send)>),
+    Streaming(Pin<Box<dyn AsyncWriteBody<W> + 'a + Send>>),
 }
 
 /// A trait implemented by HTTP client implementations.

conjure-http/src/private/client/mod.rs

@@ -74,12 +74,12 @@ where
     request
 }
 
-pub fn encode_binary_request<W>(body: &mut dyn WriteBody<W>) -> Request<RequestBody<'_, W>> {
+pub fn encode_binary_request<W>(body: Box<dyn WriteBody<W> + '_>) -> Request<RequestBody<'_, W>> {
     inner_encode_binary_request(body, RequestBody::Streaming)
 }
 
 pub fn async_encode_binary_request<W>(
-    body: Pin<&mut (dyn AsyncWriteBody<W> + Send)>,
+    body: Pin<Box<dyn AsyncWriteBody<W> + '_ + Send>>,
 ) -> Request<AsyncRequestBody<'_, W>> {
     inner_encode_binary_request(body, AsyncRequestBody::Streaming)
 }

conjure-macros/src/lib.rs

@@ -172,7 +172,7 @@ mod path;
 ///     #[endpoint(method = POST, path = "/streamData")]
 ///     fn upload_stream(
 ///         &self,
-///         #[body(serializer = StreamingRequestSerializer)] body: &mut StreamingRequest,
+///         #[body(serializer = StreamingRequestSerializer)] body: StreamingRequest,
 ///     ) -> Result<(), Error>;
 ///
 ///     #[endpoint(method = GET, path = "/streamData", accept = StreamingResponseDeserializer)]
@@ -197,16 +197,16 @@ mod path;
 ///
 /// enum StreamingRequestSerializer {}
 ///
-/// impl<'a, W> SerializeRequest<'a, &'a mut StreamingRequest, W> for StreamingRequestSerializer
+/// impl<W> SerializeRequest<'static, StreamingRequest, W> for StreamingRequestSerializer
 /// where
 ///     W: Write,
 /// {
-///     fn content_type(_: &&mut StreamingRequest) -> HeaderValue {
+///     fn content_type(_: &StreamingRequest) -> HeaderValue {
 ///         HeaderValue::from_static("text/plain")
 ///     }
 ///
-///     fn serialize(value: &'a mut StreamingRequest) -> Result<RequestBody<'a, W>, Error> {
-///         Ok(RequestBody::Streaming(value))
+///     fn serialize(value: StreamingRequest) -> Result<RequestBody<'static, W>, Error> {
+///         Ok(RequestBody::Streaming(Box::new(value)))
 ///     }
 /// }
 ///

conjure-test/src/test/clients.rs

@@ -112,7 +112,7 @@ impl<'b> Client for &'b TestClient {
         let body = match req.into_body() {
             RequestBody::Empty => TestBody::Empty,
             RequestBody::Fixed(body) => TestBody::Json(String::from_utf8(body.to_vec()).unwrap()),
-            RequestBody::Streaming(body) => {
+            RequestBody::Streaming(mut body) => {
                 let mut buf = vec![];
                 body.write_body(&mut buf).unwrap();
                 TestBody::Streaming(buf)
@@ -641,7 +641,7 @@ where
     #[endpoint(method = POST, path = "/test/streamingRequest")]
     fn streaming_request(
         &self,
-        #[body(serializer = RawRequestSerializer)] body: &mut RawRequest,
+        #[body(serializer = RawRequestSerializer)] body: RawRequest,
     ) -> Result<(), Error>;
 
     #[endpoint(method = GET, path = "/test/streamingResponse", accept = RawResponseDeserializer)]
@@ -665,16 +665,16 @@ where
 
 enum RawRequestSerializer {}
 
-impl<'a, W> SerializeRequest<'a, &'a mut RawRequest, W> for RawRequestSerializer
+impl<W> SerializeRequest<'static, RawRequest, W> for RawRequestSerializer
 where
     W: Write,
 {
-    fn content_type(_: &&mut RawRequest) -> HeaderValue {
+    fn content_type(_: &RawRequest) -> HeaderValue {
         HeaderValue::from_static("text/plain")
     }
 
-    fn serialize(value: &'a mut RawRequest) -> Result<RequestBody<'a, W>, Error> {
-        Ok(RequestBody::Streaming(value))
+    fn serialize(value: RawRequest) -> Result<RequestBody<'static, W>, Error> {
+        Ok(RequestBody::Streaming(Box::new(value)))
     }
 }
 
@@ -697,7 +697,7 @@ fn custom_streaming_request() {
         .body(TestBody::Streaming(b"hello world".to_vec()));
 
     CustomStreamingServiceClient::new(&client)
-        .streaming_request(&mut RawRequest)
+        .streaming_request(RawRequest)
         .unwrap();
 }