не в состоянии понять, сантехника в Rxjs

Question

Что я понял до сих пор о Rxjs, так это о том, что Observable - это поток, и мы можем «выдвинуть» новые значения в потоке, используя метод Observer 'next.Затем я подписываюсь на поток для получения значений.Но я не могу понять, как работает сантехника.Например,

Я могу создать Observable как

var observable = Rx.Observable.of('foo', 'bar');

Тогда я могу сделать

var subscription =  observable.subscribe(x=>console.log(x))

1. Где и как наблюдатель создан выше?
2. Кто звонит next, чтобы выдвинуть значение в потоке?
3. Я думал, что x=>console.log(x) - это Наблюдатель, но у него нет next метода.Как работает приведенный выше код?

Вот еще один пример.Ниже я передаю observer и использую его метод next для передачи значений, но я никогда не создавал observer (объект, который имеет следующий метод).

var myObservable = Rx.Observable.create(observer => {
  observer.next('foo');
  setTimeout(() => observer.next('bar'), 1000);
});
myObservable.subscribe(value => console.log(value));

Answer 1

(1) Где и как выше создается наблюдатель?

Вы создаете Обозреватель, когда подписываетесь.Вы предоставляете Observer в качестве аргумента для метода подписки.Например, что-то вроде следующего:

var subscription =  observable.subscribe(
{
next: x=>console.log(x),
error: e=>console.log(e),
complete: ()=>console.log('completed')
}
)

Метод подписки также принимает один или все методы Observer напрямую.Например, что-то вроде следующего:

var subscription =  observable.subscribe(x=>console.log(x))

Выше приведен синтаксический сахар для следующего:

var subscription =  observable.subscribe(
    {
    next:x=>console.log(x)
    }
    )

(2) Кто звонит рядом с push-значением в потоке?

В кодах для создания Observable есть вызов метода next (см. Ниже).После того как вы подписались (это означает, что вы указали Observer в качестве аргумента метода подписки), будет выполнен следующий код.Обратите внимание, что в следующих кодах используется Observer, который вы указали при подписке.Здесь вызывается метод next, который вы определили в Observer.

var myObservable = Rx.Observable.create(observer => {
  observer.next('foo');
  setTimeout(() => observer.next('bar'), 1000);
});
myObservable.subscribe(value => console.log(value));

(3) Я думал, что x => console.log (x) - это Observer, но у него нет следующего метода.Как работает приведенный выше код?

Как я объяснил ранее, здесь вы указали next метод Observer.