rxjs - react - ¿Cuál es la diferencia entre Rx.Observable subscribe y forEach?

rxjs operators

Question

Después de crear un observable como tal

var source = Rx.Observable.create(function(observer) {...});

¿Cuál es la diferencia entre suscribirse?

source.subscribe(function(x) {});

y para cada

source.forEach(function(x) {});

Answer 1

Acabo de revisar el último código disponible, técnicamente el código de foreach en realidad está llamando a suscribirse en RxScala, RxJS y RxJava. No parece una gran diferencia. Ahora tienen un tipo de retorno que permite al usuario tener una forma de detener una suscripción o similar.

Cuando trabajo en la versión anterior de RxJava, la suscripción tiene un retorno de suscripción, y forEach es solo un vacío. Que puede ver alguna respuesta diferente debido a los cambios.

/**
 * Subscribes to the [[Observable]] and receives notifications for each element.
 *
 * Alias to `subscribe(T => Unit)`.
 *
 * $noDefaultScheduler
 *  
 * @param onNext function to execute for each item.
 * @throws java.lang.IllegalArgumentException if `onNext` is null
 * @throws rx.exceptions.OnErrorNotImplementedException if the [[Observable]] tries to call `onError`
 * @since 0.19
 * @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
 */
def foreach(onNext: T => Unit): Unit = {
    asJavaObservable.subscribe(onNext)
 }

def subscribe(onNext: T => Unit): Subscription = {
    asJavaObservable.subscribe(scalaFunction1ProducingUnitToAction1(onNext))
}

/**
 *  Subscribes an o to the observable sequence.
 *  @param {Mixed} [oOrOnNext] The object that is to receive notifications or an action to invoke for each element in the observable sequence.
 *  @param {Function} [onError] Action to invoke upon exceptional termination of the observable sequence.
 *  @param {Function} [onCompleted] Action to invoke upon graceful termination of the observable sequence.
 *  @returns {Disposable} A disposable handling the subscriptions and unsubscriptions.
 */
observableProto.subscribe = observableProto.forEach = function (oOrOnNext, onError, onCompleted) {
  return this._subscribe(typeof oOrOnNext === ''object'' ?
    oOrOnNext :
    observerCreate(oOrOnNext, onError, onCompleted));
};

/**
 * Subscribes to the {@link Observable} and receives notifications for each element.
 * <p>
 * Alias to {@link #subscribe(Action1)}
 * <dl>
 *  <dt><b>Scheduler:</b></dt>
 *  <dd>{@code forEach} does not operate by default on a particular {@link Scheduler}.</dd>
 * </dl>
 * 
 * @param onNext
 *            {@link Action1} to execute for each item.
 * @throws IllegalArgumentException
 *             if {@code onNext} is null
 * @throws OnErrorNotImplementedException
 *             if the Observable calls {@code onError}
 * @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
 */
public final void forEach(final Action1<? super T> onNext) {
    subscribe(onNext);
}

public final Disposable forEach(Consumer<? super T> onNext) {
    return subscribe(onNext);
}

Answer 2

En la especificación ES7 , que sigue a RxJS 5.0 (pero no a RxJS 4.0), los dos NO son iguales.

suscribir

public subscribe(observerOrNext: Observer | Function, error: Function, complete: Function): Subscription

Observable.subscribe es donde realizará la mayor parte de su verdadero manejo de Observable. Devuelve un token de suscripción, que puede utilizar para cancelar su suscripción. Esto es importante cuando no sabe la duración de los eventos / secuencia a los que se ha suscrito, o si necesita dejar de escuchar antes de una duración conocida.

para cada

public forEach(next: Function, PromiseCtor?: PromiseConstructor): Promise

Observable.forEach devuelve una promesa que se resolverá o rechazará cuando el Observable se complete o cometa errores. Su objetivo es aclarar las situaciones en las que está procesando una secuencia observable de duración limitada / finita de una manera más ''sincrónica'', como compaginar todos los valores entrantes y luego presentarlos una vez, cumpliendo la promesa.

Efectivamente, puede actuar en cada valor, así como eventos de error y finalización de cualquier manera. Entonces, la diferencia funcional más significativa es la incapacidad de cancelar una promesa.

Answer 3

observable.subscribe(...) y observable.forEach(...) son exactamente iguales.
Así como user3743222 aludió en su answer

De la documentación de Rx:

Suscribe un observador a la secuencia observable.

Rx.Observable.prototype.subscribe([observer] | [onNext], [onError], [onCompleted])

Rx.Observable.prototype.forEach([observer] | [onNext], [onError], [onCompleted])

Ver la documentation completa para más detalles.