Merge remote-tracking branch 'origin/4.0.x' into 4.1.x

Author: sbordet

src/main/java/org/eclipse/jetty/reactive/client/ReactiveRequest.java

@@ -18,11 +18,11 @@
 import java.nio.ByteBuffer;
 import java.nio.charset.Charset;
 import java.util.function.BiFunction;
-
 import org.eclipse.jetty.client.HttpClient;
 import org.eclipse.jetty.client.Request;
 import org.eclipse.jetty.io.Content.Chunk;
 import org.eclipse.jetty.reactive.client.internal.AdapterRequestContent;
+import org.eclipse.jetty.reactive.client.internal.BytesContent;
 import org.eclipse.jetty.reactive.client.internal.PublisherContent;
 import org.eclipse.jetty.reactive.client.internal.RequestEventPublisher;
 import org.eclipse.jetty.reactive.client.internal.ResponseEventPublisher;
@@ -79,7 +79,7 @@ protected ReactiveRequest(Request request) {
     }
 
     private ReactiveRequest(Request request, boolean abortOnCancel) {
-        this.request = request.listener(requestEvents)
+        this.request = request.onRequestListener(requestEvents)
                 .onResponseBegin(r -> {
                     this.response = new ReactiveResponse(this, r);
                 })
@@ -311,6 +311,25 @@ public default boolean rewind() {
             return false;
         }
 
+        /**
+         * <p>Creates a Content from the given {@code byte[]}.</p>
+         *
+         * @param bytes the request content
+         * @param contentType the request content type
+         * @return a Content wrapping the given {@code byte[]}
+         */
+        public static Content fromBytes(byte[] bytes, String contentType) {
+            return new BytesContent(bytes, contentType);
+        }
+
+        /**
+         * <p>Creates a Content from the given String.</p>
+         *
+         * @param string the request content
+         * @param mediaType the request content media type
+         * @param charset the request content charset
+         * @return a Content wrapping the given String
+         */
         public static Content fromString(String string, String mediaType, Charset charset) {
             return new StringContent(string, mediaType, charset);
         }

src/main/java/org/eclipse/jetty/reactive/client/internal/AdapterRequestContent.java

@@ -115,6 +115,7 @@ private Bridge() {
         @Override
         public void onSubscribe(Subscription s) {
             subscription = s;
+            // Do not demand here, the application controls the demand.
         }
 
         @Override
@@ -130,7 +131,9 @@ public void onNext(Content.Chunk c) {
                 demand = null;
             }
 
-            invoker.run(() -> invokeDemand(onDemand));
+            if (onDemand != null) {
+                invoker.run(() -> invokeDemand(onDemand));
+            }
         }
 
         @Override

src/main/java/org/eclipse/jetty/reactive/client/internal/BytesContent.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright (c) 2017 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.eclipse.jetty.reactive.client.internal;
+
+import java.nio.ByteBuffer;
+import org.eclipse.jetty.io.Content;
+import org.eclipse.jetty.reactive.client.ReactiveRequest;
+import org.reactivestreams.Subscriber;
+
+/**
+ * <p>Utility class that provides a {@code byte[]} as reactive content.</p>
+ */
+public class BytesContent extends AbstractSinglePublisher<Content.Chunk> implements ReactiveRequest.Content {
+    private final byte[] bytes;
+    private final String contentType;
+
+    public BytesContent(byte[] bytes, String contentType) {
+        this.bytes = bytes;
+        this.contentType = contentType;
+    }
+
+    @Override
+    public long getLength() {
+        return bytes.length;
+    }
+
+    @Override
+    public String getContentType() {
+        return contentType;
+    }
+
+    @Override
+    public boolean rewind() {
+        return true;
+    }
+
+    @Override
+    protected void onRequest(Subscriber<? super Content.Chunk> subscriber, long n) {
+        // The whole byte[] is sent at once, so this is the last chunk.
+        emitOnNext(subscriber, Content.Chunk.from(ByteBuffer.wrap(bytes), true));
+        emitOnComplete(subscriber);
+    }
+}

src/main/java/org/eclipse/jetty/reactive/client/internal/StringContent.java

@@ -15,46 +15,13 @@
  */
 package org.eclipse.jetty.reactive.client.internal;
 
-import java.nio.ByteBuffer;
 import java.nio.charset.Charset;
-import java.util.Objects;
-import org.eclipse.jetty.io.Content;
-import org.eclipse.jetty.reactive.client.ReactiveRequest;
-import org.reactivestreams.Subscriber;
 
 /**
  * <p>Utility class that provides a String as reactive content.</p>
  */
-public class StringContent extends AbstractSinglePublisher<Content.Chunk> implements ReactiveRequest.Content {
-    private final String mediaType;
-    private final Charset encoding;
-    private final byte[] bytes;
-
+public class StringContent extends BytesContent {
     public StringContent(String string, String mediaType, Charset encoding) {
-        this.mediaType = Objects.requireNonNull(mediaType);
-        this.encoding = Objects.requireNonNull(encoding);
-        this.bytes = string.getBytes(encoding);
-    }
-
-    @Override
-    public long getLength() {
-        return bytes.length;
-    }
-
-    @Override
-    public String getContentType() {
-        return mediaType + ";charset=" + encoding.name();
-    }
-
-    @Override
-    public boolean rewind() {
-        return true;
-    }
-
-    @Override
-    protected void onRequest(Subscriber<? super Content.Chunk> subscriber, long n) {
-        // The whole string is sent at once, so this is the last chunk.
-        emitOnNext(subscriber, Content.Chunk.from(ByteBuffer.wrap(bytes), true));
-        emitOnComplete(subscriber);
+        super(string.getBytes(encoding), mediaType + ";charset=" + encoding.name());
     }
 }

src/test/java/org/eclipse/jetty/reactive/client/ReactorTest.java

@@ -36,8 +36,10 @@
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 import org.springframework.http.HttpStatusCode;
+import org.springframework.http.MediaType;
 import org.springframework.http.client.reactive.JettyClientHttpConnector;
 import org.springframework.web.reactive.function.client.WebClient;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Hooks;
 import reactor.core.publisher.Mono;
 
@@ -70,6 +72,32 @@ public boolean handle(Request request, Response response, Callback callback) {
         assertArrayEquals(data, responseContent);
     }
 
+    @ParameterizedTest
+    @MethodSource("protocols")
+    public void testRequestWithContentResponseWithContent(String protocol) throws Exception {
+        byte[] data = new byte[1024];
+        new Random().nextBytes(data);
+        prepare(protocol, new Handler.Abstract() {
+            @Override
+            public boolean handle(Request request, Response response, Callback callback) {
+                Content.copy(request, response, callback);
+                return true;
+            }
+        });
+
+        ReactiveRequest.Content requestContent = ReactiveRequest.Content.fromBytes(data, "application/octet-stream");
+        WebClient client = WebClient.builder().clientConnector(new JettyClientHttpConnector(httpClient())).build();
+        byte[] responseContent = client.post()
+                .uri(uri())
+                .contentType(MediaType.parseMediaType(requestContent.getContentType()))
+                .body(Flux.from(requestContent).map(Content.Chunk::getByteBuffer), ByteBuffer.class)
+                .retrieve()
+                .bodyToMono(byte[].class)
+                .block();
+        assertNotNull(responseContent);
+        assertArrayEquals(data, responseContent);
+    }
+
     @ParameterizedTest
     @MethodSource("protocols")
     public void testTotalTimeout(String protocol) throws Exception {