Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Adapt API to handle Vert.x 4.4.5 deprecations #35973

Merged
merged 1 commit into from
Sep 18, 2023
Merged

Adapt API to handle Vert.x 4.4.5 deprecations #35973

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
14 changes: 7 additions & 7 deletions ...n-lambda/event-server/src/main/java/io/quarkus/amazon/lambda/runtime/MockEventServer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
Expand All @@ -18,8 +19,6 @@
import org.jboss.logging.Logger;

import io.netty.handler.codec.http.HttpHeaderNames;
import io.vertx.core.Handler;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;
import io.vertx.core.VertxOptions;
import io.vertx.core.buffer.Buffer;
Expand Down Expand Up @@ -132,9 +131,9 @@ public void postEvent(RoutingContext ctx) {
}

public void nextEvent(RoutingContext ctx) {
vertx.executeBlocking(new Handler<>() {
vertx.executeBlocking(new Callable<Void>() {
@Override
public void handle(Promise<Object> event) {
public Void call() {
final AtomicBoolean closed = new AtomicBoolean(false);
ctx.response().closeHandler((v) -> closed.set(true));
ctx.response().exceptionHandler((v) -> closed.set(true));
Expand All @@ -149,12 +148,12 @@ public void handle(Promise<Object> event) {
log.debugf("Polled message %s but connection was closed, returning to queue",
request.get(AmazonLambdaApi.LAMBDA_RUNTIME_AWS_REQUEST_ID));
queue.put(request);
return;
return null;
} else {
break;
}
} else if (closed.get()) {
return;
return null;
}
}
} catch (InterruptedException e) {
Expand All @@ -180,8 +179,9 @@ public void handle(Promise<Object> event) {
} else {
ctx.response().setStatusCode(200).end();
}
return null;
}
}, false, null);
}, false);
}

protected String getEventContentType(RoutingContext request) {
Expand Down
9 changes: 5 additions & 4 deletions extensions/grpc/runtime/src/main/java/io/quarkus/grpc/runtime/GrpcServerRecorder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -495,9 +495,9 @@ private Map.Entry<Integer, Server> buildServer(Vertx vertx, GrpcServerConfigurat
applyNettySettings(configuration, vsBuilder);
if (launchMode == LaunchMode.DEVELOPMENT) {
vsBuilder.commandDecorator(command -> vertx.executeBlocking(
event -> event.complete(GrpcHotReplacementInterceptor.fire()),
false,
(Handler<AsyncResult<Boolean>>) result -> devModeWrapper.run(command)));
GrpcHotReplacementInterceptor::fire,
false)
.onComplete(result -> devModeWrapper.run(command)));
}
builder = vsBuilder;
}
Expand Down Expand Up @@ -650,7 +650,7 @@ public void start(Promise<Void> startPromise) {
});
} else {
// XDS server blocks on initialStartFuture
vertx.executeBlocking((Handler<Promise<Void>>) event -> {
vertx.executeBlocking(() -> {
try {
grpcServer.start();
int actualPort = grpcServer.getPort();
Expand All @@ -663,6 +663,7 @@ public void start(Promise<Void> startPromise) {
LOGGER.error("Unable to start gRPC server", e);
startPromise.fail(e);
}
return null;
});
}
}
Expand Down
12 changes: 4 additions & 8 deletions ...ime/src/main/java/io/quarkus/grpc/runtime/supports/blocking/BlockingExecutionHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,15 +1,14 @@
package io.quarkus.grpc.runtime.supports.blocking;

import java.util.concurrent.Callable;
import java.util.function.Consumer;

import io.grpc.Context;
import io.grpc.ServerCall;
import io.quarkus.arc.InjectableContext;
import io.quarkus.arc.ManagedContext;
import io.vertx.core.Handler;
import io.vertx.core.Promise;

class BlockingExecutionHandler<ReqT> implements Handler<Promise<Object>> {
class BlockingExecutionHandler<ReqT> implements Callable<Void> {
private final ServerCall.Listener<ReqT> delegate;
private final Context grpcContext;
private final Consumer<ServerCall.Listener<ReqT>> consumer;
Expand All @@ -30,7 +29,7 @@ public BlockingExecutionHandler(Consumer<ServerCall.Listener<ReqT>> consumer, Co
}

@Override
public void handle(Promise<Object> event) {
public Void call() {
/*
* We lock here because with client side streaming different messages from the same request
* might be served by different worker threads. This guarantees memory consistency.
Expand All @@ -42,13 +41,10 @@ public void handle(Promise<Object> event) {
requestContext.activate(state);
try {
consumer.accept(delegate);
} catch (Throwable any) {
event.fail(any);
return;
} finally {
requestContext.deactivate();
}
event.complete();
return null;
} finally {
grpcContext.detach(previous);
}
Expand Down
36 changes: 19 additions & 17 deletions ...me/src/main/java/io/quarkus/grpc/runtime/supports/blocking/BlockingServerInterceptor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
Expand All @@ -21,9 +22,6 @@
import io.quarkus.arc.InjectableContext;
import io.quarkus.arc.InjectableContext.ContextState;
import io.quarkus.arc.ManagedContext;
import io.vertx.core.AsyncResult;
import io.vertx.core.Handler;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;

/**
Expand Down Expand Up @@ -115,17 +113,17 @@ public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, Re
// that should always be called before this interceptor
ContextState state = requestContext.getState();
ReplayListener<ReqT> replay = new ReplayListener<>(state);
vertx.executeBlocking(f -> {
vertx.executeBlocking(() -> {
ServerCall.Listener<ReqT> listener;
try {
requestContext.activate(state);
listener = next.startCall(call, headers);
} finally {
requestContext.deactivate();
}
f.complete(listener);
}, false,
(Handler<AsyncResult<ServerCall.Listener<ReqT>>>) event -> replay.setDelegate(event.result()));
return listener;
}, false)
.onComplete(event -> replay.setDelegate(event.result()));

return replay;
} else {
Expand Down Expand Up @@ -185,14 +183,14 @@ private void scheduleOrEnqueue(Consumer<ServerCall.Listener<ReqT>> consumer) {
*/
private void executeBlockingWithRequestContext(Consumer<ServerCall.Listener<ReqT>> consumer) {
final Context grpcContext = Context.current();
Handler<Promise<Object>> blockingHandler = new BlockingExecutionHandler<>(consumer, grpcContext, delegate,
Callable<Void> blockingHandler = new BlockingExecutionHandler<>(consumer, grpcContext, delegate,
requestContextState, getRequestContext(), this);
if (devMode) {
blockingHandler = new DevModeBlockingExecutionHandler(Thread.currentThread().getContextClassLoader(),
blockingHandler);
}
this.isConsumingFromIncomingEvents = true;
vertx.executeBlocking(blockingHandler, true, p -> {
vertx.executeBlocking(blockingHandler, true).onComplete(p -> {
Consumer<ServerCall.Listener<ReqT>> next = incomingEvents.poll();
if (next != null) {
executeBlockingWithRequestContext(next);
Expand Down Expand Up @@ -275,21 +273,25 @@ private void scheduleOrEnqueue(Consumer<ServerCall.Listener<ReqT>> consumer) {

private void executeVirtualWithRequestContext(Consumer<ServerCall.Listener<ReqT>> consumer) {
final Context grpcContext = Context.current();
Handler<Promise<Object>> blockingHandler = new BlockingExecutionHandler<>(consumer, grpcContext, delegate,
Callable<Void> blockingHandler = new BlockingExecutionHandler<>(consumer, grpcContext, delegate,
requestContextState, getRequestContext(), this);
if (devMode) {
blockingHandler = new DevModeBlockingExecutionHandler(Thread.currentThread().getContextClassLoader(),
blockingHandler);
}
this.isConsumingFromIncomingEvents = true;
Handler<Promise<Object>> finalBlockingHandler = blockingHandler;
var finalBlockingHandler = blockingHandler;
virtualThreadExecutor.execute(() -> {
finalBlockingHandler.handle(Promise.promise());
Consumer<ServerCall.Listener<ReqT>> next = incomingEvents.poll();
if (next != null) {
executeVirtualWithRequestContext(next);
} else {
this.isConsumingFromIncomingEvents = false;
try {
finalBlockingHandler.call();
Consumer<ServerCall.Listener<ReqT>> next = incomingEvents.poll();
if (next != null) {
executeVirtualWithRequestContext(next);
} else {
this.isConsumingFromIncomingEvents = false;
}
} catch (Exception e) {
throw new RuntimeException(e);
}
});
}
Expand Down
13 changes: 6 additions & 7 deletions .../main/java/io/quarkus/grpc/runtime/supports/blocking/DevModeBlockingExecutionHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,24 +1,23 @@
package io.quarkus.grpc.runtime.supports.blocking;

import io.vertx.core.Handler;
import io.vertx.core.Promise;
import java.util.concurrent.Callable;

class DevModeBlockingExecutionHandler implements Handler<Promise<Object>> {
class DevModeBlockingExecutionHandler implements Callable<Void> {

final ClassLoader tccl;
final Handler<Promise<Object>> delegate;
final Callable<Void> delegate;

public DevModeBlockingExecutionHandler(ClassLoader tccl, Handler<Promise<Object>> delegate) {
public DevModeBlockingExecutionHandler(ClassLoader tccl, Callable<Void> delegate) {
this.tccl = tccl;
this.delegate = delegate;
}

@Override
public void handle(Promise<Object> event) {
public Void call() throws Exception {
ClassLoader originalTccl = Thread.currentThread().getContextClassLoader();
Thread.currentThread().setContextClassLoader(tccl);
try {
delegate.handle(event);
return delegate.call();
} finally {
Thread.currentThread().setContextClassLoader(originalTccl);
}
Expand Down
13 changes: 4 additions & 9 deletions extensions/quartz/runtime/src/main/java/io/quarkus/quartz/runtime/QuartzSchedulerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
import java.util.OptionalLong;
import java.util.Properties;
import java.util.TimeZone;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ConcurrentHashMap;
Expand Down Expand Up @@ -92,7 +93,6 @@
import io.smallrye.common.vertx.VertxContext;
import io.vertx.core.Context;
import io.vertx.core.Handler;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;

/**
Expand Down Expand Up @@ -894,15 +894,10 @@ public void run() {
}
});
} else {
context.executeBlocking(new Handler<Promise<Object>>() {
context.executeBlocking(new Callable<Object>() {
@Override
public void handle(Promise<Object> p) {
try {
trigger.invoker.invoke(new QuartzScheduledExecution(trigger, jobExecutionContext));
p.complete();
} catch (Exception e) {
p.tryFail(e);
}
public Object call() throws Exception {
return trigger.invoker.invoke(new QuartzScheduledExecution(trigger, jobExecutionContext));
}
}, false);
}
Expand Down
13 changes: 6 additions & 7 deletions .../runtime/src/main/java/io/quarkus/reactive/datasource/runtime/ConnectOptionsSupplier.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,15 +5,14 @@

import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.IntUnaryOperator;
import java.util.function.Supplier;
import java.util.function.UnaryOperator;

import io.quarkus.credentials.CredentialsProvider;
import io.vertx.core.Future;
import io.vertx.core.Handler;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;
import io.vertx.sqlclient.SqlConnectOptions;

Expand All @@ -24,7 +23,7 @@ public class ConnectOptionsSupplier<CO extends SqlConnectOptions> implements Sup
private final String credentialsProviderName;
private final List<CO> connectOptionsList;
private final UnaryOperator<CO> connectOptionsCopy;
private final Handler<Promise<CO>> blockingCodeHandler;
private final Callable<CO> blockingCodeHandler;

public ConnectOptionsSupplier(Vertx vertx, CredentialsProvider credentialsProvider, String credentialsProviderName,
List<CO> connectOptionsList, UnaryOperator<CO> connectOptionsCopy) {
Expand All @@ -33,20 +32,20 @@ public ConnectOptionsSupplier(Vertx vertx, CredentialsProvider credentialsProvid
this.credentialsProviderName = credentialsProviderName;
this.connectOptionsList = connectOptionsList;
this.connectOptionsCopy = connectOptionsCopy;
blockingCodeHandler = new BlockingCodeHandler();
this.blockingCodeHandler = new BlockingCodeHandler();
}

@Override
public Future<CO> get() {
return vertx.executeBlocking(blockingCodeHandler, false);
}

private class BlockingCodeHandler implements Handler<Promise<CO>>, IntUnaryOperator {
private class BlockingCodeHandler implements Callable<CO>, IntUnaryOperator {

final AtomicInteger idx = new AtomicInteger();

@Override
public void handle(Promise<CO> promise) {
public CO call() {
Map<String, String> credentials = credentialsProvider.getCredentials(credentialsProviderName);
String user = credentials.get(USER_PROPERTY_NAME);
String password = credentials.get(PASSWORD_PROPERTY_NAME);
Expand All @@ -56,7 +55,7 @@ public void handle(Promise<CO> promise) {
CO connectOptions = connectOptionsCopy.apply(connectOptionsList.get(nextIdx));
connectOptions.setUser(user).setPassword(password);

promise.complete(connectOptions);
return connectOptions;
}

@Override
Expand Down
6 changes: 3 additions & 3 deletions ...steasy/runtime/src/main/java/io/quarkus/resteasy/runtime/standalone/VertxHttpRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -262,16 +262,16 @@ public CompletionStage<Void> executeBlockingIo(RunnableWithException f, boolean
suspend();
}
CompletableFuture<Void> ret = new CompletableFuture<>();
this.request.context.executeBlocking(future -> {
this.request.context.executeBlocking(() -> {
try (CloseableContext newContext = ResteasyContext.addCloseableContextDataLevel(context)) {
f.run();
future.complete();
} catch (RuntimeException e) {
throw e;
} catch (Exception e) {
throw new RuntimeException(e);
}
}, res -> {
return null;
}).onComplete(res -> {
if (res.succeeded())
ret.complete(null);
else
Expand Down
4 changes: 2 additions & 2 deletions .../main/java/io/quarkus/resteasy/reactive/server/servlet/runtime/ServletRequestContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -236,7 +236,7 @@ public String getRequestScheme() {

@Override
public String getRequestHost() {
return context.request().host();
return context.request().authority().toString();
}

@Override
Expand All @@ -246,7 +246,7 @@ public void closeConnection() {
} catch (IOException e) {
//ignore
}
context.response().close();
context.request().connection().close();
}

@Override
Expand Down
4 changes: 2 additions & 2 deletions ...ve/deployment/src/test/java/io/quarkus/rest/client/reactive/RequestLeakDetectionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -269,9 +269,9 @@ private Task(Context context, Runnable runnable, boolean blocking) {

void run() {
if (blocking) {
context.executeBlocking(p -> {
context.executeBlocking(() -> {
runnable.run();
p.complete();
return null;
});
} else {
context.runOnContext(x -> runnable.run());
Expand Down