Skip to content

Commit

Permalink
Merge pull request #35973 from cescoffier/vertx-4.4.5-deprecations
Browse files Browse the repository at this point in the history
Adapt API to handle Vert.x 4.4.5 deprecations
  • Loading branch information
cescoffier authored Sep 18, 2023
2 parents 32827cb + 97c358a commit 54df528
Show file tree
Hide file tree
Showing 26 changed files with 131 additions and 162 deletions.
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
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
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
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
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
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
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
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
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
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
Loading

0 comments on commit 54df528

Please sign in to comment.