Skip to content
This repository was archived by the owner on Jun 21, 2023. It is now read-only.

Commit 87d7bf1

Browse files
artem-vartem-v
authored
Added RetryEmitFailureHandler (#179)
1 parent 7ead33b commit 87d7bf1

File tree

4 files changed

+83
-25
lines changed

4 files changed

+83
-25
lines changed

‎services-gateway-client-transport/src/main/java/io/scalecube/services/gateway/transport/http/HttpGatewayClient.java

Lines changed: 17 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
package io.scalecube.services.gateway.transport.http;
22

3+
import static reactor.core.publisher.Sinks.EmitResult.FAIL_NON_SERIALIZED;
4+
35
import io.netty.buffer.ByteBuf;
46
import io.scalecube.services.api.ServiceMessage;
57
import io.scalecube.services.api.ServiceMessage.Builder;
@@ -12,7 +14,10 @@
1214
import org.slf4j.LoggerFactory;
1315
import reactor.core.publisher.Flux;
1416
import reactor.core.publisher.Mono;
17+
import reactor.core.publisher.SignalType;
1518
import reactor.core.publisher.Sinks;
19+
import reactor.core.publisher.Sinks.EmitFailureHandler;
20+
import reactor.core.publisher.Sinks.EmitResult;
1621
import reactor.netty.NettyOutbound;
1722
import reactor.netty.http.client.HttpClient;
1823
import reactor.netty.http.client.HttpClientRequest;
@@ -59,7 +64,7 @@ public HttpGatewayClient(GatewayClientSettings settings, GatewayClientCodec<Byte
5964
close
6065
.asMono()
6166
.then(doClose())
62-
.doFinally(s -> onClose.tryEmitEmpty())
67+
.doFinally(s -> onClose.emitEmpty(RetryEmitFailureHandler.INSTANCE))
6368
.doOnTerminate(() -> LOGGER.info("Closed HttpGatewayClient resources"))
6469
.subscribe(null, ex -> LOGGER.warn("Exception occurred on HttpGatewayClient close: " + ex));
6570
}
@@ -100,7 +105,7 @@ public Flux<ServiceMessage> requestChannel(Flux<ServiceMessage> requests) {
100105

101106
@Override
102107
public void close() {
103-
close.tryEmitEmpty();
108+
close.emitEmpty(RetryEmitFailureHandler.INSTANCE);
104109
}
105110

106111
@Override
@@ -134,4 +139,14 @@ private ServiceMessage toMessage(HttpClientResponse httpResponse, ByteBuf conten
134139
private boolean isError(int httpCode) {
135140
return httpCode >= 400 && httpCode <= 599;
136141
}
142+
143+
private static class RetryEmitFailureHandler implements EmitFailureHandler {
144+
145+
private static final RetryEmitFailureHandler INSTANCE = new RetryEmitFailureHandler();
146+
147+
@Override
148+
public boolean onEmitFailure(SignalType signalType, EmitResult emitResult) {
149+
return emitResult == FAIL_NON_SERIALIZED;
150+
}
151+
}
137152
}

‎services-gateway-client-transport/src/main/java/io/scalecube/services/gateway/transport/rsocket/RSocketGatewayClient.java

Lines changed: 17 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
package io.scalecube.services.gateway.transport.rsocket;
22

3+
import static reactor.core.publisher.Sinks.EmitResult.FAIL_NON_SERIALIZED;
4+
35
import io.rsocket.Payload;
46
import io.rsocket.RSocket;
57
import io.rsocket.core.RSocketConnector;
@@ -15,7 +17,10 @@
1517
import org.slf4j.LoggerFactory;
1618
import reactor.core.publisher.Flux;
1719
import reactor.core.publisher.Mono;
20+
import reactor.core.publisher.SignalType;
1821
import reactor.core.publisher.Sinks;
22+
import reactor.core.publisher.Sinks.EmitFailureHandler;
23+
import reactor.core.publisher.Sinks.EmitResult;
1924
import reactor.netty.http.client.HttpClient;
2025
import reactor.netty.resources.ConnectionProvider;
2126
import reactor.netty.resources.LoopResources;
@@ -53,7 +58,7 @@ public RSocketGatewayClient(GatewayClientSettings settings, GatewayClientCodec<P
5358
close
5459
.asMono()
5560
.then(doClose())
56-
.doFinally(s -> onClose.tryEmitEmpty())
61+
.doFinally(s -> onClose.emitEmpty(RetryEmitFailureHandler.INSTANCE))
5762
.doOnTerminate(() -> LOGGER.info("Closed RSocketGatewayClient resources"))
5863
.subscribe(
5964
null, ex -> LOGGER.warn("Exception occurred on RSocketGatewayClient close: " + ex));
@@ -95,7 +100,7 @@ public Flux<ServiceMessage> requestChannel(Flux<ServiceMessage> requests) {
95100

96101
@Override
97102
public void close() {
98-
close.tryEmitEmpty();
103+
close.emitEmpty(RetryEmitFailureHandler.INSTANCE);
99104
}
100105

101106
@Override
@@ -181,4 +186,14 @@ private ServiceMessage toMessage(Payload payload) {
181186
LOGGER.debug("Received response {}", message);
182187
return message;
183188
}
189+
190+
private static class RetryEmitFailureHandler implements EmitFailureHandler {
191+
192+
private static final RetryEmitFailureHandler INSTANCE = new RetryEmitFailureHandler();
193+
194+
@Override
195+
public boolean onEmitFailure(SignalType signalType, EmitResult emitResult) {
196+
return emitResult == FAIL_NON_SERIALIZED;
197+
}
198+
}
184199
}

‎services-gateway-client-transport/src/main/java/io/scalecube/services/gateway/transport/websocket/WebsocketGatewayClient.java

Lines changed: 17 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
package io.scalecube.services.gateway.transport.websocket;
22

3+
import static reactor.core.publisher.Sinks.EmitResult.FAIL_NON_SERIALIZED;
4+
35
import io.netty.buffer.ByteBuf;
46
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
57
import io.scalecube.services.api.ServiceMessage;
@@ -13,7 +15,10 @@
1315
import org.slf4j.LoggerFactory;
1416
import reactor.core.publisher.Flux;
1517
import reactor.core.publisher.Mono;
18+
import reactor.core.publisher.SignalType;
1619
import reactor.core.publisher.Sinks;
20+
import reactor.core.publisher.Sinks.EmitFailureHandler;
21+
import reactor.core.publisher.Sinks.EmitResult;
1722
import reactor.netty.Connection;
1823
import reactor.netty.http.client.HttpClient;
1924
import reactor.netty.resources.ConnectionProvider;
@@ -74,7 +79,7 @@ public WebsocketGatewayClient(GatewayClientSettings settings, GatewayClientCodec
7479
close
7580
.asMono()
7681
.then(doClose())
77-
.doFinally(s -> onClose.tryEmitEmpty())
82+
.doFinally(s -> onClose.emitEmpty(RetryEmitFailureHandler.INSTANCE))
7883
.doOnTerminate(() -> LOGGER.info("Closed client"))
7984
.subscribe(null, ex -> LOGGER.warn("Failed to close client, cause: " + ex));
8085
}
@@ -120,7 +125,7 @@ public Flux<ServiceMessage> requestChannel(Flux<ServiceMessage> requests) {
120125

121126
@Override
122127
public void close() {
123-
close.tryEmitEmpty();
128+
close.emitEmpty(RetryEmitFailureHandler.INSTANCE);
124129
}
125130

126131
@Override
@@ -208,4 +213,14 @@ private void onReadIdle(Connection connection) {
208213
private ByteBuf encodeRequest(ServiceMessage message, long sid) {
209214
return codec.encode(ServiceMessage.from(message).header(STREAM_ID, sid).build());
210215
}
216+
217+
private static class RetryEmitFailureHandler implements EmitFailureHandler {
218+
219+
private static final RetryEmitFailureHandler INSTANCE = new RetryEmitFailureHandler();
220+
221+
@Override
222+
public boolean onEmitFailure(SignalType signalType, EmitResult emitResult) {
223+
return emitResult == FAIL_NON_SERIALIZED;
224+
}
225+
}
211226
}

‎services-gateway-client-transport/src/main/java/io/scalecube/services/gateway/transport/websocket/WebsocketGatewayClientSession.java

Lines changed: 32 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
package io.scalecube.services.gateway.transport.websocket;
22

3+
import static reactor.core.publisher.Sinks.EmitResult.FAIL_NON_SERIALIZED;
4+
35
import io.netty.buffer.ByteBuf;
46
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
57
import io.scalecube.services.api.ErrorData;
@@ -14,7 +16,10 @@
1416
import org.slf4j.Logger;
1517
import org.slf4j.LoggerFactory;
1618
import reactor.core.publisher.Mono;
19+
import reactor.core.publisher.SignalType;
1720
import reactor.core.publisher.Sinks;
21+
import reactor.core.publisher.Sinks.EmitFailureHandler;
22+
import reactor.core.publisher.Sinks.EmitResult;
1823
import reactor.netty.Connection;
1924
import reactor.netty.http.websocket.WebsocketInbound;
2025
import reactor.netty.http.websocket.WebsocketOutbound;
@@ -81,7 +86,7 @@ public final class WebsocketGatewayClientSession {
8186
});
8287

8388
connection.onDispose(
84-
() -> inboundProcessors.forEach((k, o) -> tryEmitError(o, CLOSED_CHANNEL_EXCEPTION)));
89+
() -> inboundProcessors.forEach((k, o) -> emitError(o, CLOSED_CHANNEL_EXCEPTION)));
8590
}
8691

8792
@SuppressWarnings({"rawtypes", "unchecked"})
@@ -162,54 +167,52 @@ private void handleResponse(ServiceMessage response, Object processor) {
162167
Optional<Signal> signalOptional =
163168
Optional.ofNullable(response.header(SIGNAL)).map(Signal::from);
164169

165-
if (signalOptional.isPresent()) {
166-
170+
if (!signalOptional.isPresent()) {
171+
// handle normal response
172+
emitNext(processor, response);
173+
} else {
167174
// handle completion signal
168175
Signal signal = signalOptional.get();
169176
if (signal == Signal.COMPLETE) {
170-
tryEmitComplete(processor);
177+
emitComplete(processor);
171178
}
172-
173179
if (signal == Signal.ERROR) {
174180
// decode error data to retrieve real error cause
175181
ServiceMessage errorMessage = codec.decodeData(response, ErrorData.class);
176-
tryEmitValue(processor, errorMessage);
182+
emitNext(processor, errorMessage);
177183
}
178-
} else {
179-
// handle normal response
180-
tryEmitValue(processor, response);
181184
}
182185
} catch (Exception e) {
183-
tryEmitError(processor, e);
186+
emitError(processor, e);
184187
}
185188
}
186189

187-
private static void tryEmitValue(Object processor, ServiceMessage message) {
190+
private static void emitNext(Object processor, ServiceMessage message) {
188191
if (processor instanceof Sinks.One) {
189192
//noinspection unchecked
190-
((Sinks.One<ServiceMessage>) processor).tryEmitValue(message);
193+
((Sinks.One<ServiceMessage>) processor).emitValue(message, RetryEmitFailureHandler.INSTANCE);
191194
}
192195
if (processor instanceof Sinks.Many) {
193196
//noinspection unchecked
194-
((Sinks.Many<ServiceMessage>) processor).tryEmitNext(message);
197+
((Sinks.Many<ServiceMessage>) processor).emitNext(message, RetryEmitFailureHandler.INSTANCE);
195198
}
196199
}
197200

198-
private static void tryEmitComplete(Object processor) {
201+
private static void emitComplete(Object processor) {
199202
if (processor instanceof Sinks.One) {
200-
((Sinks.One<?>) processor).tryEmitEmpty();
203+
((Sinks.One<?>) processor).emitEmpty(RetryEmitFailureHandler.INSTANCE);
201204
}
202205
if (processor instanceof Sinks.Many) {
203-
((Sinks.Many<?>) processor).tryEmitComplete();
206+
((Sinks.Many<?>) processor).emitComplete(RetryEmitFailureHandler.INSTANCE);
204207
}
205208
}
206209

207-
private static void tryEmitError(Object processor, Exception e) {
210+
private static void emitError(Object processor, Exception e) {
208211
if (processor instanceof Sinks.One) {
209-
((Sinks.One<?>) processor).tryEmitError(e);
212+
((Sinks.One<?>) processor).emitError(e, RetryEmitFailureHandler.INSTANCE);
210213
}
211214
if (processor instanceof Sinks.Many) {
212-
((Sinks.Many<?>) processor).tryEmitError(e);
215+
((Sinks.Many<?>) processor).emitError(e, RetryEmitFailureHandler.INSTANCE);
213216
}
214217
}
215218

@@ -219,4 +222,14 @@ public String toString() {
219222
.add("id=" + id)
220223
.toString();
221224
}
225+
226+
private static class RetryEmitFailureHandler implements EmitFailureHandler {
227+
228+
private static final RetryEmitFailureHandler INSTANCE = new RetryEmitFailureHandler();
229+
230+
@Override
231+
public boolean onEmitFailure(SignalType signalType, EmitResult emitResult) {
232+
return emitResult == FAIL_NON_SERIALIZED;
233+
}
234+
}
222235
}

0 commit comments

Comments
 (0)