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

[injae-kim]

Related Issue #3959

Motivation:

We have PrependingPublisher to publish ResponseHeaders and body in a StreamMessage.
Similarly, we can add AppendingPublisher to publish body and HTTP trailers more efficiently.

• On Add HttpResponseBuilder to fluently build HttpResponse #3941, we added PrependingPublisher to publish ResponseHeaders and body in a StreamMessage
• So it's nice to add SurroundingPublisher to publish body and HTTP trailers too!

Modifications:

• Add SurroundingPublisher to append trailers
• Add HttpResponse.of(header, pub, trailers) that using SurroundingPublisher

Result:

• You can now publish body and HTTP trailers easily by using SurroundingPublisher
• Close Add AppendingPublisher #3959 issue

[minwoox]

Basically, looks good! 😄
Left some suggestions.

[injae-kim]

Sorry for the late, I'll update SurroundingPublisher within this weekend! (I'm on-call this week..)

📌 Sorry once again for my late 😭 I'll must update within this weekend, until 04/02!!

[injae-kim]

@minwoox , can I get review? 🙇

[ikhoon]

Let me review this first. Minwoo is working on another issue.

[injae-kim]

Hmm I'm chekcing failed test on my local env~ I'll find root cause and fix it soon 🙇

[codecov]

Codecov Report

Patch coverage: 77.19% and project coverage change: +0.02% 🎉

Comparison is base (1ce4c69) 74.30% compared to head (732744b) 74.32%.

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #4727      +/-   ##
============================================
+ Coverage     74.30%   74.32%   +0.02%     
- Complexity    19600    19625      +25     
============================================
  Files          1682     1683       +1     
  Lines         72260    72585     +325     
  Branches       9242     9304      +62     
============================================
+ Hits          53695    53952     +257     
- Misses        14222    14265      +43     
- Partials       4343     4368      +25     

Files Changed	Coverage Δ
.../java/com/linecorp/armeria/common/HttpRequest.java	74.45% <0.00%> (-3.42%)	⬇️
...pring/web/reactive/AbstractServerHttpResponse.java	73.61% <ø> (ø)
...orp/armeria/common/PublisherBasedHttpResponse.java	71.42% <50.00%> (-28.58%)	⬇️
...meria/spring/web/reactive/ChannelSendOperator.java	65.60% <65.60%> (ø)
...spring/web/reactive/ArmeriaServerHttpResponse.java	84.14% <84.61%> (-0.14%)	⬇️
...a/internal/common/stream/SurroundingPublisher.java	87.38% <87.38%> (ø)
...java/com/linecorp/armeria/common/HttpResponse.java	88.55% <100.00%> (+0.19%)	⬆️
...m/linecorp/armeria/common/HttpResponseBuilder.java	76.08% <100.00%> (+5.87%)	⬆️

... and 23 files with indirect coverage changes

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[injae-kim]

class ChannelSendOperator {

		@Override
		public void subscribe(Subscriber<? super T> writeSubscriber) {
			synchronized (this) {
				Assert.state(this.writeSubscriber == null, "Only one write subscriber supported");
				this.writeSubscriber = writeSubscriber;
				if (this.error != null || this.completed) { // completed: true
					this.writeSubscriber.onSubscribe(Operators.emptySubscription());
					emitCachedSignals(); // 👈👈 
				}

		private boolean emitCachedSignals() {
			T item = this.item;
			this.item = null;
			if (item != null) {
				requiredWriteSubscriber().onNext(item); // 👈👈 although there's no request, onNext cached item!
			}
}

@ikhoon , I found that spring's ChannelSendOperator calls downstream.onNext(item) without any request on below case!

1. ChannelSendOperator.subscribe()
2. ChannelSendOperator is already completed
3. emitCachedSignals()
4. downstream.onNext(item) // although there's no request, onNext() is invoked!

"ChannelSendOperator(upstream) can downstream.onNext(item) without any request" -> it sounds weird, but we can't change spring ChannelSendOperator's behavior so I just removed assert upstreamRequested > 0; on L336 ~! PTAL 🙇

[jrhee17]

I see, changes still look good to me 👍

[ikhoon]

"ChannelSendOperator(upstream) can downstream.onNext(item) without any request"

It might be not related to this PR but the assertion found the bug. Let me add some logs and debug to know why ChannelSendOperator was completed before subscribe()

[injae-kim]

@ikhoon nim gentle ping 🙇

Can you check debug logs on broken test? I think I still can't check debug logs on CI/CD result.
I want to merge this PR and handle next PRs!

[ikhoon]

You need to download artifacts to see debug logs on the summary page of the GitHub Actions job.
Otherwise, let me try to merge #5104 quickly to easily see the test logs.

[ikhoon]

@injae-kim You can now check out the test results on Gradle build scans.
https://github.com/line/armeria/actions/runs/5901062192
https://ge.armeria.dev/s/r2qfke4r6ueky/tests/task/:spring:boot2-webflux-autoconfigure:shadedTest/details/com.linecorp.armeria.spring.web.reactive.MatrixVariablesTest/foo()?top-execution=1

[injae-kim]

test results on Gradle build scans.

oh it's cool! thanks a lot 🙇 I'll check test result and share to you~!!

[ikhoon]

I suspected the problem is a bug in ChannelSendOperator that violates the most important rule 1.1. https://github.com/reactive-streams/reactive-streams-jvm/blob/v1.0.4/README.md#1.1
So I forked ChannelSendOperator and modified not to publish the cached item before a request is made. https://github.com/line/armeria/pull/4727/files#r1299525470

[ikhoon]

The problem occurs only in Spring Boot 2 WebFlux. The following command can reproduce it locally.

./gradlew :spring:boot3-webflux-autoconfigure:testClasses :spring:boot2-webflux-autoconfigure:test --tests "*.MatrixVariablesTest"

ChannelSendOperator implementation has not changed between Spring Boot 2 and Spring Boot 3. The main difference was spring-projects/spring-framework#28398 which avoids collecting Flux before passing to ChannelSendOperator. Consequently, WriteBarrier.onComplete() was not invoked before WriteBarrier.subscribe() is called and backpressure looks respected.

[injae-kim]

Really thanks for your detailed investigation 🙇 If there's something I have to do, please tell me!

[github-actions]

🔍 Build Scan® (commit: 732744b)

Job name	Status	Build Scan®
build-windows-latest-jdk-19	✅	https://ge.armeria.dev/s/teteqdifibwo2
build-self-hosted-jdk-8	✅	https://ge.armeria.dev/s/6k6dbjnmnrq2u
build-self-hosted-jdk-19-snapshot-blockhound	✅	https://ge.armeria.dev/s/kfkdtqqdydcjs
build-self-hosted-jdk-17-min-java-17-coverage	✅	https://ge.armeria.dev/s/adsvztmpwiny2
build-self-hosted-jdk-17-min-java-11	✅	https://ge.armeria.dev/s/lokt3gtkyy22u
build-self-hosted-jdk-17-leak	✅	https://ge.armeria.dev/s/7zdolb7jhihng
build-self-hosted-jdk-11	✅	https://ge.armeria.dev/s/bnc2ya4njrfhg
build-macos-12-jdk-19	✅	https://ge.armeria.dev/s/whav3m3twjv2u

[ikhoon]

The actual diff with the upstream code.

--- spring-framework/spring-web/src/main/java/org/springframework/http/server/reactive/ChannelSendOperator.java	2023-01-15 20:37:55
+++ armeria/spring/boot3-webflux-autoconfigure/src/main/java/com/linecorp/armeria/spring/web/reactive/ChannelSendOperator.java	2023-08-21 11:37:11
@@ -1,4 +1,19 @@
 /*
+ * Copyright 2023 LINE Corporation
+ *
+ * LINE Corporation licenses this file to you 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:
+ *
+ *   https://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.
+ */
+/*
  * Copyright 2002-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -14,13 +29,18 @@
  * limitations under the License.
  */
 
-package org.springframework.http.server.reactive;
+package com.linecorp.armeria.spring.web.reactive;
 
 import java.util.function.Function;
 
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferUtils;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
 import reactor.core.CoreSubscriber;
 import reactor.core.Scannable;
 import reactor.core.publisher.Flux;
@@ -28,11 +48,6 @@
 import reactor.core.publisher.Operators;
 import reactor.util.context.Context;
 
-import org.springframework.core.io.buffer.DataBuffer;
-import org.springframework.core.io.buffer.DataBufferUtils;
-import org.springframework.lang.Nullable;
-import org.springframework.util.Assert;
-
 /**
  * Given a write function that accepts a source {@code Publisher<T>} to write
  * with and returns {@code Publisher<Void>} for the result, this operator helps
@@ -48,6 +63,8 @@
  */
 public class ChannelSendOperator<T> extends Mono<Void> implements Scannable {
 
+	// Forked from https://github.com/spring-projects/spring-framework/blob/1e3099759e2d823b6dd1c0c43895abcbe3e02a12/spring-web/src/main/java/org/springframework/http/server/reactive/ChannelSendOperator.java
+	// and modified at L370 for not publishing the cache item before receiving request(n) from the subscriber.
 	private final Function<Publisher<T>, Publisher<Void>> writeFunction;
 
 	private final Flux<T> source;
@@ -350,7 +367,7 @@
 			synchronized (this) {
 				Assert.state(this.writeSubscriber == null, "Only one write subscriber supported");
 				this.writeSubscriber = writeSubscriber;
-				if (this.error != null || this.completed) {
+				if (this.error != null || (this.completed && this.item == null)) {
 					this.writeSubscriber.onSubscribe(Operators.emptySubscription());
 					emitCachedSignals();
 				}