Add SurroundingPublisher and HttpResponse.of(headers, pub, trailers)

core/src/main/java/com/linecorp/armeria/common/HttpRequest.java

@@ -48,6 +48,7 @@
 import com.linecorp.armeria.common.stream.StreamMessage;
 import com.linecorp.armeria.internal.common.DefaultHttpRequest;
 import com.linecorp.armeria.internal.common.DefaultSplitHttpRequest;
+import com.linecorp.armeria.internal.common.stream.SurroundingPublisher;
 import com.linecorp.armeria.unsafe.PooledObjects;
 
 import io.netty.buffer.ByteBufAllocator;
@@ -280,6 +281,22 @@ static HttpRequest of(RequestHeaders headers, Publisher<? extends HttpObject> pu
         }
     }
 
+    /**
+     * Creates a new instance from an existing {@link RequestHeaders}, {@link Publisher} and trailers.
+     *
+     * <p>Note that the {@link HttpData}s in the {@link Publisher} are not released when
+     * {@link Subscription#cancel()} or {@link #abort()} is called. You should add a hook in order to
+     * release the elements. See {@link PublisherBasedStreamMessage} for more information.
+     */
+    static HttpRequest of(RequestHeaders headers,
+                          Publisher<? extends HttpData> publisher,
+                          HttpHeaders trailers) {
+        requireNonNull(headers, "headers");
+        requireNonNull(publisher, "publisher");
+        requireNonNull(trailers, "trailers");
+        return of(headers, new SurroundingPublisher<>(null, publisher, trailers));
+    }
+
     /**
      * Returns a new {@link HttpRequestBuilder}.
      */

core/src/main/java/com/linecorp/armeria/common/HttpResponse.java

@@ -481,6 +481,23 @@ static HttpResponse of(ResponseHeaders headers, Publisher<? extends HttpObject>
         return PublisherBasedHttpResponse.from(headers, publisher);
     }
 
+    /**
+     * Creates a new HTTP response with the specified headers and trailers
+     * whose stream is produced from an existing {@link Publisher}.
+     *
+     * <p>Note that the {@link HttpData}s in the {@link Publisher} are not released when
+     * {@link Subscription#cancel()} or {@link #abort()} is called. You should add a hook in order to
+     * release the elements. See {@link PublisherBasedStreamMessage} for more information.
+     */
+    static HttpResponse of(ResponseHeaders headers,
+                           Publisher<? extends HttpData> publisher,
+                           HttpHeaders trailers) {
+        requireNonNull(headers, "headers");
+        requireNonNull(publisher, "publisher");
+        requireNonNull(trailers, "trailers");
+        return PublisherBasedHttpResponse.from(headers, publisher, trailers);
+    }
+
     /**
      * Creates a new HTTP response with the specified {@code content} that is converted into JSON using the
      * default {@link ObjectMapper}.

core/src/main/java/com/linecorp/armeria/common/HttpResponseBuilder.java

@@ -28,7 +28,6 @@
 import com.google.errorprone.annotations.FormatString;
 
 import com.linecorp.armeria.common.annotation.UnstableApi;
-import com.linecorp.armeria.common.stream.StreamMessage;
 
 /**
  * Builds a new {@link HttpResponse}.
@@ -299,8 +298,7 @@ public HttpResponse build() {
             if (trailers == null) {
                 return HttpResponse.of(responseHeaders, publisher);
             } else {
-                return HttpResponse.of(responseHeaders,
-                                       StreamMessage.concat(publisher, StreamMessage.of(trailers.build())));
+                return HttpResponse.of(responseHeaders, publisher, trailers.build());
             }
         }
     }

core/src/main/java/com/linecorp/armeria/common/PublisherBasedHttpResponse.java

@@ -21,12 +21,18 @@
 import org.reactivestreams.Publisher;
 
 import com.linecorp.armeria.common.stream.PublisherBasedStreamMessage;
-import com.linecorp.armeria.internal.common.stream.PrependingPublisher;
+import com.linecorp.armeria.internal.common.stream.SurroundingPublisher;
 
 final class PublisherBasedHttpResponse extends PublisherBasedStreamMessage<HttpObject> implements HttpResponse {
 
     static PublisherBasedHttpResponse from(ResponseHeaders headers, Publisher<? extends HttpObject> publisher) {
-        return new PublisherBasedHttpResponse(new PrependingPublisher<>(headers, publisher));
+        return new PublisherBasedHttpResponse(new SurroundingPublisher<>(headers, publisher, null));
+    }
+
+    static PublisherBasedHttpResponse from(ResponseHeaders headers,
+                                           Publisher<? extends HttpData> publisher,
+                                           HttpHeaders trailers) {
+        return new PublisherBasedHttpResponse(new SurroundingPublisher<>(headers, publisher, trailers));
     }
 
     PublisherBasedHttpResponse(Publisher<? extends HttpObject> publisher) {