src/main/java/io/reactivex/rxjava4/internal/subscriptions/AsyncSubscription.java - 4.x - githubmirror/RxJava

src/main/java/io/reactivex/rxjava4/internal/subscriptions/AsyncSubscription.java
96 строк3 KB
David Karnok
4.x: Cleanup: Serial annotation, getFirst usage (#8180)
26 июн 2026, 00:06
26 июн 2026, 00:06f57ca39
/*
 * Copyright (c) 2016-present, RxJava Contributors.
 *
 * 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.rxjava4.internal.subscriptions;

import java.io.Serial;
import java.util.concurrent.atomic.*;

import static java.util.concurrent.Flow.*;

import io.reactivex.rxjava4.disposables.Disposable;
import io.reactivex.rxjava4.internal.disposables.DisposableHelper;

/**
 * A subscription implementation that arbitrates exactly one other Subscription and can
 * hold a single disposable resource.
 *
 * <p>All methods are thread-safe.
 */
public final class AsyncSubscription extends AtomicLong implements Subscription, Disposable {

    @Serial
    private static final long serialVersionUID = 7028635084060361255L;

    final AtomicReference<Subscription> actual;

    final AtomicReference<Disposable> resource;

    public AsyncSubscription() {
        resource = new AtomicReference<>();
        actual = new AtomicReference<>();
    }

    public AsyncSubscription(Disposable resource) {
        this();
        this.resource.lazySet(resource);
    }

    @Override
    public void request(long n) {
        SubscriptionHelper.deferredRequest(actual, this, n);
    }

    @Override
    public void cancel() {
        dispose();
    }

    @Override
    public void dispose() {
        SubscriptionHelper.cancel(actual);
        DisposableHelper.dispose(resource);
    }

    @Override
    public boolean isDisposed() {
        return actual.get() == SubscriptionHelper.CANCELLED;
    }

    /**
     * Sets a new resource and disposes the currently held resource.
     * @param r the new resource to set
     * @return false if this AsyncSubscription has been cancelled/disposed
     * @see #replaceResource(Disposable)
     */
    public boolean setResource(Disposable r) {
        return DisposableHelper.set(resource, r);
    }

    /**
     * Replaces the currently held resource with the given new one without disposing the old.
     * @param r the new resource to set
     * @return false if this AsyncSubscription has been cancelled/disposed
     */
    public boolean replaceResource(Disposable r) {
        return DisposableHelper.replace(resource, r);
    }

    /**
     * Sets the given subscription if there isn't any subscription held.
     * @param s the first and only subscription to set
     */
    public void setSubscription(Subscription s) {
        SubscriptionHelper.deferredSetOnce(actual, this, s);
    }
}