<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.operators;

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

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

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

import rx.Observable;
import rx.Observer;
import rx.Subscription;
import rx.concurrency.TestScheduler;
import rx.subscriptions.Subscriptions;
import rx.util.functions.Action0;
import rx.util.functions.Func1;


/**
 * This operation transforms an {@link Observable} sequence of {@link Observable} sequences into a single 
 * {@link Observable} sequence which only produces values from the most recently published {@link Observable} 
 * sequence in the sequence. 
 */
public final class OperationSwitch {

    /**
     * This function transforms an {@link Observable} sequence of {@link Observable} sequences into a single 
     * {@link Observable} sequence which produces values from the most recently published {@link Observable}.
     * 
     * @param sequences   The {@link Observable} sequence consisting of {@link Observable} sequences.
     * @return A {@link Func1} which does this transformation.
     */
    public static <T> Func1<Observer<T>, Subscription> switchDo(final Observable<Observable<T>> sequences) {
        return new Func1<Observer<T>, Subscription>() {
            @Override
            public Subscription call(Observer<T> observer) {
                return new Switch<T>(sequences).call(observer);
            }
        };
    }

    private static class Switch<T> implements Func1<Observer<T>, Subscription> {

        private final Observable<Observable<T>> sequences;

        public Switch(Observable<Observable<T>> sequences) {
            this.sequences = sequences;
        }

        @Override
        public Subscription call(Observer<T> observer) {
            return sequences.subscribe(new SwitchObserver<T>(observer));
        }
    }

    private static class SwitchObserver<T> implements Observer<Observable<T>> {

        private final AtomicReference<Subscription> subscription = new AtomicReference<Subscription>();
        
        private final Observer<T> observer;

        public SwitchObserver(Observer<T> observer) {
            this.observer = observer;
        }

        @Override
        public void onCompleted() {
            observer.onCompleted();
        }

        @Override
        public void onError(Exception e) {
            observer.onError(e);
        }

        @Override
        public void onNext(Observable<T> args) {
            synchronized (subscription) {
                Subscription previousSubscription = subscription.get();
                if (previousSubscription != null) {
                    previousSubscription.unsubscribe();
                }
                
                subscription.set(args.subscribe(new Observer<T>() {
                    @Override
                    public void onCompleted() {
                        // Do nothing.
                    }

                    @Override
                    public void onError(Exception e) {
                        // Do nothing.
                    }

                    @Override
                    public void onNext(T args) {
                        observer.onNext(args);
                    }
                }));
            }
        }
    }
    
    public static class UnitTest {

        private TestScheduler scheduler;
        private Observer<String> observer;

        @Before
        @SuppressWarnings("unchecked")
        public void before() {
            scheduler = new TestScheduler();
            observer = mock(Observer.class);
        }

        @Test
        public void testSwitchWithComplete() {
            Observable<Observable<String>> source = Observable.create(new Func1<Observer<Observable<String>>, Subscription>() {
                @Override
                public Subscription call(Observer<Observable<String>> observer) {
                    publishNext(observer, 50, Observable.create(new Func1<Observer<String>, Subscription>() {
                        @Override
                        public Subscription call(Observer<String> observer) {
                            publishNext(observer, 50, "one");
                            publishNext(observer, 100, "two");<commit_msg>Removed synchronized block as per RxJava guidelines (6.7).<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.operators;

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

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

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

import rx.Observable;
import rx.Observer;
import rx.Subscription;
import rx.concurrency.TestScheduler;
import rx.subscriptions.Subscriptions;
import rx.util.functions.Action0;
import rx.util.functions.Func1;


/**
 * This operation transforms an {@link Observable} sequence of {@link Observable} sequences into a single 
 * {@link Observable} sequence which only produces values from the most recently published {@link Observable} 
 * sequence in the sequence. 
 */
public final class OperationSwitch {

    /**
     * This function transforms an {@link Observable} sequence of {@link Observable} sequences into a single 
     * {@link Observable} sequence which produces values from the most recently published {@link Observable}.
     * 
     * @param sequences   The {@link Observable} sequence consisting of {@link Observable} sequences.
     * @return A {@link Func1} which does this transformation.
     */
    public static <T> Func1<Observer<T>, Subscription> switchDo(final Observable<Observable<T>> sequences) {
        return new Func1<Observer<T>, Subscription>() {
            @Override
            public Subscription call(Observer<T> observer) {
                return new Switch<T>(sequences).call(observer);
            }
        };
    }

    private static class Switch<T> implements Func1<Observer<T>, Subscription> {

        private final Observable<Observable<T>> sequences;

        public Switch(Observable<Observable<T>> sequences) {
            this.sequences = sequences;
        }

        @Override
        public Subscription call(Observer<T> observer) {
            return sequences.subscribe(new SwitchObserver<T>(observer));
        }
    }

    private static class SwitchObserver<T> implements Observer<Observable<T>> {

        private final AtomicReference<Subscription> subscription = new AtomicReference<Subscription>();
        
        private final Observer<T> observer;

        public SwitchObserver(Observer<T> observer) {
            this.observer = observer;
        }

        @Override
        public void onCompleted() {
            observer.onCompleted();
        }

        @Override
        public void onError(Exception e) {
            observer.onError(e);
        }

        @Override
        public void onNext(Observable<T> args) {
            Subscription previousSubscription = subscription.get();
            if (previousSubscription != null) {
                previousSubscription.unsubscribe();
            }
            
            subscription.set(args.subscribe(new Observer<T>() {
                @Override
                public void onCompleted() {
                    // Do nothing.
                }

                @Override
                public void onError(Exception e) {
                    // Do nothing.
                }

                @Override
                public void onNext(T args) {
                    observer.onNext(args);
                }
            }));
        }
    }
    
    public static class UnitTest {

        private TestScheduler scheduler;
        private Observer<String> observer;

        @Before
        @SuppressWarnings("unchecked")
        public void before() {
            scheduler = new TestScheduler();
            observer = mock(Observer.class);
        }

        @Test
        public void testSwitchWithComplete() {
            Observable<Observable<String>> source = Observable.create(new Func1<Observer<Observable<String>>, Subscription>() {
                @Override
                public Subscription call(Observer<Observable<String>> observer) {
                    publishNext(observer, 50, Observable.create(new Func1<Observer<String>, Subscription>() {
                        @Override
                        public Subscription call(Observer<String> observer) {
                            publishNext(observer, 50, "one");
                            publishNext(observer, 100, "two");
{'fraction_non_alphanumeric': 0.0878605646723066, 'fraction_numerical': 0.0048924676383651005, 'mean_word_length': 2.2276315789473684, 'pattern_counts': {'":': 0, '<': 79, '<?xml version=': 0, '>': 79, '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': '5829500', 'n_tokens_mistral': 2396, 'n_tokens_neox': 2270, 'n_words': 720}