Как создать наблюдаемый из функции?

Question

Я хочу вызвать функцию (синхронно), а затем использовать ее возвращаемое значение как начальное излучение (впоследствии связать некоторые другие операторы по полученным наблюдаемым).

Я хочу использовать эту функцию во время подписки, поэтому я не могу просто использовать Observable.of(() => getSomeValue()). Я видел bindCallback (ранее fromCallback), но я не думаю, что он может быть использован для этой задачи (исправьте меня, если я ошибаюсь). Я видел start статический оператор в документах v4, но, по-видимому, он не реализован в версии 5 (и никаких указаний на то, что он в пути). У RxJava также есть оператор fromCallable, который делает именно этот афайк.

Только так я мог думать о том, как это:

Observable.create((observer: Observer<void>) => { 
    let val = getSomeValue(); 
    observer.next(val); 
    observer.complete(); 
}) 

, который я думаю, что делает именно это. Но это просто кажется сложным для простой вещи, которая, вероятно, была бы такой, как Observable.fromFunction(() => getSomeValue()) И что, если я хочу запустить ее асинхронно, например, start оператор делает? Как я могу сделать это в текущем правиле RxJS?

Answer 1

На самом деле я думаю, что лучший вариант - использовать Observable.create, потому что это самое универсальное решение для синхронных и асинхронных начальных значений.

Если вы уверены, что используете синхронную функцию, вы можете использовать оператор startWith() (это имеет смысл только в том случае, если возвращаемое значение от getSomeValue() должно быть одинаковым для всех наблюдателей).

Использование Observable.bindCallback в качестве источника. Наблюдаемое, конечно, возможно, однако я лично рекомендую избегать его, потому что он делает ваш код очень трудным для понимания, и это обычно не обязательно, потому что вы можете использовать только Observable.create.

Answer 2

Я стараюсь избегать явного использования Observable.create, где это возможно, потому что, как правило, это источник ошибок, которым приходится управлять не только выделением событий, но и вашей логикой разрыва.

Вместо этого вы можете использовать Observable.defer. Он принимает функцию, которая возвращает Observable или вещь Observable-like (читай: Promise, Array, Iterators). Так что если у вас есть функция, которая возвращает асинхронной вещь так же легко, как:

Observable.defer(() => doSomethingAsync()); 

Если вы хотите, чтобы работать с синхронным результатом то сделать:

Observable.defer(() => Observable.of(doSomethingSync())); 

Примечание: Это как create этой будет повторно запускать функцию для каждой подписки. Это иначе, чем результат Observable.bindCallback, который сохраняет результат вызова функции без повторного выполнения функции. Поэтому, если вам нужно такое поведение, вам нужно будет использовать соответствующий оператор multicasting.

Answer 3

Реализация в fromFunction$, который я использовал в моем проекте:

function fromFunction$(factory:() => any) { 
    return Rx.Observable.create((observer) => { 
    try { 
     observer.next(factory()); 
     observer.complete(); 
    } catch (error) { 
     observer.error(error); 
    } 
    }); 
} 

Используется как:

fromFunction$(() => 0).subscribe((value) => console.log(`Value is '${value}'`), null,() => console.log('Completed')); 
fromFunction$(() => [1, 2, 3]).subscribe((value) => console.log(`Value is '${value}'`), null,() => console.log('Completed')); 
fromFunction$(() => { throw 'Something' }).subscribe(null, (error) => console.error(`Error: ${error}`)); 

Дает:

Value is '0' 
Completed 

Value is '1,2,3' 
Completed 

Error: Something 

Пока не существует такая реализация.