File tree Expand file tree Collapse file tree 1 file changed +9
-2
lines changed
src/main/java/rx/internal/operators Expand file tree Collapse file tree 1 file changed +9
-2
lines changed Original file line number Diff line number Diff line change 2121
2222import rx .*;
2323import rx .Completable .OnSubscribe ;
24- import rx .internal .util .unsafe .MpscLinkedQueue ;
24+ import rx .internal .util .atomic .MpscLinkedAtomicQueue ;
25+ import rx .internal .util .unsafe .*;
2526import rx .subscriptions .CompositeSubscription ;
2627
2728public final class CompletableOnSubscribeMergeDelayErrorIterable implements OnSubscribe {
@@ -53,7 +54,13 @@ public void call(final CompletableSubscriber s) {
5354
5455 final AtomicInteger wip = new AtomicInteger (1 );
5556
56- final Queue <Throwable > queue = new MpscLinkedQueue <Throwable >();
57+ final Queue <Throwable > queue ;
58+
59+ if (UnsafeAccess .isUnsafeAvailable ()) {
60+ queue = new MpscLinkedQueue <Throwable >();
61+ } else {
62+ queue = new MpscLinkedAtomicQueue <Throwable >();
63+ }
5764
5865 for (;;) {
5966 if (set .isUnsubscribed ()) {
You can’t perform that action at this time.
0 commit comments