request = context.getBean(HttpClient.class).get()
+ .uri(path).response();
+
+ TestHttpCallbackSubscriber.subscribe(request, r -> r.status().code(), callback);
+ }
+
+}
diff --git a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TestHttpCallbackSubscriber.java b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TestHttpCallbackSubscriber.java
new file mode 100644
index 000000000..b56557f03
--- /dev/null
+++ b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TestHttpCallbackSubscriber.java
@@ -0,0 +1,108 @@
+/*
+ * Copyright 2013-2020 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
+ *
+ * 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.
+ */
+
+package org.springframework.cloud.sleuth.instrument.web.client;
+
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
+
+import org.reactivestreams.Subscription;
+import reactor.core.CoreSubscriber;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.Operators;
+import reactor.util.context.Context;
+import zipkin2.Callback;
+
+/**
+ * {@link #subscribe} is made for reactor-netty and WebFlux client requests used in tests.
+ * This does more assertions than normal, to ensure instrumentation isn't redundantly
+ * signalling, or missing signals.
+ *
+ *
+ * The implementation forwards signals to the supplied {@link Callback}, enforcing
+ * assumptions about a non-empty, {@link Mono} subscription.
+ */
+final class TestHttpCallbackSubscriber implements CoreSubscriber {
+
+ static void subscribe(Mono mono, Function statusCodeFunction,
+ Callback callback) {
+ mono.subscribe(new TestHttpCallbackSubscriber<>(statusCodeFunction, callback));
+ }
+
+ final Function statusCodeFunction;
+
+ final Callback callback;
+
+ final AtomicReference ref = new AtomicReference<>();
+
+ private TestHttpCallbackSubscriber(Function statusCodeFunction,
+ Callback callback) {
+ this.statusCodeFunction = statusCodeFunction;
+ this.callback = callback;
+ }
+
+ @Override
+ public void onSubscribe(Subscription s) {
+ if (Operators.validate(ref.getAndSet(s), s)) {
+ s.request(Long.MAX_VALUE);
+ }
+ else {
+ // We don't intentionally call subscribe() multiple times in our tests. If we
+ // reach here, possibly instrumentation is redundantly subscribing.
+ callback.onError(new AssertionError("onSubscribe() called twice!"));
+ }
+ }
+
+ @Override
+ public void onNext(T t) {
+ if (ref.getAndSet(null) != null) {
+ callback.onSuccess(statusCodeFunction.apply(t));
+ }
+ else {
+ // This is a Mono, which doesn't signal onNext() twice. If we reach here,
+ // possibly instrumentation is signaling twice.
+ callback.onError(new AssertionError("onNext() called twice!"));
+ }
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ if (ref.getAndSet(null) != null) {
+ callback.onError(t);
+ }
+ else {
+ // We don't expect onError() to signal twice. If we reach here, possibly
+ // instrumentation is signaling twice or onSuccess() threw an exception.
+ callback.onError(new AssertionError("onError() called twice: " + t, t));
+ }
+ }
+
+ @Override
+ public void onComplete() {
+ if (ref.getAndSet(null) != null) {
+ // Tests make a non-empty Mono subscription, which should not signal
+ // onComplete() before onNext(). If we reach here, possibly instrumentation
+ // is not signaling onNext() when it should.
+ callback.onError(new AssertionError("onComplete() called before onNext!"));
+ }
+ }
+
+ @Override
+ public Context currentContext() {
+ return Context.empty();
+ }
+
+}
diff --git a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/WebClientBraveTests.java b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/WebClientBraveTests.java
new file mode 100644
index 000000000..d5859a3d5
--- /dev/null
+++ b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/WebClientBraveTests.java
@@ -0,0 +1,128 @@
+/*
+ * Copyright 2013-2019 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
+ *
+ * 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.
+ */
+
+package org.springframework.cloud.sleuth.instrument.web.client;
+
+import brave.http.HttpTracing;
+import brave.test.http.ITHttpAsyncClient;
+import org.junit.Ignore;
+import org.junit.Test;
+import reactor.core.publisher.Mono;
+import reactor.netty.http.client.HttpClient;
+import zipkin2.Callback;
+
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.http.client.reactive.ReactorClientHttpConnector;
+import org.springframework.web.reactive.function.BodyInserters;
+import org.springframework.web.reactive.function.client.ClientResponse;
+import org.springframework.web.reactive.function.client.WebClient;
+
+/**
+ * This runs Brave's integration tests without underlying instrumentation, which would
+ * happen when a 3rd party client like Jetty is in use.
+ */
+// Function of spring context so that shutdown hooks happen!
+public class WebClientBraveTests
+ extends ITHttpAsyncClient {
+
+ /**
+ * This uses Spring to instrument the {@link WebClient} using a
+ * {@link BeanPostProcessor}.
+ */
+ @Override
+ protected AnnotationConfigApplicationContext newClient(int port) {
+ AnnotationConfigApplicationContext result = new AnnotationConfigApplicationContext();
+ result.registerBean(HttpTracing.class, () -> httpTracing);
+ result.register(WebClientBuilderConfiguration.class);
+ result.register(TraceWebClientBeanPostProcessor.class);
+ result.refresh();
+ return result;
+ }
+
+ @Override
+ protected void closeClient(AnnotationConfigApplicationContext context) {
+ context.close(); // ensures shutdown hooks fire
+ }
+
+ @Override
+ protected void get(AnnotationConfigApplicationContext context,
+ String pathIncludingQuery) {
+ client(context).get().uri(pathIncludingQuery).exchange().block();
+ }
+
+ @Override
+ protected void post(AnnotationConfigApplicationContext context,
+ String pathIncludingQuery, String body) {
+ client(context).post().uri(pathIncludingQuery).body(BodyInserters.fromValue(body))
+ .exchange().block();
+ }
+
+ @Override
+ protected void getAsync(AnnotationConfigApplicationContext context, String path,
+ Callback callback) {
+ Mono request = client(context).get().uri(path).exchange();
+
+ TestHttpCallbackSubscriber.subscribe(request, ClientResponse::rawStatusCode,
+ callback);
+ }
+
+ @Test
+ @Ignore("TODO: reactor/reactor-netty#1000")
+ @Override
+ public void redirect() {
+ }
+
+ @Test
+ @Ignore("WebClient has no portable function to retrieve the server address")
+ @Override
+ public void reportsServerAddress() {
+ }
+
+ @Test
+ @Ignore("TODO: maybe refactor as an ExchangeFilterFunction to get the request from response")
+ @Override
+ public void readsRequestAtResponseTime() {
+ }
+
+ WebClient client(AnnotationConfigApplicationContext context) {
+ return context.getBean(WebClient.Builder.class)
+ .baseUrl("http://127.0.0.1:" + server.getPort()).build();
+ }
+
+ /**
+ * This fakes auto-configuration which wouldn't configure reactor's trace
+ * instrumentation.
+ */
+ @Configuration
+ static class WebClientBuilderConfiguration {
+
+ @Bean
+ HttpClient httpClient() {
+ return ReactorNettyHttpClientBraveTests.testHttpClient();
+ }
+
+ @Bean
+ WebClient.Builder webClientBuilder(HttpClient httpClient) {
+ return WebClient.builder()
+ .clientConnector(new ReactorClientHttpConnector(httpClient));
+ }
+
+ }
+
+}