-
Notifications
You must be signed in to change notification settings - Fork 2.7k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
CDI context propagation improvements for the reactive stack
== ArC Introduce the VertxCurrentContextFactory so that the Vert.x duplicated context can be used to store the "current context" for normal scopes. == resteasy-reactive Don't clear our CDI current request when suspending. This used to be done in order to prevent subsequent requests running on the same thread as a suspended request from accessing the former's data. With the advent of DuplicatedContext backed storage, there is no longer any chance of mixing data so there is no need to clear it out. Furthermore, by not clearing out current request, code that accesses the request scoped CurrentVertxRequest that is executed while the request is suspended, can now work even if context propagation is not in play. == Tests Add leak detection tests for resteasy reactive, graphql and reactive rest client. Also improve the OpenTelemetry reactive tests. == OpenTelemetry Only register the OpenTelemetryClientFilter for RESTEasy Client Classic. Use Capabilities to determine when to register the OpenTelemetryClientFilter. Co-authored-by: Georgios Andrianakis <[email protected]> Co-authored-by: Clement Escoffier <[email protected]> Co-authored-by: Roberto Cortez <[email protected]> Co-authored-by: brunobat <[email protected]> Co-authored-by: Martin Kouba <[email protected]>
- Loading branch information
6 people
committed
Sep 7, 2022
1 parent
96183c6
commit 00ce709
Showing
19 changed files
with
939 additions
and
41 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
192 changes: 192 additions & 0 deletions
192
...ment/src/test/java/io/quarkus/resteasy/reactive/server/test/RequestLeakDetectionTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,192 @@ | ||
| package io.quarkus.resteasy.reactive.server.test; | ||
|
|
||
| import static org.awaitility.Awaitility.await; | ||
|
|
||
| import java.util.ArrayList; | ||
| import java.util.HashSet; | ||
| import java.util.List; | ||
| import java.util.Set; | ||
| import java.util.concurrent.CopyOnWriteArrayList; | ||
| import java.util.concurrent.atomic.AtomicInteger; | ||
|
|
||
| import javax.enterprise.context.ApplicationScoped; | ||
| import javax.enterprise.context.RequestScoped; | ||
| import javax.inject.Inject; | ||
| import javax.ws.rs.GET; | ||
| import javax.ws.rs.Path; | ||
| import javax.ws.rs.core.MediaType; | ||
|
|
||
| import org.jboss.shrinkwrap.api.asset.EmptyAsset; | ||
| import org.junit.jupiter.api.Assertions; | ||
| import org.junit.jupiter.api.Test; | ||
| import org.junit.jupiter.api.extension.RegisterExtension; | ||
|
|
||
| import io.quarkus.test.QuarkusUnitTest; | ||
| import io.restassured.RestAssured; | ||
| import io.restassured.response.ResponseBody; | ||
| import io.smallrye.common.vertx.VertxContext; | ||
| import io.smallrye.mutiny.Uni; | ||
| import io.vertx.core.Context; | ||
| import io.vertx.core.Vertx; | ||
|
|
||
| public class RequestLeakDetectionTest { | ||
|
|
||
| @RegisterExtension | ||
| static QuarkusUnitTest test = new QuarkusUnitTest() | ||
| .withApplicationRoot((jar) -> jar.addClasses(MyRestAPI.class, MyRequestScopeBean.class, Barrier.class, Task.class) | ||
| .addAsManifestResource(EmptyAsset.INSTANCE, "beans.xml")); | ||
|
|
||
| @Inject | ||
| Barrier barrier; | ||
|
|
||
| @Test | ||
| public void testWithConcurrentCalls() { | ||
| List<String> results = new CopyOnWriteArrayList<>(); | ||
| int count = 100; | ||
| barrier.setMaxConcurrency(count); | ||
| for (int i = 0; i < count; i++) { | ||
| int c = i; | ||
| new Thread(() -> { | ||
| ResponseBody<?> body = RestAssured.given().pathParam("val", c).contentType(MediaType.APPLICATION_JSON) | ||
| .get("/test/{val}").thenReturn().body(); | ||
| String value = body.toString(); | ||
| results.add(value); | ||
| }).start(); | ||
| } | ||
| await().until(() -> results.size() == count); | ||
| Set<String> asSet = new HashSet<>(results); | ||
| Assertions.assertEquals(asSet.size(), count); | ||
| } | ||
|
|
||
| @Test | ||
| public void testWithConcurrentBlockingCalls() { | ||
| List<String> results = new CopyOnWriteArrayList<>(); | ||
| int count = 100; | ||
| barrier.setMaxConcurrency(count); | ||
| for (int i = 0; i < count; i++) { | ||
| int c = i; | ||
| new Thread(() -> { | ||
| ResponseBody<?> body = RestAssured.given().pathParam("val", c).contentType(MediaType.APPLICATION_JSON) | ||
| .get("/test/blocking/{val}").thenReturn().body(); | ||
| String value = body.toString(); | ||
| results.add(value); | ||
| }).start(); | ||
| } | ||
| await().until(() -> results.size() == count); | ||
| Set<String> asSet = new HashSet<>(results); | ||
| Assertions.assertEquals(asSet.size(), count); | ||
| } | ||
|
|
||
| @ApplicationScoped | ||
| @Path("/test") | ||
| public static class MyRestAPI { | ||
|
|
||
| @Inject | ||
| MyRequestScopeBean bean; | ||
|
|
||
| @Inject | ||
| Barrier barrier; | ||
|
|
||
| @GET | ||
| @Path("/{val}") | ||
| public Uni<Foo> foo(int val) { | ||
| Assertions.assertTrue(VertxContext.isOnDuplicatedContext()); | ||
| Vertx.currentContext().putLocal("count", val); | ||
| bean.setValue(val); | ||
|
|
||
| return Uni.createFrom().<Integer> emitter(e -> { | ||
| barrier.enqueue(Vertx.currentContext(), () -> { | ||
| Assertions.assertTrue(VertxContext.isOnDuplicatedContext()); | ||
| int r = Vertx.currentContext().getLocal("count"); | ||
| Assertions.assertEquals(r, val); | ||
| e.complete(bean.getValue()); | ||
| }); | ||
| }).map(i -> new Foo(Integer.toString(i))); | ||
| } | ||
|
|
||
| @GET | ||
| @Path("/blocking/{val}") | ||
| public Foo blocking(int val) { | ||
| Assertions.assertTrue(VertxContext.isOnDuplicatedContext()); | ||
| Vertx.currentContext().putLocal("count", val); | ||
| bean.setValue(val); | ||
|
|
||
| return Uni.createFrom().<Integer> emitter(e -> { | ||
| barrier.enqueue(Vertx.currentContext(), () -> { | ||
| Assertions.assertTrue(VertxContext.isOnDuplicatedContext()); | ||
| int r = Vertx.currentContext().getLocal("count"); | ||
| Assertions.assertEquals(r, val); | ||
| e.complete(bean.getValue()); | ||
| }); | ||
| }) | ||
| .map(i -> new Foo(Integer.toString(i))) | ||
| .await().indefinitely(); | ||
| } | ||
| } | ||
|
|
||
| @ApplicationScoped | ||
| public static class Barrier { | ||
|
|
||
| private int level; | ||
|
|
||
| public void setMaxConcurrency(int level) { | ||
| this.level = level; | ||
| } | ||
|
|
||
| private final AtomicInteger counter = new AtomicInteger(); | ||
| private final List<Task> tasks = new CopyOnWriteArrayList<>(); | ||
|
|
||
| public void enqueue(Context context, Runnable runnable) { | ||
| Task task = new Task(context, runnable); | ||
| tasks.add(task); | ||
| if (counter.incrementAndGet() >= level) { | ||
| for (Task tbr : new ArrayList<>(tasks)) { | ||
| tbr.run(); | ||
| tasks.remove(tbr); | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private static class Task { | ||
| private final Context context; | ||
| private final Runnable runnable; | ||
|
|
||
| private Task(Context context, Runnable runnable) { | ||
| this.context = context; | ||
| this.runnable = runnable; | ||
| } | ||
|
|
||
| void run() { | ||
| context.runOnContext(x -> runnable.run()); | ||
| } | ||
| } | ||
|
|
||
| @RequestScoped | ||
| public static class MyRequestScopeBean { | ||
|
|
||
| private int value = -1; | ||
|
|
||
| public void setValue(int v) { | ||
| if (value != -1) { | ||
| throw new IllegalStateException("Already initialized"); | ||
| } | ||
| value = v; | ||
| } | ||
|
|
||
| public int getValue() { | ||
| return value; | ||
| } | ||
|
|
||
| } | ||
|
|
||
| public static class Foo { | ||
|
|
||
| public final String value; | ||
|
|
||
| public Foo(String value) { | ||
| this.value = value; | ||
| } | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.