/*
 * 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 org.runestar.client.game.raw.base;

import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;

import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.function.Consumer;

public final class PublishRelay<T> extends Observable<T> implements Consumer<T> {

    private static final AtomicReferenceFieldUpdater<PublishRelay, PublishDisposable[]> SUBSCRIBERS =
            AtomicReferenceFieldUpdater.newUpdater(PublishRelay.class, PublishDisposable[].class, "subscribers");

    /** An empty subscribers array to avoid allocating it all the time. */
    @SuppressWarnings("rawtypes")
    private static final PublishDisposable[] EMPTY = new PublishDisposable[0];

    /** The array of currently subscribed subscribers. */
    private volatile PublishDisposable<T>[] subscribers;

    /**
     * Constructs a PublishRelay.
     */
    @SuppressWarnings("unchecked") PublishRelay() {
        subscribers = EMPTY;
    }

    @Override
    public void subscribeActual(Observer<? super T> t) {
        PublishDisposable<T> ps = new PublishDisposable<>(t, this);
        t.onSubscribe(ps);
        add(ps);
        // if cancellation happened while a successful add, the remove() didn't work
        // so we need to do it again
        if (ps.isDisposed()) {
            remove(ps);
        }
    }

    /**
     * Tries to add the given subscriber to the subscribers array atomically
     * or returns false if the subject has terminated.
     * @param ps the subscriber to add
     */
    private void add(PublishDisposable<T> ps) {
        for (;;) {
            PublishDisposable<T>[] a = subscribers;
            int n = a.length;
            @SuppressWarnings("unchecked")
            PublishDisposable<T>[] b = new PublishDisposable[n + 1];
            System.arraycopy(a, 0, b, 0, n);
            b[n] = ps;

            if (SUBSCRIBERS.compareAndSet(this, a, b)) {
                return;
            }
        }
    }

    /**
     * Atomically removes the given subscriber if it is subscribed to the subject.
     * @param ps the subject to remove
     */
    @SuppressWarnings({"unchecked", "WeakerAccess"})
    void remove(PublishDisposable<T> ps) {
        for (;;) {
            PublishDisposable<T>[] a = subscribers;
            if (a == EMPTY) {
                return;
            }

            int n = a.length;
            int j = -1;
            for (int i = 0; i < n; i++) {
                if (a[i] == ps) {
                    j = i;
                    break;
                }
            }

            if (j < 0) {
                return;
            }

            PublishDisposable<T>[] b;

            if (n == 1) {
                b = EMPTY;
            } else {
                b = new PublishDisposable[n - 1];
                System.arraycopy(a, 0, b, 0, j);
                System.arraycopy(a, j + 1, b, j, n - j - 1);
            }
            if (SUBSCRIBERS.compareAndSet(this, a, b)) {
                return;
            }
        }
    }

    @Override
    public void accept(T value) {
        if (value == null) throw new NullPointerException("value == null");
        for (PublishDisposable<T> s : subscribers) {
            s.onNext(value);
        }
    }

    public boolean hasObservers() {
        return subscribers.length != 0;
    }

    /**
     * Wraps the actual subscriber, tracks its requests and makes cancellation
     * to remove itself from the current subscribers array.
     *
     * @param <T> the value type
     */
    static final class PublishDisposable<T> extends AtomicBoolean implements Disposable {

        /** The actual subscriber. */
        final Observer<? super T> actual;
        /** The subject state. */
        final PublishRelay<T> parent;

        /**
         * Constructs a PublishSubscriber, wraps the actual subscriber and the state.
         * @param actual the actual subscriber
         * @param parent the parent PublishProcessor
         */
        PublishDisposable(Observer<? super T> actual, PublishRelay<T> parent) {
            this.actual = actual;
            this.parent = parent;
        }

        void onNext(T t) {
            if (!get()) {
                actual.onNext(t);
            }
        }

        @Override
        public void dispose() {
            if (compareAndSet(false, true)) {
                parent.remove(this);
            }
        }

        @Override
        public boolean isDisposed() {
            return get();
        }
    }
}

@@ -1,10 +1,11 @@
 /*
+*
 %0A * Copy
@@ -14,25 +14,45 @@
 ght 
-2016 Netflix, Inc
+(c) 2016-present, RxJava Contributors
 .%0A *
@@ -746,24 +746,50 @@
 isposable;%0A%0A
+import java.util.Objects;%0A
 import java.
@@ -1336,32 +1336,24 @@
 types%22)%0A    
-private 
 static final
@@ -1470,24 +1470,16 @@
  */%0A    
-private 
 volatile
@@ -1600,16 +1600,20 @@
 hecked%22)
+%0A   
  Publish
@@ -1673,29 +1673,32 @@
 erride%0A    p
-ublic
+rotected
  void subscr
@@ -1788,16 +1788,17 @@
 posable%3C
+T
 %3E(t, thi
@@ -2058,20 +2058,12 @@
   * 
-Tries to a
+A
 dd
+s
  the
@@ -2119,62 +2119,8 @@
 ally
-%0A     * or returns false if the subject has terminated
 .%0A  
@@ -2168,24 +2168,16 @@
  */%0A    
-private 
 void add
@@ -2758,17 +2758,16 @@
 arnings(
-%7B
 %22uncheck
@@ -2773,25 +2773,8 @@
 ked%22
-, %22WeakerAccess%22%7D
 )%0A  
@@ -3694,73 +3694,36 @@
     
-if (value == null) throw new NullPointerException(%22value == null%22
+Objects.requireNonNull(value
 );%0A 
@@ -3886,20 +3886,17 @@
 bers
-.length != 0
+ != EMPTY
 ;%0A  
@@ -4258,23 +4258,28 @@
 uper T%3E 
-actual;
+downstream;%0A
 %0A       
@@ -4651,22 +4651,26 @@
    this.
-actual
+downstream
  = actua
@@ -4721,24 +4721,31 @@
  %7D%0A%0A        
+public 
 void onNext(
@@ -4793,22 +4793,26 @@
         
-actual
+downstream
 .onNext(
@@ -5097,9 +5097,8 @@
 %0A    %7D%0A%7D
-%0A

Update PublishRelay