<commit_before>/**
 * Copyright 2013 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 rx.subjects;

import static org.junit.Assert.*;
import static org.mockito.Matchers.*;
import static org.mockito.Mockito.*;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

import junit.framework.Assert;

import org.junit.Test;
import org.mockito.InOrder;
import org.mockito.Mockito;

import rx.Notification;
import rx.Observable;
import rx.Observer;
import rx.Subscription;
import rx.operators.AtomicObservableSubscription;
import rx.subscriptions.Subscriptions;
import rx.util.functions.Action1;
import rx.util.functions.Func0;
import rx.util.functions.Func1;

/**
 * Subject that publishes a single event to each {@link Observer} that has subscribed. 
 * <p>
 * Example usage:
 * <p>
 * <pre> {@code
 
  PublishSubject<Object> subject = PublishSubject.create();
  // observer1 will receive all onNext events
  subject.subscribe(observer1);
  subject.onNext("one");
  subject.onNext("two");
  // observer2 will only receive "three" and onCompleted
  subject.subscribe(observer2);
  subject.onNext("three");
  subject.onCompleted();
 
  } </pre>
 * 
 * @param <T>
 */
public class PublishSubject<T> extends Subject<T, T> {
    public static <T> PublishSubject<T> create() {
        final ConcurrentHashMap<Subscription, Observer<T>> observers = new ConcurrentHashMap<Subscription, Observer<T>>();
        final AtomicReference<Notification<T>> terminalState = new AtomicReference<Notification<T>>();
        
        Func1<Observer<T>, Subscription> onSubscribe = new Func1<Observer<T>, Subscription>() {
            @Override
            public Subscription call(Observer<T> observer) {
                // first check if terminal state exist
                Subscription s = checkTerminalState(observer);
                if(s != null) return s;
                
                final AtomicObservableSubscription subscription = new AtomicObservableSubscription();

                subscription.wrap(new Subscription() {
                    @Override
                    public void unsubscribe() {
                        // on unsubscribe remove it from the map of outbound observers to notify
                        observers.remove(subscription);
                    }
                });

                // on subscribe add it to the map of outbound observers to notify
                observers.put(subscription, observer);
                
                // check terminal state again 
                s = checkTerminalState(observer);
                if(s != null) return s;
                
                /**
                 * NOTE: There is a race condition here.
                 * 
                 * 1) terminal state gets set in onError or onCompleted
                 * 2) observers.put adds a new observer
                 * 3) checkTerminalState emits onError/onCompleted
                 * 4) onError or onCompleted also emits onError/onCompleted since it was adds to observers
                 * 
                 * Thus the terminal state could end up being sent twice.
                 * 
                 * I'm going to leave this for now as AtomicObserver will protect against this
                 * and I'd rather not add blocking synchronization in here unless the above race condition
                 * truly is an issue.
                 */
                
                return subscription;
            }
            
            private Subscription checkTerminalState(Observer<T> observer) {
                Notification<T> n = terminalState.get();
                if (n != null) {
                    // we are terminated to immediately emit and don't continue with subscription
                    if (n.isOnCompleted()) {
                        observer.onCompleted();
                    } else {
                        observer.onError(n.getException());
                    }
                    return Subscriptions.empty();
                } else {
                    return null;
                }
            }
        };

        return new PublishSubject<T>(onSubscribe, observers, terminalState);
    }

    private final ConcurrentHashMap<Subscription, Observer<T>> observers;
    private final AtomicReference<Notification<T>> terminalState;

    protected PublishSubject(Func1<Observer<T>, Subscription> onSubscribe, ConcurrentHashMap<Subscription, Observer<T>> observers, AtomicReference<Notification<T>> terminalState) {
        super(onSubscribe);
        this.observers = observers;
        this.terminalState = terminalState;
    }

    @Override
    public void onCompleted() {
        terminalState.set(new Notification<T>());
        for (Observer<T> observer : snapshotOfValues()) {
            observer.onCompleted();
        }
        observers.clear();
    }

    @Override
    public void onError(Exception e) {
        terminalState.set(new Notification<T>(e));
        for (Observer<T> observer : snapshotOfValues()) {
            observer.onError(e);
        }
        observers.clear();
    }

    @Override
    public void onNext(T args) {
        for (Observer<T> observer : snapshotOfValues()) {<commit_msg>Synchronize terminalState and subscription logic<commit_after>/**
 * Copyright 2013 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 rx.subjects;

import static org.junit.Assert.*;
import static org.mockito.Matchers.*;
import static org.mockito.Mockito.*;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

import junit.framework.Assert;

import org.junit.Test;
import org.mockito.InOrder;
import org.mockito.Mockito;

import rx.Notification;
import rx.Observable;
import rx.Observer;
import rx.Subscription;
import rx.operators.AtomicObservableSubscription;
import rx.subscriptions.Subscriptions;
import rx.util.functions.Action1;
import rx.util.functions.Func0;
import rx.util.functions.Func1;

/**
 * Subject that publishes a single event to each {@link Observer} that has subscribed. 
 * <p>
 * Example usage:
 * <p>
 * <pre> {@code
 
  PublishSubject<Object> subject = PublishSubject.create();
  // observer1 will receive all onNext events
  subject.subscribe(observer1);
  subject.onNext("one");
  subject.onNext("two");
  // observer2 will only receive "three" and onCompleted
  subject.subscribe(observer2);
  subject.onNext("three");
  subject.onCompleted();
 
  } </pre>
 * 
 * @param <T>
 */
public class PublishSubject<T> extends Subject<T, T> {
    public static <T> PublishSubject<T> create() {
        final ConcurrentHashMap<Subscription, Observer<T>> observers = new ConcurrentHashMap<Subscription, Observer<T>>();
        final AtomicReference<Notification<T>> terminalState = new AtomicReference<Notification<T>>();
        
        Func1<Observer<T>, Subscription> onSubscribe = new Func1<Observer<T>, Subscription>() {
            @Override
            public Subscription call(Observer<T> observer) {
                // shortcut check if terminal state exists already
                Subscription s = checkTerminalState(observer);
                if(s != null) return s;
                
                final AtomicObservableSubscription subscription = new AtomicObservableSubscription();

                subscription.wrap(new Subscription() {
                    @Override
                    public void unsubscribe() {
                        // on unsubscribe remove it from the map of outbound observers to notify
                        observers.remove(subscription);
                    }
                });

                /**
                 * NOTE: We are synchronizing to avoid a race condition between terminalState being set and 
                 * a new observer being added to observers.
                 * 
                 * The synchronization only occurs on subscription and terminal states, it does not affect onNext calls
                 * so a high-volume hot-observable will not pay this cost for emitting data.
                 * 
                 * Due to the restricted impact of blocking synchronization here I have not pursued more complicated 
                 * approaches to try and stay completely non-blocking.
                 */
                synchronized (terminalState) {
                    // check terminal state again 
                    s = checkTerminalState(observer);
                    if (s != null)
                        return s;
                    
                    // on subscribe add it to the map of outbound observers to notify
                    observers.put(subscription, observer);

                    return subscription;
                }
            }
            
            private Subscription checkTerminalState(Observer<T> observer) {
                Notification<T> n = terminalState.get();
                if (n != null) {
                    // we are terminated to immediately emit and don't continue with subscription
                    if (n.isOnCompleted()) {
                        observer.onCompleted();
                    } else {
                        observer.onError(n.getException());
                    }
                    return Subscriptions.empty();
                } else {
                    return null;
                }
            }
        };

        return new PublishSubject<T>(onSubscribe, observers, terminalState);
    }

    private final ConcurrentHashMap<Subscription, Observer<T>> observers;
    private final AtomicReference<Notification<T>> terminalState;

    protected PublishSubject(Func1<Observer<T>, Subscription> onSubscribe, ConcurrentHashMap<Subscription, Observer<T>> observers, AtomicReference<Notification<T>> terminalState) {
        super(onSubscribe);
        this.observers = observers;
        this.terminalState = terminalState;
    }

    @Override
    public void onCompleted() {
        /**
         * Synchronizing despite terminalState being an AtomicReference because of multi-step logic in subscription.
         * Why use AtomicReference then? Convenient for passing around a mutable reference holder between the 
         * onSubscribe function and PublishSubject instance... and it's a "better volatile" for the shortcut codepath.
         */
        synchronized (terminalState) {
            terminalState.set(new Notification<T>());
        }
        for (Observer<T> observer : snapshotOfValues()) {
            observer.onCompleted();
        }
        observers.clear();
    }

    @Override
    public void onError(Exception e) {
        /**
         * Synchronizing despite terminalState being an AtomicReference because of multi-step logic in subscription.
         * Why use AtomicReference then? Convenient for passing around a mutable reference holder between the 
         * onSubscribe function and PublishSubject instance... and it's a "better volatile" for the shortcut codepath.
         */
        synchronized (terminalState) {
            terminalState.set(new Notification<T>(e));
        }
        for (Observer<T> observer : snapshotOfValues()) {
            observer.onError(e);
        }
        observers.clear();
    }

    @Override
    public void onNext(T args) {
        for (Observer<T> observer : snapshotOfValues()) {
{'fraction_non_alphanumeric': 0.08456344221105527, 'fraction_numerical': 0.00314070351758794, 'mean_word_length': 2.2188526661612333, 'pattern_counts': {'":': 0, '<': 85, '<?xml version=': 0, '>': 85, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4329967', 'n_tokens_mistral': 3163, 'n_tokens_neox': 3004, 'n_words': 978}