Skip to content

Commit 777e35b

Browse files
phillip-krugerphillip-kruger
authored
Merge pull request #31743 from quarkusio/new-dev-ui-uni-and-blocking
Allows JSON-RPC method to return Uni<?> and add support for @Blocking and @nonblocking
2 parents 9d23daa + 46fd61e commit 777e35b

File tree

5 files changed

+156
-51
lines changed

5 files changed

+156
-51
lines changed

‎extensions/vertx-http/deployment/src/main/java/io/quarkus/devui/deployment/DevUIProcessor.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,8 @@
5959
import io.quarkus.vertx.http.deployment.RouteBuildItem;
6060
import io.quarkus.vertx.http.deployment.webjar.WebJarBuildItem;
6161
import io.quarkus.vertx.http.deployment.webjar.WebJarResultsBuildItem;
62+
import io.smallrye.common.annotation.Blocking;
63+
import io.smallrye.common.annotation.NonBlocking;
6264
import io.smallrye.mutiny.Multi;
6365
import io.vertx.core.Handler;
6466
import io.vertx.ext.web.RoutingContext;
@@ -272,9 +274,15 @@ void findAllJsonRPCMethods(BuildProducer<JsonRPCMethodsBuildItem> jsonRPCMethods
272274
params.put(parameterName, parameterClass);
273275
}
274276
JsonRpcMethod jsonRpcMethod = new JsonRpcMethod(clazz, method.name(), params);
277+
jsonRpcMethod.setExplicitlyBlocking(method.hasAnnotation(Blocking.class));
278+
jsonRpcMethod
279+
.setExplicitlyNonBlocking(method.hasAnnotation(NonBlocking.class));
275280
jsonRpcMethods.put(jsonRpcMethodName, jsonRpcMethod);
276281
} else {
277282
JsonRpcMethod jsonRpcMethod = new JsonRpcMethod(clazz, method.name(), null);
283+
jsonRpcMethod.setExplicitlyBlocking(method.hasAnnotation(Blocking.class));
284+
jsonRpcMethod
285+
.setExplicitlyNonBlocking(method.hasAnnotation(NonBlocking.class));
278286
jsonRpcMethods.put(jsonRpcMethodName, jsonRpcMethod);
279287
}
280288
}

‎extensions/vertx-http/runtime/src/main/java/io/quarkus/devui/runtime/comms/JsonRpcRouter.java‎

Lines changed: 93 additions & 49 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,6 @@
22

33
import static io.quarkus.devui.runtime.jsonrpc.JsonRpcKeys.MessageType;
44

5-
import java.lang.reflect.InvocationTargetException;
65
import java.lang.reflect.Method;
76
import java.util.ArrayList;
87
import java.util.HashMap;
@@ -11,14 +10,20 @@
1110
import java.util.concurrent.ConcurrentHashMap;
1211

1312
import jakarta.enterprise.context.ApplicationScoped;
13+
import jakarta.inject.Inject;
14+
15+
import org.jboss.logging.Logger;
1416

1517
import io.quarkus.arc.Arc;
1618
import io.quarkus.devui.runtime.jsonrpc.JsonRpcMethod;
1719
import io.quarkus.devui.runtime.jsonrpc.JsonRpcMethodName;
1820
import io.quarkus.devui.runtime.jsonrpc.JsonRpcReader;
1921
import io.quarkus.devui.runtime.jsonrpc.JsonRpcWriter;
2022
import io.smallrye.mutiny.Multi;
23+
import io.smallrye.mutiny.Uni;
24+
import io.smallrye.mutiny.infrastructure.Infrastructure;
2125
import io.smallrye.mutiny.subscription.Cancellable;
26+
import io.smallrye.mutiny.unchecked.Unchecked;
2227
import io.vertx.core.http.ServerWebSocket;
2328
import io.vertx.core.json.JsonObject;
2429

@@ -60,7 +65,7 @@ public void populateJsonRPCMethods(Map<String, Map<JsonRpcMethodName, JsonRpcMet
6065
javaMethod = providerInstance.getClass().getMethod(jsonRpcMethod.getMethodName());
6166
}
6267
ReflectionInfo reflectionInfo = new ReflectionInfo(jsonRpcMethod.getClazz(), providerInstance, javaMethod,
63-
params);
68+
params, jsonRpcMethod.getExplicitlyBlocking(), jsonRpcMethod.getExplicitlyNonBlocking());
6469
String jsonRpcMethodName = extensionName + DOT + methodName;
6570
jsonRpcToJava.put(jsonRpcMethodName, reflectionInfo);
6671
} catch (NoSuchMethodException | SecurityException ex) {
@@ -70,21 +75,40 @@ public void populateJsonRPCMethods(Map<String, Map<JsonRpcMethodName, JsonRpcMet
7075
}
7176
}
7277

78+
private Uni<?> invoke(ReflectionInfo info, Object target, Object[] args) {
79+
if (info.isReturningUni()) {
80+
try {
81+
Uni<?> uni = ((Uni<?>) info.method.invoke(target, args));
82+
if (info.isExplicitlyBlocking()) {
83+
return uni.runSubscriptionOn(Infrastructure.getDefaultExecutor());
84+
} else {
85+
return uni;
86+
}
87+
} catch (Exception e) {
88+
return Uni.createFrom().failure(e);
89+
}
90+
} else {
91+
Uni<?> uni = Uni.createFrom().item(Unchecked.supplier(() -> info.method.invoke(target, args)));
92+
if (!info.isExplicitlyNonBlocking()) {
93+
return uni.runSubscriptionOn(Infrastructure.getDefaultExecutor());
94+
} else {
95+
return uni;
96+
}
97+
}
98+
}
99+
73100
public void addSocket(ServerWebSocket socket) {
74101
socket.textMessageHandler((e) -> {
75-
socket.writeTextMessage(route(e, socket));
102+
JsonRpcReader jsonRpcRequest = JsonRpcReader.read(e);
103+
route(jsonRpcRequest, socket);
76104
});
77105
}
78106

79-
private String route(String message, ServerWebSocket s) {
80-
JsonRpcReader jsonRpcRequest = JsonRpcReader.read(message);
81-
JsonObject jsonRpcResponse = route(jsonRpcRequest, s);
82-
return jsonRpcResponse.encodePrettily();
83-
}
107+
@Inject
108+
Logger logger;
84109

85110
@SuppressWarnings("unchecked")
86-
private JsonObject route(JsonRpcReader jsonRpcRequest, ServerWebSocket s) {
87-
111+
private void route(JsonRpcReader jsonRpcRequest, ServerWebSocket s) {
88112
String jsonRpcMethodName = jsonRpcRequest.getMethod();
89113

90114
// First check some internal methods
@@ -95,52 +119,72 @@ private JsonObject route(JsonRpcReader jsonRpcRequest, ServerWebSocket s) {
95119
Cancellable cancellable = this.subscriptions.remove(jsonRpcRequest.getId());
96120
cancellable.cancel();
97121
}
98-
return jsonRpcResponse;
99-
122+
s.writeTextMessage(jsonRpcResponse.encode());
100123
} else if (this.jsonRpcToJava.containsKey(jsonRpcMethodName)) { // Route to extension
101124
ReflectionInfo reflectionInfo = this.jsonRpcToJava.get(jsonRpcMethodName);
102-
Object providerInstance = Arc.container().select(reflectionInfo.bean).get();
103-
try {
104-
Object result;
105-
if (jsonRpcRequest.hasParams()) {
106-
Object[] args = getArgsAsObjects(reflectionInfo.params, jsonRpcRequest);
107-
result = reflectionInfo.method.invoke(providerInstance, args);
108-
} else {
109-
result = reflectionInfo.method.invoke(providerInstance);
110-
}
125+
Object target = Arc.container().select(reflectionInfo.bean).get();
111126

112-
// Here wrap in our own object that contain some more metadata
113-
JsonObject jsonRpcResponse;
114-
if (reflectionInfo.isSubscription()) {
115-
// Subscription
116-
Multi<?> subscription = (Multi) result;
117-
118-
// TODO: If Jackson is on the classpath ?
119-
120-
Cancellable cancellable = subscription.subscribe().with((t) -> {
121-
JsonObject jsonResponse = JsonRpcWriter.writeResponse(jsonRpcRequest.getId(), t,
122-
MessageType.SubscriptionMessage);
123-
s.writeTextMessage(jsonResponse.encodePrettily());
124-
});
125-
126-
this.subscriptions.put(jsonRpcRequest.getId(), cancellable);
127-
128-
jsonRpcResponse = JsonRpcWriter.writeResponse(jsonRpcRequest.getId(), null, MessageType.Void);
129-
130-
} else {
131-
// Normal response
132-
jsonRpcResponse = JsonRpcWriter.writeResponse(jsonRpcRequest.getId(), result, MessageType.Response);
127+
if (reflectionInfo.isReturningMulti()) {
128+
Multi<?> multi;
129+
try {
130+
if (jsonRpcRequest.hasParams()) {
131+
Object[] args = getArgsAsObjects(reflectionInfo.params, jsonRpcRequest);
132+
multi = (Multi<?>) reflectionInfo.method.invoke(target, args);
133+
} else {
134+
multi = (Multi<?>) reflectionInfo.method.invoke(target);
135+
}
136+
} catch (Exception e) {
137+
logger.errorf(e, "Unable to invoke method %s using JSON-RPC, request was: %s", jsonRpcMethodName,
138+
jsonRpcRequest);
139+
s.writeTextMessage(JsonRpcWriter.writeErrorResponse(jsonRpcRequest.getId(), jsonRpcMethodName, e).encode());
140+
return;
133141
}
134142

135-
return jsonRpcResponse;
136-
} catch (IllegalAccessException | IllegalArgumentException | InvocationTargetException ex) {
137-
throw new RuntimeException(ex);
143+
Cancellable cancellable = multi.subscribe()
144+
.with(
145+
item -> {
146+
JsonObject jsonResponse = JsonRpcWriter.writeResponse(jsonRpcRequest.getId(), item,
147+
MessageType.SubscriptionMessage);
148+
s.writeTextMessage(jsonResponse.encodePrettily());
149+
},
150+
failure -> {
151+
s.writeTextMessage(JsonRpcWriter
152+
.writeErrorResponse(jsonRpcRequest.getId(), jsonRpcMethodName, failure).encode());
153+
this.subscriptions.remove(jsonRpcRequest.getId());
154+
},
155+
() -> this.subscriptions.remove(jsonRpcRequest.getId()));
156+
157+
this.subscriptions.put(jsonRpcRequest.getId(), cancellable);
158+
s.writeTextMessage(JsonRpcWriter.writeResponse(jsonRpcRequest.getId(), null, MessageType.Void).encode());
159+
} else {
160+
// The invocation will return a Uni<JsonObject>
161+
Uni<?> uni;
162+
try {
163+
if (jsonRpcRequest.hasParams()) {
164+
Object[] args = getArgsAsObjects(reflectionInfo.params, jsonRpcRequest);
165+
uni = invoke(reflectionInfo, target, args);
166+
} else {
167+
uni = invoke(reflectionInfo, target, new Object[0]);
168+
}
169+
} catch (Exception e) {
170+
logger.errorf(e, "Unable to invoke method %s using JSON-RPC, request was: %s", jsonRpcMethodName,
171+
jsonRpcRequest);
172+
s.writeTextMessage(JsonRpcWriter.writeErrorResponse(jsonRpcRequest.getId(), jsonRpcMethodName, e).encode());
173+
return;
174+
}
175+
uni.subscribe()
176+
.with(item -> {
177+
s.writeTextMessage(JsonRpcWriter.writeResponse(jsonRpcRequest.getId(), item,
178+
MessageType.Response).encode());
179+
}, failure -> {
180+
s.writeTextMessage(JsonRpcWriter
181+
.writeErrorResponse(jsonRpcRequest.getId(), jsonRpcMethodName, failure).encode());
182+
});
138183
}
184+
} else {
185+
// Method not found
186+
s.writeTextMessage(JsonRpcWriter.writeMethodNotFoundResponse(jsonRpcRequest.getId(), jsonRpcMethodName).encode());
139187
}
140-
141-
// Method not found
142-
return JsonRpcWriter.writeMethodNotFoundResponse(jsonRpcRequest.getId(), jsonRpcMethodName);
143-
144188
}
145189

146190
private Object[] getArgsAsObjects(Map<String, Class> params, JsonRpcReader jsonRpcRequest) {

‎extensions/vertx-http/runtime/src/main/java/io/quarkus/devui/runtime/comms/ReflectionInfo.java‎

Lines changed: 24 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,25 +4,47 @@
44
import java.util.Map;
55

66
import io.smallrye.mutiny.Multi;
7+
import io.smallrye.mutiny.Uni;
78

89
/**
910
* Contains reflection info on the beans that needs to be called from the jsonrpc router
1011
*/
1112
public class ReflectionInfo {
13+
private final boolean blocking;
14+
private final boolean nonBlocking;
1215
public Class bean;
1316
public Object instance;
1417
public Method method;
1518
public Map<String, Class> params;
1619

17-
public ReflectionInfo(Class bean, Object instance, Method method, Map<String, Class> params) {
20+
public ReflectionInfo(Class bean, Object instance, Method method, Map<String, Class> params, boolean explicitlyBlocking,
21+
boolean explicitlyNonBlocking) {
1822
this.bean = bean;
1923
this.instance = instance;
2024
this.method = method;
2125
this.params = params;
26+
this.blocking = explicitlyBlocking;
27+
this.nonBlocking = explicitlyNonBlocking;
28+
if (blocking && nonBlocking) {
29+
throw new IllegalArgumentException("The method " + method.getDeclaringClass().getName() + "." + method.getName()
30+
+ " cannot be annotated with @Blocking and @NonBlocking");
31+
}
2232
}
2333

24-
public boolean isSubscription() {
34+
public boolean isReturningMulti() {
2535
Class<?> returnType = this.method.getReturnType();
2636
return returnType.getName().equals(Multi.class.getName());
2737
}
38+
39+
public boolean isExplicitlyBlocking() {
40+
return blocking;
41+
}
42+
43+
public boolean isExplicitlyNonBlocking() {
44+
return nonBlocking;
45+
}
46+
47+
public boolean isReturningUni() {
48+
return method.getReturnType().getName().equals(Uni.class.getName());
49+
}
2850
}

‎extensions/vertx-http/runtime/src/main/java/io/quarkus/devui/runtime/jsonrpc/JsonRpcMethod.java‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,9 @@ public final class JsonRpcMethod {
77
private String methodName;
88
private Map<String, Class> params;
99

10+
private boolean isExplicitlyBlocking;
11+
private boolean isExplicitlyNonBlocking;
12+
1013
public JsonRpcMethod() {
1114
}
1215

@@ -44,6 +47,22 @@ public void setParams(Map<String, Class> params) {
4447
this.params = params;
4548
}
4649

50+
public boolean getExplicitlyBlocking() {
51+
return isExplicitlyBlocking;
52+
}
53+
54+
public void setExplicitlyBlocking(boolean blocking) {
55+
isExplicitlyBlocking = blocking;
56+
}
57+
58+
public boolean getExplicitlyNonBlocking() {
59+
return isExplicitlyNonBlocking;
60+
}
61+
62+
public void setExplicitlyNonBlocking(boolean nonblocking) {
63+
isExplicitlyNonBlocking = nonblocking;
64+
}
65+
4766
@Override
4867
public String toString() {
4968
return clazz.getName() + ":" + methodName + "(" + params + ")";

‎extensions/vertx-http/runtime/src/main/java/io/quarkus/devui/runtime/jsonrpc/JsonRpcWriter.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
import static io.quarkus.devui.runtime.jsonrpc.JsonRpcKeys.CODE;
44
import static io.quarkus.devui.runtime.jsonrpc.JsonRpcKeys.ERROR;
55
import static io.quarkus.devui.runtime.jsonrpc.JsonRpcKeys.ID;
6+
import static io.quarkus.devui.runtime.jsonrpc.JsonRpcKeys.INTERNAL_ERROR;
67
import static io.quarkus.devui.runtime.jsonrpc.JsonRpcKeys.JSONRPC;
78
import static io.quarkus.devui.runtime.jsonrpc.JsonRpcKeys.MESSAGE;
89
import static io.quarkus.devui.runtime.jsonrpc.JsonRpcKeys.MESSAGE_TYPE;
@@ -43,4 +44,15 @@ public static JsonObject writeMethodNotFoundResponse(int id, String jsonRpcMetho
4344
ERROR, jsonRpcError);
4445
}
4546

47+
public static JsonObject writeErrorResponse(int id, String jsonRpcMethodName, Throwable exception) {
48+
JsonObject jsonRpcError = JsonObject.of(
49+
CODE, INTERNAL_ERROR,
50+
MESSAGE, "Method [" + jsonRpcMethodName + "] failed: " + exception.getMessage());
51+
52+
return JsonObject.of(
53+
ID, id,
54+
JSONRPC, VERSION,
55+
ERROR, jsonRpcError);
56+
}
57+
4658
}

0 commit comments

Comments
 (0)