Skip to content

Commit 5afbd9d

Browse files
author
OlegDokuka
committed
fixes with UnboundedProcessor and TCP / WS connections design
Signed-off-by: Oleh Dokuka <[email protected]> Signed-off-by: Oleh Dokuka <[email protected]> Signed-off-by: OlegDokuka <[email protected]>
1 parent 608c9eb commit 5afbd9d

File tree

10 files changed

+109
-182
lines changed

10 files changed

+109
-182
lines changed

‎rsocket-core/src/main/java/io/rsocket/core/RSocketRequester.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -388,7 +388,7 @@ private void terminate(Throwable e) {
388388
requesterLeaseTracker.dispose(e);
389389
}
390390

391-
final Collection<FrameHandler> activeStreamsCopy;
391+
final Collection<FrameHandler> activeStreamsCopy; // in case of graceful shut down is empty
392392
synchronized (this) {
393393
final IntObjectMap<FrameHandler> activeStreams = this.activeStreams;
394394
activeStreamsCopy = new ArrayList<>(activeStreams.values());

‎rsocket-core/src/main/java/io/rsocket/core/RequesterResponderSupport.java

+2-1
Original file line numberDiff line numberDiff line change
@@ -135,7 +135,8 @@ public int addAndGetNextStreamId(FrameHandler frameHandler) {
135135

136136
public synchronized boolean add(int streamId, FrameHandler frameHandler) {
137137
if (this.terminating) {
138-
throw new CanceledException("Disposed");
138+
throw new CanceledException(
139+
"This RSocket is either disposed or disposing, and no longer accepting new requests");
139140
}
140141

141142
final IntObjectMap<FrameHandler> activeStreams = this.activeStreams;

‎rsocket-core/src/main/java/io/rsocket/internal/BaseDuplexConnection.java

+2-1
Original file line numberDiff line numberDiff line change
@@ -27,8 +27,9 @@ public abstract class BaseDuplexConnection implements DuplexConnection {
2727
protected final UnboundedProcessor sender =
2828
new UnboundedProcessor(
2929
() -> {
30+
System.out.println("queue is done");
3031
onClose.tryEmitEmpty();
31-
dispose();
32+
// dispose();
3233
},
3334
(__) -> {});
3435

‎rsocket-core/src/main/java/io/rsocket/internal/UnboundedProcessor.java

+2-1
Original file line numberDiff line numberDiff line change
@@ -203,8 +203,8 @@ public boolean tryEmitFinal(ByteBuf t) {
203203
return false;
204204
}
205205

206-
this.done = true;
207206
this.last = t;
207+
this.done = true;
208208

209209
final long previousState = markValueAddedAndTerminated(this);
210210
if (isFinalized(previousState)) {
@@ -216,6 +216,7 @@ public boolean tryEmitFinal(ByteBuf t) {
216216
if (this.outputFused) {
217217
// fast path for fusion
218218
this.actual.onNext(null);
219+
this.actual.onComplete();
219220
return true;
220221
}
221222

‎rsocket-core/src/main/java/io/rsocket/resume/ClientRSocketSession.java

+6-11
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,6 @@
2929
import io.rsocket.frame.ResumeOkFrameCodec;
3030
import io.rsocket.keepalive.KeepAliveSupport;
3131
import java.time.Duration;
32-
import java.util.concurrent.TimeoutException;
3332
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
3433
import java.util.function.Function;
3534
import org.reactivestreams.Subscription;
@@ -189,7 +188,7 @@ void tryReestablishSession(Tuple2<ByteBuf, DuplexConnection> tuple2) {
189188
}
190189
final ConnectionErrorException connectionErrorException =
191190
new ConnectionErrorException("RESUME_OK frame must be received before any others");
192-
resumableConnection.dispose(connectionErrorException);
191+
resumableConnection.dispose(nextDuplexConnection, connectionErrorException);
193192
nextDuplexConnection.sendErrorAndClose(connectionErrorException);
194193
nextDuplexConnection.receive().subscribe().dispose();
195194

@@ -227,7 +226,7 @@ void tryReestablishSession(Tuple2<ByteBuf, DuplexConnection> tuple2) {
227226
}
228227
final ConnectionErrorException t = new ConnectionErrorException(e.getMessage(), e);
229228

230-
resumableConnection.dispose(t);
229+
resumableConnection.dispose(nextDuplexConnection, t);
231230

232231
nextDuplexConnection.sendErrorAndClose(t);
233232
nextDuplexConnection.receive().subscribe().dispose();
@@ -278,7 +277,7 @@ void tryReestablishSession(Tuple2<ByteBuf, DuplexConnection> tuple2) {
278277
final ConnectionErrorException connectionErrorException =
279278
new ConnectionErrorException("resumption_server_pos=[" + remoteImpliedPos + "]");
280279

281-
resumableConnection.dispose(connectionErrorException);
280+
resumableConnection.dispose(nextDuplexConnection, connectionErrorException);
282281

283282
nextDuplexConnection.sendErrorAndClose(connectionErrorException);
284283
nextDuplexConnection.receive().subscribe().dispose();
@@ -292,7 +291,7 @@ void tryReestablishSession(Tuple2<ByteBuf, DuplexConnection> tuple2) {
292291
exception);
293292
}
294293
if (exception instanceof RejectedResumeException) {
295-
resumableConnection.dispose(exception);
294+
resumableConnection.dispose(nextDuplexConnection, exception);
296295
nextDuplexConnection.dispose();
297296
nextDuplexConnection.receive().subscribe().dispose();
298297
return;
@@ -309,7 +308,7 @@ void tryReestablishSession(Tuple2<ByteBuf, DuplexConnection> tuple2) {
309308
final ConnectionErrorException connectionErrorException =
310309
new ConnectionErrorException("RESUME_OK frame must be received before any others");
311310

312-
resumableConnection.dispose(connectionErrorException);
311+
resumableConnection.dispose(nextDuplexConnection, connectionErrorException);
313312

314313
nextDuplexConnection.sendErrorAndClose(connectionErrorException);
315314
nextDuplexConnection.receive().subscribe().dispose();
@@ -349,11 +348,7 @@ public void onError(Throwable t) {
349348
Operators.onErrorDropped(t, currentContext());
350349
}
351350

352-
if (t instanceof TimeoutException) {
353-
resumableConnection.dispose();
354-
} else {
355-
resumableConnection.dispose(t);
356-
}
351+
resumableConnection.dispose();
357352
}
358353

359354
@Override

‎rsocket-core/src/main/java/io/rsocket/resume/ResumableDuplexConnection.java

+61-51
Original file line numberDiff line numberDiff line change
@@ -21,8 +21,8 @@
2121
import io.netty.util.CharsetUtil;
2222
import io.rsocket.DuplexConnection;
2323
import io.rsocket.RSocketErrorException;
24-
import io.rsocket.exceptions.ConnectionCloseException;
2524
import io.rsocket.exceptions.ConnectionErrorException;
25+
import io.rsocket.frame.ErrorFrameCodec;
2626
import io.rsocket.frame.FrameHeaderCodec;
2727
import io.rsocket.internal.UnboundedProcessor;
2828
import java.net.SocketAddress;
@@ -50,8 +50,8 @@ public class ResumableDuplexConnection extends Flux<ByteBuf>
5050
final ResumableFramesStore resumableFramesStore;
5151

5252
final UnboundedProcessor savableFramesSender;
53-
final Disposable framesSaverDisposable;
54-
final Sinks.Empty<Void> onClose;
53+
final Sinks.Empty<Void> onQueueClose;
54+
final Sinks.Empty<Void> onLastConnectionClose;
5555
final SocketAddress remoteAddress;
5656
final Sinks.Many<Integer> onConnectionClosedSink;
5757

@@ -79,11 +79,13 @@ public ResumableDuplexConnection(
7979
this.session = session.toString(CharsetUtil.UTF_8);
8080
this.onConnectionClosedSink = Sinks.unsafe().many().unicast().onBackpressureBuffer();
8181
this.resumableFramesStore = resumableFramesStore;
82-
this.savableFramesSender = new UnboundedProcessor();
83-
this.framesSaverDisposable = resumableFramesStore.saveFrames(savableFramesSender).subscribe();
84-
this.onClose = Sinks.empty();
82+
this.onQueueClose = Sinks.unsafe().empty();
83+
this.onLastConnectionClose = Sinks.unsafe().empty();
84+
this.savableFramesSender = new UnboundedProcessor(onQueueClose::tryEmitEmpty, __ -> {});
8585
this.remoteAddress = initialConnection.remoteAddress();
8686

87+
resumableFramesStore.saveFrames(savableFramesSender).subscribe();
88+
8789
ACTIVE_CONNECTION.lazySet(this, initialConnection);
8890
}
8991

@@ -120,10 +122,12 @@ void initConnection(DuplexConnection nextConnection) {
120122
.resumeStream()
121123
.subscribe(
122124
f -> nextConnection.sendFrame(FrameHeaderCodec.streamId(f), f),
123-
t -> sendErrorAndClose(new ConnectionErrorException(t.getMessage())),
125+
t ->
126+
nextConnection.sendErrorAndClose(
127+
new ConnectionErrorException(t.getMessage(), t)),
124128
() ->
125-
sendErrorAndClose(
126-
new ConnectionCloseException("Connection Closed Unexpectedly")));
129+
nextConnection.sendErrorAndClose(
130+
new ConnectionErrorException("Connection Closed Unexpectedly")));
127131
nextConnection.receive().subscribe(frameReceivingSubscriber);
128132
nextConnection
129133
.onClose()
@@ -161,9 +165,9 @@ public void disconnect() {
161165
@Override
162166
public void sendFrame(int streamId, ByteBuf frame) {
163167
if (streamId == 0) {
164-
savableFramesSender.onNextPrioritized(frame);
168+
savableFramesSender.tryEmitPrioritized(frame);
165169
} else {
166-
savableFramesSender.onNext(frame);
170+
savableFramesSender.tryEmitNormal(frame);
167171
}
168172
}
169173

@@ -184,32 +188,25 @@ public void sendErrorAndClose(RSocketErrorException rSocketErrorException) {
184188
return;
185189
}
186190

187-
activeConnection.sendErrorAndClose(rSocketErrorException);
191+
savableFramesSender.tryEmitFinal(
192+
ErrorFrameCodec.encode(activeConnection.alloc(), 0, rSocketErrorException));
193+
188194
activeConnection
189195
.onClose()
190196
.subscribe(
191197
null,
192198
t -> {
193-
framesSaverDisposable.dispose();
194-
activeReceivingSubscriber.dispose();
195-
savableFramesSender.onComplete();
196-
savableFramesSender.cancel();
197199
onConnectionClosedSink.tryEmitComplete();
198-
199-
onClose.tryEmitError(t);
200+
onLastConnectionClose.tryEmitEmpty();
200201
},
201202
() -> {
202-
framesSaverDisposable.dispose();
203-
activeReceivingSubscriber.dispose();
204-
savableFramesSender.onComplete();
205-
savableFramesSender.cancel();
206203
onConnectionClosedSink.tryEmitComplete();
207204

208205
final Throwable cause = rSocketErrorException.getCause();
209206
if (cause == null) {
210-
onClose.tryEmitEmpty();
207+
onLastConnectionClose.tryEmitEmpty();
211208
} else {
212-
onClose.tryEmitError(cause);
209+
onLastConnectionClose.tryEmitError(cause);
213210
}
214211
});
215212
}
@@ -226,50 +223,62 @@ public ByteBufAllocator alloc() {
226223

227224
@Override
228225
public Mono<Void> onClose() {
229-
return onClose.asMono();
226+
return Mono.whenDelayError(
227+
onQueueClose.asMono().log(side + "_queue"),
228+
resumableFramesStore.onClose().log(side + "_frame_store"),
229+
onLastConnectionClose.asMono().log(side + "_last_connection"));
230230
}
231231

232232
@Override
233233
public void dispose() {
234-
dispose(null);
235-
}
236-
237-
void dispose(@Nullable Throwable e) {
234+
logger.info(side + "_disposing");
238235
final DuplexConnection activeConnection =
239236
ACTIVE_CONNECTION.getAndSet(this, DisposedConnection.INSTANCE);
240237
if (activeConnection == DisposedConnection.INSTANCE) {
241238
return;
242239
}
243-
244-
if (activeConnection != null) {
245-
activeConnection.dispose();
246-
}
247-
248-
if (logger.isDebugEnabled()) {
249-
logger.debug(
250-
"Side[{}]|Session[{}]|DuplexConnection[{}]. Disposing...",
251-
side,
252-
session,
253-
connectionIndex);
254-
}
255-
256-
framesSaverDisposable.dispose();
257-
activeReceivingSubscriber.dispose();
240+
logger.info(side + "_disposing2");
258241
savableFramesSender.onComplete();
259-
savableFramesSender.cancel();
260-
onConnectionClosedSink.tryEmitComplete();
242+
activeConnection
243+
.onClose()
244+
.subscribe(
245+
null,
246+
t -> {
247+
onConnectionClosedSink.tryEmitComplete();
248+
onLastConnectionClose.tryEmitEmpty();
249+
},
250+
() -> {
251+
onConnectionClosedSink.tryEmitComplete();
252+
onLastConnectionClose.tryEmitEmpty();
253+
});
254+
}
261255

262-
if (e != null) {
263-
onClose.tryEmitError(e);
264-
} else {
265-
onClose.tryEmitEmpty();
256+
void dispose(DuplexConnection nextConnection, @Nullable Throwable e) {
257+
final DuplexConnection activeConnection =
258+
ACTIVE_CONNECTION.getAndSet(this, DisposedConnection.INSTANCE);
259+
if (activeConnection == DisposedConnection.INSTANCE) {
260+
return;
266261
}
262+
savableFramesSender.onComplete();
263+
nextConnection
264+
.onClose()
265+
.subscribe(
266+
null,
267+
t -> {
268+
onConnectionClosedSink.tryEmitComplete();
269+
onLastConnectionClose.tryEmitEmpty();
270+
},
271+
() -> {
272+
onConnectionClosedSink.tryEmitComplete();
273+
onLastConnectionClose.tryEmitEmpty();
274+
});
267275
}
268276

269277
@Override
270278
@SuppressWarnings("ConstantConditions")
271279
public boolean isDisposed() {
272-
return onClose.scan(Scannable.Attr.TERMINATED) || onClose.scan(Scannable.Attr.CANCELLED);
280+
return onQueueClose.scan(Scannable.Attr.TERMINATED)
281+
|| onQueueClose.scan(Scannable.Attr.CANCELLED);
273282
}
274283

275284
@Override
@@ -280,6 +289,7 @@ public SocketAddress remoteAddress() {
280289
@Override
281290
public void request(long n) {
282291
if (state == 1 && STATE.compareAndSet(this, 1, 2)) {
292+
// happens for the very first time with the initial connection
283293
initConnection(this.activeConnection);
284294
}
285295
}

‎rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/gracefulshutdown/ClientStreamingToServer.java

+24-17
Original file line numberDiff line numberDiff line change
@@ -29,35 +29,39 @@
2929
import org.slf4j.Logger;
3030
import org.slf4j.LoggerFactory;
3131
import reactor.core.publisher.Flux;
32-
import reactor.core.publisher.Sinks;
32+
import reactor.core.publisher.Mono;
3333

3434
public final class ClientStreamingToServer {
3535

3636
private static final Logger logger = LoggerFactory.getLogger(ClientStreamingToServer.class);
3737

3838
public static void main(String[] args) throws InterruptedException {
3939
RSocketServer.create(
40-
SocketAcceptor.with(
41-
new RSocket() {
42-
final Sinks.Empty<Void> onGracefulShutdownSink = Sinks.unsafe().empty();
40+
(setup, sendingSocket) -> {
41+
sendingSocket.disposeGracefully();
4342

44-
@Override
45-
public Flux<Payload> requestStream(Payload payload) {
46-
return Flux.interval(Duration.ofMillis(100))
47-
.takeUntilOther(onGracefulShutdownSink.asMono())
48-
.map(aLong -> DefaultPayload.create("Interval: " + aLong));
49-
}
43+
return Mono.just(
44+
new RSocket() {
45+
@Override
46+
public Flux<Payload> requestStream(Payload payload) {
47+
return Flux.interval(Duration.ofMillis(100))
48+
.map(aLong -> DefaultPayload.create("Interval: " + aLong));
49+
}
5050

51-
@Override
52-
public void disposeGracefully() {
53-
// can be intercepted there
54-
// onGracefulShutdownSink.tryEmitEmpty();
55-
}
56-
}))
51+
@Override
52+
public void disposeGracefully() {}
53+
});
54+
})
5755
.bindNow(TcpServerTransport.create("localhost", 7000));
5856

5957
RSocket socket =
6058
RSocketConnector.create()
59+
.acceptor(
60+
SocketAcceptor.with(
61+
new RSocket() {
62+
@Override
63+
public void disposeGracefully() {}
64+
}))
6165
.setupPayload(DefaultPayload.create("test", "test"))
6266
.connect(TcpClientTransport.create("localhost", 7000))
6367
.block();
@@ -73,11 +77,14 @@ public void disposeGracefully() {
7377
logger.debug(msg);
7478
counter.incrementAndGet();
7579
})
76-
.take(100)
7780
.subscribe();
7881

7982
logger.debug("dispose gracefully");
8083
socket.disposeGracefully();
84+
//
85+
// Mono.delay(Duration.ofSeconds(10))
86+
// .doFinally((__) -> socket.dispose())
87+
// .subscribe();
8188

8289
socket.onClose().block();
8390

0 commit comments

Comments
 (0)