Skip to content

Commit ce8d851

Browse files
artem-vartem-v
committed
Refactored request-releaser
1 parent c274c9b commit ce8d851

File tree

5 files changed

+44
-115
lines changed

5 files changed

+44
-115
lines changed

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

Lines changed: 30 additions & 85 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,6 @@
55
import io.scalecube.net.Address;
66
import io.scalecube.services.api.ErrorData;
77
import io.scalecube.services.api.ServiceMessage;
8-
import io.scalecube.services.exceptions.DefaultErrorMapper;
98
import io.scalecube.services.exceptions.ServiceClientErrorMapper;
109
import io.scalecube.services.exceptions.ServiceUnavailableException;
1110
import io.scalecube.services.methods.MethodInfo;
@@ -22,8 +21,6 @@
2221
import java.util.HashMap;
2322
import java.util.Map;
2423
import java.util.Optional;
25-
import java.util.concurrent.Callable;
26-
import java.util.function.Consumer;
2724
import java.util.function.Function;
2825
import org.reactivestreams.Publisher;
2926
import org.slf4j.Logger;
@@ -39,66 +36,33 @@ public class ServiceCall {
3936
private static final ServiceMessage UNEXPECTED_EMPTY_RESPONSE =
4037
ServiceMessage.error(503, 503, "Unexpected empty response");
4138

42-
private ClientTransport transport;
43-
private ServiceMethodRegistry methodRegistry;
44-
private ServiceRegistry serviceRegistry;
39+
private final ClientTransport transport;
40+
private final ServiceMethodRegistry methodRegistry;
41+
private final ServiceRegistry serviceRegistry;
42+
4543
private Router router;
46-
private ServiceClientErrorMapper errorMapper = DefaultErrorMapper.INSTANCE;
47-
private Consumer<Object> requestReleaser =
48-
req -> {
49-
// no-op
50-
};
44+
private ServiceClientErrorMapper errorMapper;
5145
private Map<String, String> credentials = Collections.emptyMap();
5246
private String contentType;
5347

54-
/** Default constructor. */
55-
public ServiceCall() {}
48+
ServiceCall(
49+
ClientTransport transport,
50+
ServiceMethodRegistry methodRegistry,
51+
ServiceRegistry serviceRegistry) {
52+
this.transport = transport;
53+
this.methodRegistry = methodRegistry;
54+
this.serviceRegistry = serviceRegistry;
55+
}
5656

5757
private ServiceCall(ServiceCall other) {
5858
this.transport = other.transport;
5959
this.methodRegistry = other.methodRegistry;
6060
this.serviceRegistry = other.serviceRegistry;
61+
// Set resettable fields
6162
this.router = other.router;
6263
this.errorMapper = other.errorMapper;
6364
this.contentType = other.contentType;
64-
this.requestReleaser = other.requestReleaser;
65-
this.credentials = new HashMap<>(other.credentials);
66-
}
67-
68-
/**
69-
* Creates new {@link ServiceCall}'s definition with a given client transport.
70-
*
71-
* @param clientTransport client transport.
72-
* @return new {@link ServiceCall} instance.
73-
*/
74-
public ServiceCall transport(ClientTransport clientTransport) {
75-
ServiceCall target = new ServiceCall(this);
76-
target.transport = clientTransport;
77-
return target;
78-
}
79-
80-
/**
81-
* Creates new {@link ServiceCall}'s definition with a given service registry.
82-
*
83-
* @param serviceRegistry service registry.
84-
* @return new {@link ServiceCall} instance.
85-
*/
86-
public ServiceCall serviceRegistry(ServiceRegistry serviceRegistry) {
87-
ServiceCall target = new ServiceCall(this);
88-
target.serviceRegistry = serviceRegistry;
89-
return target;
90-
}
91-
92-
/**
93-
* Creates new {@link ServiceCall}'s definition with a given method registry.
94-
*
95-
* @param methodRegistry method registry.
96-
* @return new {@link ServiceCall} instance.
97-
*/
98-
public ServiceCall methodRegistry(ServiceMethodRegistry methodRegistry) {
99-
ServiceCall target = new ServiceCall(this);
100-
target.methodRegistry = methodRegistry;
101-
return target;
65+
this.credentials = Collections.unmodifiableMap(new HashMap<>(other.credentials));
10266
}
10367

10468
/**
@@ -137,18 +101,6 @@ public ServiceCall errorMapper(ServiceClientErrorMapper errorMapper) {
137101
return target;
138102
}
139103

140-
/**
141-
* Creates new {@link ServiceCall}'s definition with a given requestReleaser.
142-
*
143-
* @param requestReleaser given.
144-
* @return new {@link ServiceCall} instance.
145-
*/
146-
public ServiceCall requestReleaser(Consumer<Object> requestReleaser) {
147-
ServiceCall target = new ServiceCall(this);
148-
target.requestReleaser = requestReleaser;
149-
return target;
150-
}
151-
152104
/**
153105
* Creates new {@link ServiceCall}'s definition with a given credentials.
154106
*
@@ -219,11 +171,12 @@ public Mono<ServiceMessage> requestOne(ServiceMessage request, Type responseType
219171
&& methodRegistry.containsInvoker(qualifier)) { // local service
220172
return methodRegistry
221173
.getInvoker(request.qualifier())
222-
.invokeOne(request, requestReleaser)
174+
.invokeOne(request)
223175
.map(this::throwIfError);
224176
} else {
225-
return addressLookup(request)
226-
.flatMap(address -> requestOne(request, responseType, address)); // remote service
177+
// remote service
178+
return Mono.fromCallable(() -> addressLookup(request))
179+
.flatMap(address -> requestOne(request, responseType, address));
227180
}
228181
});
229182
}
@@ -274,12 +227,13 @@ public Flux<ServiceMessage> requestMany(ServiceMessage request, Type responseTyp
274227
&& methodRegistry.containsInvoker(qualifier)) { // local service
275228
return methodRegistry
276229
.getInvoker(request.qualifier())
277-
.invokeMany(request, requestReleaser)
230+
.invokeMany(request)
278231
.map(this::throwIfError);
279232
} else {
280-
return addressLookup(request)
233+
// remote service
234+
return Mono.fromCallable(() -> addressLookup(request))
281235
.flatMapMany(
282-
address -> requestMany(request, responseType, address)); // remote service
236+
address -> requestMany(request, responseType, address));
283237
}
284238
});
285239
}
@@ -335,11 +289,11 @@ public Flux<ServiceMessage> requestBidirectional(
335289
&& methodRegistry.containsInvoker(qualifier)) { // local service
336290
return methodRegistry
337291
.getInvoker(qualifier)
338-
.invokeBidirectional(messages, requestReleaser)
292+
.invokeBidirectional(messages)
339293
.map(this::throwIfError);
340294
} else {
341295
// remote service
342-
return addressLookup(request)
296+
return Mono.fromCallable(() -> addressLookup(request))
343297
.flatMapMany(
344298
address -> requestBidirectional(messages, responseType, address));
345299
}
@@ -437,14 +391,11 @@ public Object invoke(Object proxy, Method method, Object[] params) {
437391
});
438392
}
439393

440-
private Mono<Address> addressLookup(ServiceMessage request) {
441-
Callable<Address> callable =
442-
() ->
443-
router
444-
.route(serviceRegistry, request)
445-
.map(ServiceReference::address)
446-
.orElseThrow(() -> noReachableMemberException(request));
447-
return Mono.fromCallable(callable).doOnError(th -> applyRequestReleaser(request));
394+
private Address addressLookup(ServiceMessage request) {
395+
return router
396+
.route(serviceRegistry, request)
397+
.map(ServiceReference::address)
398+
.orElseThrow(() -> noReachableMemberException(request));
448399
}
449400

450401
private ServiceMessage toServiceMessage(MethodInfo methodInfo, Object request) {
@@ -517,10 +468,4 @@ private ServiceMessage throwIfError(ServiceMessage message) {
517468
}
518469
return message;
519470
}
520-
521-
private void applyRequestReleaser(ServiceMessage request) {
522-
if (request.data() != null) {
523-
requestReleaser.accept(request.data());
524-
}
525-
}
526471
}

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

Lines changed: 3 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,6 @@
1111
import java.lang.reflect.Method;
1212
import java.util.Optional;
1313
import java.util.StringJoiner;
14-
import java.util.function.Consumer;
1514
import org.reactivestreams.Publisher;
1615
import reactor.core.publisher.Flux;
1716
import reactor.core.publisher.Mono;
@@ -60,12 +59,10 @@ public ServiceMethodInvoker(
6059
* Invokes service method with single response.
6160
*
6261
* @param message request service message
63-
* @param requestReleaser request releaser
6462
* @return mono of service message
6563
*/
66-
public Mono<ServiceMessage> invokeOne(ServiceMessage message, Consumer<Object> requestReleaser) {
64+
public Mono<ServiceMessage> invokeOne(ServiceMessage message) {
6765
return authenticate(message)
68-
.doOnError(th -> applyRequestReleaser(message, requestReleaser))
6966
.flatMap(principal -> Mono.from(invoke(toRequest(message), principal)))
7067
.map(response -> toResponse(response, message.dataFormat()))
7168
.onErrorResume(throwable -> Mono.just(errorMapper.toMessage(throwable)));
@@ -75,12 +72,10 @@ public Mono<ServiceMessage> invokeOne(ServiceMessage message, Consumer<Object> r
7572
* Invokes service method with message stream response.
7673
*
7774
* @param message request service message
78-
* @param requestReleaser request releaser
7975
* @return flux of service messages
8076
*/
81-
public Flux<ServiceMessage> invokeMany(ServiceMessage message, Consumer<Object> requestReleaser) {
77+
public Flux<ServiceMessage> invokeMany(ServiceMessage message) {
8278
return authenticate(message)
83-
.doOnError(th -> applyRequestReleaser(message, requestReleaser))
8479
.flatMapMany(principal -> Flux.from(invoke(toRequest(message), principal)))
8580
.map(response -> toResponse(response, message.dataFormat()))
8681
.onErrorResume(throwable -> Flux.just(errorMapper.toMessage(throwable)));
@@ -90,16 +85,13 @@ public Flux<ServiceMessage> invokeMany(ServiceMessage message, Consumer<Object>
9085
* Invokes service method with bidirectional communication.
9186
*
9287
* @param publisher request service message
93-
* @param requestReleaser request releaser
9488
* @return flux of service messages
9589
*/
96-
public Flux<ServiceMessage> invokeBidirectional(
97-
Publisher<ServiceMessage> publisher, Consumer<Object> requestReleaser) {
90+
public Flux<ServiceMessage> invokeBidirectional(Publisher<ServiceMessage> publisher) {
9891
return Flux.from(publisher)
9992
.switchOnFirst(
10093
(first, messages) ->
10194
authenticate(first.get())
102-
.doOnError(th -> applyRequestReleaser(first.get(), requestReleaser))
10395
.flatMapMany(
10496
principal ->
10597
messages
@@ -202,12 +194,6 @@ private ServiceMessage toResponse(Object response, String dataFormat) {
202194
.build();
203195
}
204196

205-
private void applyRequestReleaser(ServiceMessage request, Consumer<Object> requestReleaser) {
206-
if (request.data() != null) {
207-
requestReleaser.accept(request.data());
208-
}
209-
}
210-
211197
public Object service() {
212198
return service;
213199
}

‎services-api/src/test/java/io/scalecube/services/methods/ServiceMethodInvokerTest.java‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -60,7 +60,7 @@ void testInvokeOneWhenReturnNull() throws Exception {
6060
ServiceMessage message =
6161
ServiceMessage.builder().qualifier(qualifierPrefix + methodName).build();
6262

63-
StepVerifier.create(serviceMethodInvoker.invokeOne(message, requestReleaser)).verifyComplete();
63+
StepVerifier.create(serviceMethodInvoker.invokeOne(message)).verifyComplete();
6464
}
6565

6666
@Test
@@ -94,7 +94,7 @@ void testInvokeManyWhenReturnNull() throws Exception {
9494
ServiceMessage message =
9595
ServiceMessage.builder().qualifier(qualifierPrefix + methodName).build();
9696

97-
StepVerifier.create(serviceMethodInvoker.invokeMany(message, requestReleaser)).verifyComplete();
97+
StepVerifier.create(serviceMethodInvoker.invokeMany(message)).verifyComplete();
9898
}
9999

100100
@Test
@@ -129,7 +129,7 @@ void testInvokeBidirectionalWhenReturnNull() throws Exception {
129129
ServiceMessage.builder().qualifier(qualifierPrefix + methodName).build();
130130

131131
StepVerifier.create(
132-
serviceMethodInvoker.invokeBidirectional(Flux.just(message), requestReleaser))
132+
serviceMethodInvoker.invokeBidirectional(Flux.just(message)))
133133
.verifyComplete();
134134
}
135135

@@ -165,7 +165,7 @@ void testInvokeOneWhenThrowException() throws Exception {
165165
ServiceMessage.builder().qualifier(qualifierPrefix + methodName).build();
166166

167167
// invokeOne
168-
final Mono<ServiceMessage> invokeOne = serviceMethodInvoker.invokeOne(message, requestReleaser);
168+
final Mono<ServiceMessage> invokeOne = serviceMethodInvoker.invokeOne(message);
169169

170170
StepVerifier.create(invokeOne).assertNext(ServiceMessage::isError).verifyComplete();
171171
}
@@ -202,7 +202,7 @@ void testInvokeManyWhenThrowException() throws Exception {
202202
ServiceMessage.builder().qualifier(qualifierPrefix + methodName).build();
203203

204204
final Flux<ServiceMessage> invokeOne =
205-
serviceMethodInvoker.invokeMany(message, requestReleaser);
205+
serviceMethodInvoker.invokeMany(message);
206206

207207
StepVerifier.create(invokeOne).assertNext(ServiceMessage::isError).verifyComplete();
208208
}
@@ -240,7 +240,7 @@ void testInvokeBidirectionalWhenThrowException() throws Exception {
240240

241241
// invokeOne
242242
final Flux<ServiceMessage> invokeOne =
243-
serviceMethodInvoker.invokeBidirectional(Flux.just(message), requestReleaser);
243+
serviceMethodInvoker.invokeBidirectional(Flux.just(message));
244244

245245
StepVerifier.create(invokeOne).assertNext(ServiceMessage::isError).verifyComplete();
246246
}

‎services-transport-parent/services-transport-rsocket/src/main/java/io/scalecube/services/transport/rsocket/RSocketServiceAcceptor.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -54,7 +54,7 @@ public Mono<Payload> requestResponse(Payload payload) {
5454
.flatMap(
5555
message -> {
5656
ServiceMethodInvoker methodInvoker = methodRegistry.getInvoker(message.qualifier());
57-
return methodInvoker.invokeOne(message, requestReleaser);
57+
return methodInvoker.invokeOne(message);
5858
})
5959
.map(this::toPayload);
6060
}
@@ -66,7 +66,7 @@ public Flux<Payload> requestStream(Payload payload) {
6666
.flatMapMany(
6767
message -> {
6868
ServiceMethodInvoker methodInvoker = methodRegistry.getInvoker(message.qualifier());
69-
return methodInvoker.invokeMany(message, requestReleaser);
69+
return methodInvoker.invokeMany(message);
7070
})
7171
.map(this::toPayload);
7272
}
@@ -82,7 +82,7 @@ public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
8282
validateRequest(message);
8383
ServiceMethodInvoker methodInvoker =
8484
methodRegistry.getInvoker(message.qualifier());
85-
return methodInvoker.invokeBidirectional(messages, requestReleaser);
85+
return methodInvoker.invokeBidirectional(messages);
8686
}
8787
return messages;
8888
})

‎services/src/main/java/io/scalecube/services/Microservices.java‎

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -242,11 +242,9 @@ public Address serviceAddress() {
242242
* @return new {@code ServiceCall} instance.
243243
*/
244244
public ServiceCall call() {
245-
return new ServiceCall()
246-
.transport(transportBootstrap.clientTransport)
247-
.serviceRegistry(serviceRegistry)
248-
.methodRegistry(methodRegistry)
245+
return new ServiceCall(transportBootstrap.clientTransport, methodRegistry, serviceRegistry)
249246
.contentType(contentType)
247+
.errorMapper(DefaultErrorMapper.INSTANCE)
250248
.router(Routers.getRouter(RoundRobinServiceRouter.class));
251249
}
252250

0 commit comments

Comments
 (0)