/** * Copyright 2016 Netflix, Inc. * * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in * compliance with the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software distributed under the License is * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See * the License for the specific language governing permissions and limitations under the License. */ package io.reactivex.internal.subscriptions; import org.reactivestreams.Subscriber; /** * A subscription that signals a single value eventually. *
* Note that the class leaks all methods of {@link java.util.concurrent.atomic.AtomicLong}. * Use {@link #complete(Object)} to signal the single value. *
* The this atomic integer stores a bitfield: Make sure this is called exactly once.
* @param v the value to signal, not null (not validated)
*/
public final void complete(T v) {
int state = get();
for (;;) {
if (state == FUSED_EMPTY) {
value = v;
lazySet(FUSED_READY);
Subscriber super T> a = actual;
a.onNext(v);
if (get() != CANCELLED) {
a.onComplete();
}
return;
}
// if state is >= CANCELLED or bit zero is set (*_HAS_VALUE) case, return
if ((state & ~HAS_REQUEST_NO_VALUE) != 0) {
return;
}
if (state == HAS_REQUEST_NO_VALUE) {
lazySet(HAS_REQUEST_HAS_VALUE);
Subscriber super T> a = actual;
a.onNext(v);
if (get() != CANCELLED) {
a.onComplete();
}
return;
}
value = v;
if (compareAndSet(NO_REQUEST_NO_VALUE, NO_REQUEST_HAS_VALUE)) {
return;
}
state = get();
if (state == CANCELLED) {
value = null;
return;
}
}
}
@Override
public final int requestFusion(int mode) {
if ((mode & ASYNC) != 0) {
lazySet(FUSED_EMPTY);
return ASYNC;
}
return NONE;
}
@Override
public final T poll() {
if (get() == FUSED_READY) {
lazySet(FUSED_CONSUMED);
T v = value;
value = null;
return v;
}
return null;
}
@Override
public final boolean isEmpty() {
return get() != FUSED_READY;
}
@Override
public final void clear() {
lazySet(FUSED_CONSUMED);
value = null;
}
@Override
public void cancel() {
set(CANCELLED);
value = null;
}
/**
* Returns true if this Subscription has been cancelled.
* @return true if this Subscription has been cancelled
*/
public final boolean isCancelled() {
return get() == CANCELLED;
}
/**
* Atomically sets a cancelled state and returns true if
* the current thread did it successfully.
* @return true if the current thread cancelled
*/
public final boolean tryCancel() {
return getAndSet(CANCELLED) != CANCELLED;
}
}
* bit 0: indicates that there is a value available
* bit 1: indicates that there was a request made
* bit 2: indicates there was a cancellation, exclusively set
* bit 3: indicates in fusion mode but no value yet, exclusively set
* bit 4: indicates in fusion mode and value is available, exclusively set
* bit 5: indicates in fusion mode and value has been consumed, exclusively set
* Where exclusively set means any other bits are 0 when that bit is set.
* @param