Skip to content

Commit a23f9f7

Browse files
segabrielartem-v
segabriel
and
artem-v
authored
Polymorphic/multi-type responses (#960)
* Use dataType header for encoding/decoding data --------- Co-authored-by: Artem Vysochyn <artem-v@users.noreply.github.com> Co-authored-by: Artem Vysochyn <artem@exberry.io>
1 parent a86d6c0 commit a23f9f7

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

45 files changed

+1584
-185
lines changed

‎pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -59,7 +59,7 @@
5959
<properties>
6060
<scalecube-cluster.version>2.7.7</scalecube-cluster.version>
6161
<scalecube-security.version>1.1.11</scalecube-security.version>
62-
<scalecube-test-support.version>0.1.3</scalecube-test-support.version>
62+
<scalecube-test-support.version>0.1.4</scalecube-test-support.version>
6363

6464
<distributionManagement.url>https://maven.pkg.github.com/scalecube/scalecube-services
6565
</distributionManagement.url>

‎services-api/src/main/java/io/scalecube/services/ServiceCall.java‎

Lines changed: 63 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
import static io.scalecube.services.Reflect.requestType;
99
import static io.scalecube.services.Reflect.restMethod;
1010
import static io.scalecube.services.Reflect.serviceName;
11+
import static io.scalecube.services.api.ServiceMessage.HEADER_PROPAGATE_DATA_TYPE_HEADER;
1112
import static io.scalecube.services.auth.Principal.NULL_PRINCIPAL;
1213

1314
import io.scalecube.services.api.ErrorData;
@@ -21,6 +22,7 @@
2122
import io.scalecube.services.routing.Router;
2223
import io.scalecube.services.routing.Routers;
2324
import io.scalecube.services.transport.api.ClientTransport;
25+
import io.scalecube.services.transport.api.ServiceMessageDataDecoder;
2426
import java.lang.reflect.Method;
2527
import java.lang.reflect.Proxy;
2628
import java.lang.reflect.Type;
@@ -29,6 +31,7 @@
2931
import java.util.Map;
3032
import java.util.Objects;
3133
import java.util.Optional;
34+
import java.util.concurrent.ConcurrentHashMap;
3235
import java.util.function.Function;
3336
import org.reactivestreams.Publisher;
3437
import reactor.core.Exceptions;
@@ -44,8 +47,9 @@ public class ServiceCall implements AutoCloseable {
4447
private ServiceClientErrorMapper errorMapper = DefaultErrorMapper.INSTANCE;
4548
private Map<String, String> credentials = Collections.emptyMap();
4649
private String contentType = ServiceMessage.DEFAULT_DATA_FORMAT;
50+
private ServiceMessageDataDecoder dataDecoder = ServiceMessageDataDecoder.INSTANCE;
4751

48-
// private Logger logger;
52+
private final Map<String, Type> resolvedTypes = new ConcurrentHashMap<>();
4953

5054
public ServiceCall() {}
5155

@@ -55,6 +59,7 @@ private ServiceCall(ServiceCall other) {
5559
this.router = other.router;
5660
this.errorMapper = other.errorMapper;
5761
this.contentType = other.contentType;
62+
this.dataDecoder = other.dataDecoder;
5863
this.credentials = Collections.unmodifiableMap(new HashMap<>(other.credentials));
5964
}
6065

@@ -142,6 +147,18 @@ public ServiceCall contentType(String contentType) {
142147
return target;
143148
}
144149

150+
/**
151+
* Setter for {@code dataDecoder}.
152+
*
153+
* @param dataDecoder dataDecoder.
154+
* @return new {@link ServiceCall} instance.
155+
*/
156+
public ServiceCall dataDecoder(ServiceMessageDataDecoder dataDecoder) {
157+
ServiceCall target = new ServiceCall(this);
158+
target.dataDecoder = dataDecoder;
159+
return target;
160+
}
161+
145162
/**
146163
* Invokes fire-and-forget request.
147164
*
@@ -178,7 +195,7 @@ public Mono<ServiceMessage> requestOne(ServiceMessage request, Type responseType
178195
// local service
179196
return methodInvoker
180197
.invokeOne(request)
181-
.map(this::throwIfError)
198+
.map(message -> onMessage(message, responseType))
182199
.contextWrite(
183200
context -> {
184201
if (context.hasKey(RequestContext.class)) {
@@ -198,8 +215,8 @@ public Mono<ServiceMessage> requestOne(ServiceMessage request, Type responseType
198215
serviceReference ->
199216
transport
200217
.create(serviceReference)
201-
.requestResponse(request, responseType)
202-
.map(this::throwIfError));
218+
.requestResponse(request)
219+
.map(message -> onMessage(message, responseType)));
203220
}
204221
});
205222
}
@@ -230,7 +247,7 @@ public Flux<ServiceMessage> requestMany(ServiceMessage request, Type responseTyp
230247
// local service
231248
return methodInvoker
232249
.invokeMany(request)
233-
.map(this::throwIfError)
250+
.map(message -> onMessage(message, responseType))
234251
.contextWrite(
235252
context -> {
236253
if (context.hasKey(RequestContext.class)) {
@@ -250,8 +267,8 @@ public Flux<ServiceMessage> requestMany(ServiceMessage request, Type responseTyp
250267
serviceReference ->
251268
transport
252269
.create(serviceReference)
253-
.requestStream(request, responseType)
254-
.map(this::throwIfError));
270+
.requestStream(request)
271+
.map(message -> onMessage(message, responseType)));
255272
}
256273
});
257274
}
@@ -286,7 +303,7 @@ public Flux<ServiceMessage> requestBidirectional(
286303
// local service
287304
return methodInvoker
288305
.invokeBidirectional(messages)
289-
.map(this::throwIfError)
306+
.map(message -> onMessage(message, responseType))
290307
.contextWrite(
291308
context -> {
292309
if (context.hasKey(RequestContext.class)) {
@@ -306,8 +323,8 @@ public Flux<ServiceMessage> requestBidirectional(
306323
serviceReference ->
307324
transport
308325
.create(serviceReference)
309-
.requestChannel(messages, responseType)
310-
.map(this::throwIfError));
326+
.requestChannel(messages)
327+
.map(message -> onMessage(message, responseType)));
311328
}
312329
}
313330
return messages;
@@ -354,10 +371,9 @@ public <T> T api(Class<T> serviceInterface) {
354371
case REQUEST_CHANNEL:
355372
// this is REQUEST_CHANNEL so it means params[0] must
356373
// be a publisher - its safe to cast.
357-
//noinspection rawtypes
358374
return serviceCall
359375
.requestBidirectional(
360-
Flux.from((Publisher) request)
376+
Flux.from((Publisher<?>) request)
361377
.map(data -> toServiceMessage(methodInfo, data)),
362378
returnType)
363379
.transform(asFlux(isReturnTypeServiceMessage));
@@ -376,18 +392,15 @@ private ServiceReference serviceLookup(ServiceMessage request) {
376392
}
377393

378394
private ServiceMessage toServiceMessage(MethodInfo methodInfo, Object request) {
379-
if (request instanceof ServiceMessage) {
380-
return ServiceMessage.from((ServiceMessage) request)
381-
.qualifier(methodInfo.serviceName(), methodInfo.methodName())
382-
.headers(credentials)
383-
.dataFormatIfAbsent(contentType)
384-
.build();
385-
}
395+
final var builder =
396+
request instanceof ServiceMessage
397+
? ServiceMessage.from((ServiceMessage) request)
398+
: ServiceMessage.builder().data(request);
386399

387-
return ServiceMessage.builder()
400+
return builder
388401
.qualifier(methodInfo.serviceName(), methodInfo.methodName())
389402
.headers(credentials)
390-
.data(request)
403+
.header(HEADER_PROPAGATE_DATA_TYPE_HEADER, true)
391404
.dataFormatIfAbsent(contentType)
392405
.build();
393406
}
@@ -438,13 +451,6 @@ private static Function<Mono<ServiceMessage>, Mono<Object>> asMono(
438451
: mono.mapNotNull(ServiceMessage::data);
439452
}
440453

441-
private ServiceMessage throwIfError(ServiceMessage message) {
442-
if (message.isError() && message.hasData(ErrorData.class)) {
443-
throw Exceptions.propagate(errorMapper.toError(message));
444-
}
445-
return message;
446-
}
447-
448454
private static MethodInfo getMethodInfo(Class<?> serviceInterface, Method method) {
449455
return new MethodInfo(
450456
serviceName(serviceInterface),
@@ -461,6 +467,35 @@ private static MethodInfo getMethodInfo(Class<?> serviceInterface, Method method
461467
Collections.emptyList());
462468
}
463469

470+
private ServiceMessage onMessage(ServiceMessage message, Type returnType) {
471+
if (returnType == null) {
472+
return message;
473+
}
474+
475+
if (message.isError()) {
476+
throw Exceptions.propagate(
477+
errorMapper.toError(dataDecoder.decodeData(message, ErrorData.class)));
478+
}
479+
480+
return dataDecoder.decodeData(
481+
message, TypeUtil.isWildcardType(returnType) ? getDataType(message) : returnType);
482+
}
483+
484+
private Type getDataType(ServiceMessage message) {
485+
final var dataType = message.header(ServiceMessage.HEADER_DATA_TYPE);
486+
487+
if (dataType == null) {
488+
return Object.class;
489+
}
490+
491+
return resolvedTypes.computeIfAbsent(
492+
dataType,
493+
s -> {
494+
final var type = TypeUtil.parseTypeDescriptor(dataType);
495+
return type != null ? type : Object.class;
496+
});
497+
}
498+
464499
@Override
465500
public void close() {
466501
if (transport != null) {

‎services-api/src/main/java/io/scalecube/services/TypeUtil.java‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
package io.scalecube.services;
2+
3+
import java.lang.reflect.Type;
4+
import java.lang.reflect.WildcardType;
5+
6+
public final class TypeUtil {
7+
8+
private TypeUtil() {
9+
// Do not instantiate
10+
}
11+
12+
public static String getTypeDescriptor(Object object) {
13+
return object != null ? object.getClass().getName() : null;
14+
}
15+
16+
public static Type parseTypeDescriptor(String descriptor) {
17+
if (descriptor == null) {
18+
return null;
19+
}
20+
try {
21+
return Class.forName(descriptor);
22+
} catch (ClassNotFoundException e) {
23+
throw new IllegalArgumentException("Invalid or unknown type: " + descriptor, e);
24+
}
25+
}
26+
27+
public static boolean isWildcardType(Type type) {
28+
return type == Object.class || type instanceof WildcardType;
29+
}
30+
}

‎services-api/src/main/java/io/scalecube/services/api/ServiceMessage.java‎

Lines changed: 28 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -16,12 +16,8 @@ public final class ServiceMessage {
1616
/** Qualifier header. */
1717
public static final String HEADER_QUALIFIER = "q";
1818

19-
/**
20-
* This is a system header which used by transport for serialization and deserialization purpose.
21-
* It is not supposed to be used by application directly and it is subject to changes in future
22-
* releases.
23-
*/
24-
public static final String HEADER_DATA_TYPE = "type";
19+
/** Data type header. */
20+
public static final String HEADER_DATA_TYPE = "dataType";
2521

2622
/** Data format header. */
2723
public static final String HEADER_DATA_FORMAT = "dataFormat";
@@ -35,6 +31,9 @@ public final class ServiceMessage {
3531
/** Upload filename header. */
3632
public static final String HEADER_UPLOAD_FILENAME = "uploadFilename";
3733

34+
/** Propagate data-type header. */
35+
public static final String HEADER_PROPAGATE_DATA_TYPE_HEADER = "propagateDataType";
36+
3837
/** Null value for error type. */
3938
public static final int NULL_ERROR_TYPE = -1;
4039

@@ -128,6 +127,15 @@ public String qualifier() {
128127
return header(HEADER_QUALIFIER);
129128
}
130129

130+
/**
131+
* Returns data type of the message data.
132+
*
133+
* @return data type of the data
134+
*/
135+
public String dataType() {
136+
return header(HEADER_DATA_TYPE);
137+
}
138+
131139
/**
132140
* Returns data format of the message data.
133141
*
@@ -223,6 +231,16 @@ public String uploadFilename() {
223231
return headers.get(HEADER_UPLOAD_FILENAME);
224232
}
225233

234+
/**
235+
* Returns whether data type header should be propagated downstream.
236+
*
237+
* @return whether data type header should be propagated downstream
238+
*/
239+
public boolean propagateDataType() {
240+
final var s = headers.get(HEADER_PROPAGATE_DATA_TYPE_HEADER);
241+
return Boolean.parseBoolean(s);
242+
}
243+
226244
@Override
227245
public String toString() {
228246
return new StringJoiner(", ", ServiceMessage.class.getSimpleName() + "[", "]")
@@ -275,14 +293,13 @@ public Builder data(Object data) {
275293
/**
276294
* Setter for {@code dataType}.
277295
*
278-
* @deprecated in future releases will be dropped without replacement
279296
* @param dataType data type; no null
280297
* @return this
281298
*/
282-
@Deprecated
283-
public Builder dataType(Class<?> dataType) {
284-
Objects.requireNonNull(dataType, "dataType");
285-
headers.put(HEADER_DATA_TYPE, dataType.getName());
299+
public Builder dataType(String dataType) {
300+
if (dataType != null) {
301+
headers.put(HEADER_DATA_TYPE, dataType);
302+
}
286303
return this;
287304
}
288305

‎services-api/src/main/java/io/scalecube/services/methods/ServiceMethodInvoker.java‎

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

33
import io.scalecube.services.CommunicationMode;
44
import io.scalecube.services.RequestContext;
5+
import io.scalecube.services.TypeUtil;
56
import io.scalecube.services.api.ServiceMessage;
67
import io.scalecube.services.auth.Principal;
78
import io.scalecube.services.auth.PrincipalMapper;
@@ -86,7 +87,7 @@ public Mono<ServiceMessage> invokeOne(ServiceMessage message) {
8687
}
8788
});
8889
})
89-
.map(response -> toResponse(response, message.qualifier(), message.dataFormat()))
90+
.map(response -> toResponse(response, message.headers()))
9091
.onErrorResume(ex -> Mono.just(errorMapper.toMessage(message.qualifier(), ex)))
9192
.subscribeOn(methodInfo.scheduler());
9293
}
@@ -140,7 +141,7 @@ public Flux<ServiceMessage> invokeMany(ServiceMessage message) {
140141
}
141142
});
142143
})
143-
.map(response -> toResponse(response, message.qualifier(), message.dataFormat()))
144+
.map(response -> toResponse(response, message.headers()))
144145
.onErrorResume(ex -> Flux.just(errorMapper.toMessage(message.qualifier(), ex)))
145146
.subscribeOn(methodInfo.scheduler());
146147
}
@@ -161,7 +162,6 @@ public Flux<ServiceMessage> invokeBidirectional(Publisher<ServiceMessage> publis
161162
final var message = first.get();
162163
final var request = copyRequest(message);
163164
final var qualifier = message.qualifier();
164-
final var dataFormat = message.dataFormat();
165165

166166
return mapPrincipal(context)
167167
.flatMapMany(
@@ -203,7 +203,7 @@ public Flux<ServiceMessage> invokeBidirectional(Publisher<ServiceMessage> publis
203203
ex);
204204
}
205205
})
206-
.map(response -> toResponse(response, qualifier, dataFormat))
206+
.map(response -> toResponse(response, message.headers()))
207207
.onErrorResume(ex -> Flux.just(errorMapper.toMessage(qualifier, ex)))
208208
.subscribeOn(methodInfo.scheduler());
209209
}));
@@ -277,18 +277,20 @@ private Object copyRequest(ServiceMessage message) {
277277
return methodInfo.isRequestTypeServiceMessage() ? request : request.data();
278278
}
279279

280-
private static ServiceMessage toResponse(Object response, String qualifier, String dataFormat) {
280+
private ServiceMessage toResponse(Object response, Map<String, String> headers) {
281+
final var dataType = getDataType(response);
282+
281283
if (response instanceof ServiceMessage message) {
282-
final var builder = ServiceMessage.from(message).qualifier(qualifier);
283-
return dataFormat != null && !dataFormat.equals(message.dataFormat())
284-
? builder.dataFormat(dataFormat).build()
285-
: builder.build();
284+
return ServiceMessage.from(message).dataType(dataType).build();
286285
}
287-
return ServiceMessage.builder()
288-
.qualifier(qualifier)
289-
.data(response)
290-
.dataFormatIfAbsent(dataFormat)
291-
.build();
286+
287+
return ServiceMessage.builder().headers(headers).dataType(dataType).data(response).build();
288+
}
289+
290+
private static String getDataType(Object response) {
291+
return response instanceof ServiceMessage message
292+
? getDataType(message.data())
293+
: TypeUtil.getTypeDescriptor(response);
292294
}
293295

294296
public Object service() {

0 commit comments

Comments
 (0)