Commit9334a56

committed

Don't automatically start requesting on subscribe

Motivation:HandlerSubscriber automatically starts requests when callingonSubscribe.But when a publisher subscribes, RxJava automatically requests theSubscription if there are some pending requests.When such race condition happen, we end up with 2 threads (calling andevent loop) competing for requesting the Subscription. This results isout of order published messages.Modification:Make HandlerSubscriber#onSubscribe noop and actually delay it to afterPublisher has subscribed.Result:No more out of order messages.

1 parent8c27b67 commit9334a56Copy full SHA for 9334a56

File tree

1 file changed

+12

-0

lines changed

client/src/main/java/org/asynchttpclient/netty/request/body
- NettyReactiveStreamsBody.java

1 file changed

+12

-0

lines changed

`‎client/src/main/java/org/asynchttpclient/netty/request/body/NettyReactiveStreamsBody.java‎`

Lines changed: 12 additions & 0 deletions

Original file line number	Diff line number	Diff line change
`@@ -58,6 +58,7 @@ public void write(Channel channel, NettyResponseFuture<?> future) throws IOExcep`
`58`	`58`	`NettySubscribersubscriber =newNettySubscriber(channel,future);`
`59`	`59`	`channel.pipeline().addLast(NAME_IN_CHANNEL_PIPELINE,subscriber);`
`60`	`60`	`publisher.subscribe(newSubscriberAdapter(subscriber));`
	`61`	`+subscriber.delayedStart();`
`61`	`62`	`}`
`62`	`63`	`}`
`63`	`64`
`@@ -108,6 +109,17 @@ protected void complete() {`
`108`	`109`	`.addListener(future ->removeFromPipeline()));`
`109`	`110`	`}`
`110`	`111`
	`112`	`+privatevolatileSubscriptiondeferredSubscription;`
	`113`	`+`
	`114`	`+@Override`
	`115`	`+publicvoidonSubscribe(Subscriptionsubscription) {`
	`116`	`+deferredSubscription =subscription;`
	`117`	`+}`
	`118`	`+`
	`119`	`+publicvoiddelayedStart() {`
	`120`	`+super.onSubscribe(deferredSubscription);`
	`121`	`+}`
	`122`	`+`
`111`	`123`	`@Override`
`112`	`124`	`protectedvoiderror(Throwableerror) {`
`113`	`125`	`if (error ==null)`

0 commit comments

Comments

(0)

Movatterモバイル変換

Navigation Menu

Search code, repositories, users, issues, pull requests...

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Commit9334a56

File tree

1 file changed

1 file changed

`‎client/src/main/java/org/asynchttpclient/netty/request/body/NettyReactiveStreamsBody.java‎`

0 commit comments