@@ -117,7 +117,7 @@ void connect(Flow.Subscriber<? super List<ByteBuffer>> downReader,
117117 // Connect the read sink first. That's the left-hand side
118118 // downstream subscriber from the HttpConnection (or more
119119 // accurately, the SSLSubscriberWrapper that will wrap it
120- // when SSLTube::connectFlows is called.
120+ // when SSLTube::connectFlows is called) .
121121 reader .subscribe (downReader );
122122
123123 // Connect the right hand side tube (the socket tube).
@@ -182,7 +182,7 @@ public boolean isFinished() {
182182 private volatile Flow .Subscription readSubscription ;
183183
184184 // The DelegateWrapper wraps a subscribed {@code Flow.Subscriber} and
185- // tracks the subscriber's state. In particular it makes sure that
185+ // tracks the subscriber's state. In particular, it makes sure that
186186 // onComplete/onError are not called before onSubscribed.
187187 static final class DelegateWrapper implements FlowTube .TubeSubscriber {
188188 private final FlowTube .TubeSubscriber delegate ;
@@ -293,7 +293,7 @@ public String toString() {
293293
294294 // Used to read data from the SSLTube.
295295 final class SSLSubscriberWrapper implements FlowTube .TubeSubscriber {
296- private AtomicReference <DelegateWrapper > pendingDelegate =
296+ private final AtomicReference <DelegateWrapper > pendingDelegate =
297297 new AtomicReference <>();
298298 private volatile DelegateWrapper subscribed ;
299299 private volatile boolean onCompleteReceived ;
@@ -346,15 +346,15 @@ void setDelegate(Flow.Subscriber<? super List<ByteBuffer>> delegate) {
346346 return ;
347347 }
348348 // sslDelegate field should have been initialized by the
349- // the time we reach here, as there can be no subscriber
349+ // time we reach here, as there can be no subscriber
350350 // until SSLTube is fully constructed.
351351 if (handleNow || !sslDelegate .resumeReader ()) {
352352 processPendingSubscriber ();
353353 }
354354 }
355355
356- // Can be called outside of the flow if an error has already been
357- // raise . Otherwise, must be called within the SSLFlowDelegate
356+ // Can be called outside the flow if an error has already been
357+ // raised . Otherwise, must be called within the SSLFlowDelegate
358358 // downstream reader flow.
359359 // If there is a subscription, and if there is a pending delegate,
360360 // calls dropSubscription() on the previous delegate (if any),
@@ -612,43 +612,73 @@ final class SSLSubscriptionWrapper implements Flow.Subscription {
612612 private volatile boolean cancelled ;
613613
614614 void setSubscription (Flow .Subscription sub ) {
615- long demand = writeDemand .get (); // FIXME: isn't it a racy way of passing the demand?
616- delegate = sub ;
617- if (debug .on ())
618- debug .log ("setSubscription: demand=%d, cancelled:%s" , demand , cancelled );
615+ long demand ;
616+ // Avoid race condition and requesting demand twice if
617+ // request() runs concurrently with setSubscription()
618+ boolean cancelled ;
619+ synchronized (this ) {
620+ demand = writeDemand .get ();
621+ delegate = sub ;
622+ cancelled = this .cancelled ;
623+ }
624+ if (debug .on ()) {
625+ debug .log ("setSubscription: demand=%d, cancelled:%s, new subscription %s" ,
626+ demand , cancelled , sub );
627+ }
619628
620629 if (cancelled )
621- delegate .cancel ();
630+ sub .cancel ();
622631 else if (demand > 0 )
623632 sub .request (demand );
624633 }
625634
626635 @ Override
627636 public void request (long n ) {
628- writeDemand .increase (n );
629- if (debug .on ()) debug .log ("request: n=%d" , n );
630- Flow .Subscription sub = delegate ;
637+ // Avoid race condition and requesting demand twice if
638+ // request() runs concurrently with setSubscription()
639+ Flow .Subscription sub ;
640+ long demanded ;
641+ synchronized (this ) {
642+ sub = delegate ;
643+ demanded = writeDemand .get ();
644+ writeDemand .increase (n );
645+ }
646+ if (debug .on ()) {
647+ debug .log ("request: n=%s to %s (%s already demanded)" ,
648+ n , sub , demanded );
649+ }
631650 if (sub != null && n > 0 ) {
651+ if (debug .on ()) debug .log ("requesting %s from %s" , n , sub );
632652 sub .request (n );
633653 }
634654 }
635655
636656 @ Override
637657 public void cancel () {
638- cancelled = true ;
639- if (delegate != null )
640- delegate .cancel ();
658+ Flow .Subscription sub ;
659+ synchronized (this ) {
660+ cancelled = true ;
661+ sub = delegate ;
662+ }
663+ if (debug .on ()) debug .log ("cancel: cancelling subscription: " + sub );
664+ if (sub != null ) sub .cancel ();
641665 }
642666 }
643667
644668 /* Subscriber - writing side */
645669 @ Override
646670 public void onSubscribe (Flow .Subscription subscription ) {
647671 Objects .requireNonNull (subscription );
648- Flow .Subscription x = writeSubscription .delegate ;
649- if (x != null )
650- x .cancel ();
672+ Flow .Subscription old ;
673+ synchronized (this ) {
674+ old = writeSubscription .delegate ;
675+ }
676+ if (old != null && old != subscription ) {
677+ if (debug .on ()) debug .log ("onSubscribe: cancelling old subscription: " + old );
678+ old .cancel ();
679+ }
651680
681+ if (debug .on ()) debug .log ("onSubscribe: new subscription: " + subscription );
652682 writeSubscription .setSubscription (subscription );
653683 }
654684
@@ -657,8 +687,10 @@ public void onNext(List<ByteBuffer> item) {
657687 Objects .requireNonNull (item );
658688 boolean decremented = writeDemand .tryDecrement ();
659689 assert decremented : "Unexpected writeDemand: " ;
660- if (debug .on ())
661- debug .log ("sending %d buffers to SSL flow delegate" , item .size ());
690+ if (debug .on ()) {
691+ debug .log ("sending %s buffers to SSL flow delegate (%s bytes)" ,
692+ item .size (), Utils .remaining (item ));
693+ }
662694 sslDelegate .upstreamWriter ().onNext (item );
663695 }
664696
0 commit comments