<commit_before>/*
 * Copyright 2016 DiffPlug
 *
 * 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 com.diffplug.common.rx;

import static java.util.Objects.requireNonNull;

import java.util.Optional;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.Executor;
import java.util.concurrent.Future;
import java.util.function.Consumer;

import rx.Observable;
import rx.Scheduler;
import rx.Subscription;
import rx.schedulers.Schedulers;

import com.diffplug.common.base.Consumers;
import com.diffplug.common.base.DurianPlugins;
import com.diffplug.common.base.Errors;
import com.diffplug.common.util.concurrent.ListenableFuture;
import com.diffplug.common.util.concurrent.MoreExecutors;
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;

/**
 * Unifies the listener models of {@link rx.Observable RxJava's Observable} 
 * with <code><a href="https://code.google.com/p/guava-libraries/wiki/ListenableFutureExplained">Guava's ListenableFuture</a></code>
 * , and also adds tracing capabilities.
 * <p>
 * TL;DR
 * <pre>
 * // subscribe to values, termination, or both
 * Rx.subscribe(listenableOrObservable, val -> doSomething(val));                             // errors are passed to Errors.log()
 * Rx.subscribe(listenableOrObservable, Rx.onTerminate(optionalError -> maybeHandleError());  // values are ignored
 * Rx.subscribe(listenableOrObservable, Rx.onValueOrTerminate(val -> doSomething(val), optionalError -> maybeHandleError()));
 * // receive callbacks on a specific executor
 * Rx.on(someExecutor).subscribe(listenableOrObservable, val -> doSomething(val));
 * // call unsubscribe() on the subscription to cancel it 
 * rx.Subscription subscription = Rx.subscribe(listenableOrObservable, val -> doSomething);
 * </pre>
 * Long version: {@code Rx} implements both the {@link rx.Observer} and {@link com.diffplug.common.util.concurrent.FutureCallback}
 * interfaces by mapping them to two {@code Consumer}s:
 * <ul>
 * <li>{@code Consumer<T> onValue}</li>
 * <li>{@code Consumer<Optional<Throwable>> onTerminate}</li>
 * </ul>
 * Which are mapped as follows:
 * <ul>
 * <li>{@code Observable.onNext(T value)          -> onValue.accept(value)}</li>
 * <li>{@code Observable.onCompleted()            -> onTerminate.accept(Optional.empty())}</li>
 * <li>{@code Observable.onError(Throwable error) -> onTerminate.accept(Optional.of(error))}</li>
 * <li>{@code FutureCallback.onSuccess(T value)       -> onValue.accept(value); onTerminate.accept(Optional.empty());}</li>
 * <li>{@code FutureCallback.onError(Throwable error) -> onTerminate.accept(Optional.of(error))}</li>
 * </ul>
 * An instance of Rx is created by calling one of Rx's static creator methods:
 * <ul>
 * <li>{@link #onValue(Consumer)     onValue(Consumer&lt;T&gt;)}</li>
 * <li>{@link #onTerminate(Consumer) onTerminate(Consumer&lt;Optional&lt;Throwable&gt;&gt;)}</li>
 * <li>{@link #onFailure(Consumer)   onFailure(Consumer&lt;Throwable&gt;)}</li>
 * <li>{@link #onValueOnTerminate    onValueOrTerminate(Consumer&lt;T&gt;, Consumer&lt;Optional&lt;Throwable&gt;&gt;)}</li>
 * <li>{@link #onValueOnFailure      onValueOrFailure(Consumer&lt;T&gt;, Consumer&lt;Throwable&gt;)}</li>
 * </ul>
 * Once you have an instance of Rx, you can subscribe it using the normal RxJava or Guava calls:
 * <ul>
 * <li>{@code rxObservable.subscribe(Rx.onValue(val -> doSomething(val));}</li>
 * <li>{@code Futures.addCallback(listenableFuture, Rx.onValue(val -> doSomething(val));}</li>
 * </ul>
 * But the recommended way to subscribe is to use:
 * <ul>
 * <li>{@code Rx.subscribe(listenableOrObservable, Rx.onValue(val -> doSomething(val)));}</li>
 * <li>{@code Rx.subscribe(listenableOrObservable, val -> doSomething(val)); // automatically uses Rx.onValue()}</li>
 * </ul>
 * The advantage of this latter method is that it returns {@link rx.Subscription} instances
 * which allow you to unsubscribe from futures in the same manner as for observables.
 * <ul>
 * <li>{@code subscription = Rx.subscribe( ... )}</li>
 * </ul>
 * If you wish to receive callbacks on a specific thread, you can use:
 * <ul>
 * <li>{@code Rx.on(someExecutor).subscribe( ... )}</li>
 * </ul>
 * Because RxJava's Observables use {@link rx.Scheduler}s rather than {@link java.util.concurrent.Executor}s,
 * a Scheduler is automatically created using {@link rx.Schedulers#from}. If you'd like to specify the Scheduler manually, you can use {@link Rx#on(Executor, Scheduler)}
 * or you can create an executor which implements {@link Rx.HasRxExecutor}.
 * 
 * @see <a href="https://diffplug.github.io/durian-swt/javadoc/snapshot/com/diffplug/common/swt/SwtExec.html">SwtExec</a>
 */
public class Rx {
	/**
	 * Creates an Rx instance which will call the given consumer whenever a value is received.
	 * Any errors are sent to ErrorHandler.log().
	 */
	public static <T> RxListener<T> onValue(Consumer<T> onValue) {
		return new RxListener<T>(onValue, RxListener.logErrors);
	}<commit_msg>Rx's constructor is now private, to indicate that it is just a bunch of static methods.<commit_after>/*
 * Copyright 2016 DiffPlug
 *
 * 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 com.diffplug.common.rx;

import static java.util.Objects.requireNonNull;

import java.util.Optional;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.Executor;
import java.util.concurrent.Future;
import java.util.function.Consumer;

import rx.Observable;
import rx.Scheduler;
import rx.Subscription;
import rx.schedulers.Schedulers;

import com.diffplug.common.base.Consumers;
import com.diffplug.common.base.DurianPlugins;
import com.diffplug.common.base.Errors;
import com.diffplug.common.util.concurrent.ListenableFuture;
import com.diffplug.common.util.concurrent.MoreExecutors;
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;

/**
 * Unifies the listener models of {@link rx.Observable RxJava's Observable} 
 * with <code><a href="https://code.google.com/p/guava-libraries/wiki/ListenableFutureExplained">Guava's ListenableFuture</a></code>
 * , and also adds tracing capabilities.
 * <p>
 * TL;DR
 * <pre>
 * // subscribe to values, termination, or both
 * Rx.subscribe(listenableOrObservable, val -> doSomething(val));                             // errors are passed to Errors.log()
 * Rx.subscribe(listenableOrObservable, Rx.onTerminate(optionalError -> maybeHandleError());  // values are ignored
 * Rx.subscribe(listenableOrObservable, Rx.onValueOrTerminate(val -> doSomething(val), optionalError -> maybeHandleError()));
 * // receive callbacks on a specific executor
 * Rx.on(someExecutor).subscribe(listenableOrObservable, val -> doSomething(val));
 * // call unsubscribe() on the subscription to cancel it 
 * rx.Subscription subscription = Rx.subscribe(listenableOrObservable, val -> doSomething);
 * </pre>
 * Long version: {@code Rx} implements both the {@link rx.Observer} and {@link com.diffplug.common.util.concurrent.FutureCallback}
 * interfaces by mapping them to two {@code Consumer}s:
 * <ul>
 * <li>{@code Consumer<T> onValue}</li>
 * <li>{@code Consumer<Optional<Throwable>> onTerminate}</li>
 * </ul>
 * Which are mapped as follows:
 * <ul>
 * <li>{@code Observable.onNext(T value)          -> onValue.accept(value)}</li>
 * <li>{@code Observable.onCompleted()            -> onTerminate.accept(Optional.empty())}</li>
 * <li>{@code Observable.onError(Throwable error) -> onTerminate.accept(Optional.of(error))}</li>
 * <li>{@code FutureCallback.onSuccess(T value)       -> onValue.accept(value); onTerminate.accept(Optional.empty());}</li>
 * <li>{@code FutureCallback.onError(Throwable error) -> onTerminate.accept(Optional.of(error))}</li>
 * </ul>
 * An instance of Rx is created by calling one of Rx's static creator methods:
 * <ul>
 * <li>{@link #onValue(Consumer)     onValue(Consumer&lt;T&gt;)}</li>
 * <li>{@link #onTerminate(Consumer) onTerminate(Consumer&lt;Optional&lt;Throwable&gt;&gt;)}</li>
 * <li>{@link #onFailure(Consumer)   onFailure(Consumer&lt;Throwable&gt;)}</li>
 * <li>{@link #onValueOnTerminate    onValueOrTerminate(Consumer&lt;T&gt;, Consumer&lt;Optional&lt;Throwable&gt;&gt;)}</li>
 * <li>{@link #onValueOnFailure      onValueOrFailure(Consumer&lt;T&gt;, Consumer&lt;Throwable&gt;)}</li>
 * </ul>
 * Once you have an instance of Rx, you can subscribe it using the normal RxJava or Guava calls:
 * <ul>
 * <li>{@code rxObservable.subscribe(Rx.onValue(val -> doSomething(val));}</li>
 * <li>{@code Futures.addCallback(listenableFuture, Rx.onValue(val -> doSomething(val));}</li>
 * </ul>
 * But the recommended way to subscribe is to use:
 * <ul>
 * <li>{@code Rx.subscribe(listenableOrObservable, Rx.onValue(val -> doSomething(val)));}</li>
 * <li>{@code Rx.subscribe(listenableOrObservable, val -> doSomething(val)); // automatically uses Rx.onValue()}</li>
 * </ul>
 * The advantage of this latter method is that it returns {@link rx.Subscription} instances
 * which allow you to unsubscribe from futures in the same manner as for observables.
 * <ul>
 * <li>{@code subscription = Rx.subscribe( ... )}</li>
 * </ul>
 * If you wish to receive callbacks on a specific thread, you can use:
 * <ul>
 * <li>{@code Rx.on(someExecutor).subscribe( ... )}</li>
 * </ul>
 * Because RxJava's Observables use {@link rx.Scheduler}s rather than {@link java.util.concurrent.Executor}s,
 * a Scheduler is automatically created using {@link rx.Schedulers#from}. If you'd like to specify the Scheduler manually, you can use {@link Rx#on(Executor, Scheduler)}
 * or you can create an executor which implements {@link Rx.HasRxExecutor}.
 * 
 * @see <a href="https://diffplug.github.io/durian-swt/javadoc/snapshot/com/diffplug/common/swt/SwtExec.html">SwtExec</a>
 */
public class Rx {
	private Rx() {}

	/**
	 * Creates an Rx instance which will call the given consumer whenever a value is received.
	 * Any errors are sent to ErrorHandler.log().
	 */
	public static <T> RxListener<T> onValue(Consumer<T> onValue) {
		return new RxListener<T>(onValue, RxListener.logErrors);
	}
{'fraction_non_alphanumeric': 0.1415985467756585, 'fraction_numerical': 0.0014532243415077202, 'mean_word_length': 4.770964360587002, 'pattern_counts': {'":': 0, '<': 135, '<?xml version=': 0, '>': 165, 'https://': 4, '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': '3085247', 'n_tokens_mistral': 3385, 'n_tokens_neox': 3304, 'n_words': 923}