如何在对象方法上使用fromEvent而不是bindCallback

Question

我需要将rn fetch blob的onData回调方法连接到一个可观察对象。

据我所知，这不是一个事件，不能使用fromEventPattern。如果这是我问题的解决方案，我不知道如何使用create。

我发现bindCallback看起来很有前景，但文档说我可能应该使用fromEvent：

请注意，输出函数创建的Observable将始终发出一个值，然后立即完成。如果func多次调用回调，则后续调用的值将不会出现在流中。如果需要监听多个调用，则可能需要使用fromEvent或fromEventPattern。

我确实需要接听多个电话。

无论如何，我正在尝试对一个对象方法使用bindCallback，如文档中所示。在我的Typescript文件中：

import { bindCallback } from 'rxjs';

在我的班级：

private emitter!: Observable<any>;

在私有方法中：

RNFetchBlob.fs
.readStream(
filePath,
"utf8",
-1,
10
)
.then(ifstream => {
ifstream.open();
this.emitter = bindCallback(ifstream.onData);

但无法编译：

error TS2322: Type '() => Observable<string | number[]>' is not assignable to type 'Observable<any>'.
Property '_isScalar' is missing in type '() => Observable<string | number[]>'.

我真的不知道在我的情况下如何使用fromEvent。

感谢您的帮助。

编辑：为那些寻找答案的人添加了工作代码：

RNFetchBlob.fs
.readStream(
// file path
peripheral.name,
// encoding, should be one of `base64`, `utf8`, `ascii`
"utf8",
// (optional) buffer size, default to 4096 (4095 for BASE64 encoded data)
// when reading file in BASE64 encoding, buffer size must be multiples of 3.
-1,
10
)
.then(ifstream => {
ifstream.open();
this.emitter = new Observable(subscriber => {
ifstream.onData(chunk => {
// chunk will be a string when encoding is 'utf8'
logging.logWithTimestamp(`Received [${chunk}]`);
subscriber.next(chunk);
});
ifstream.onError(err => {
logging.logWithTimestamp(`oops [${err}]`);
subscriber.error(err);
});
ifstream.onEnd(() => {
subscriber.complete();
});
});
this.rxSubscription = this.emitter
.pipe(
concatMap(value =>
this.handleUpdatedValuesComingFromCSVFile(value)
)
)
.subscribe();

Answer 1

对rn-fetch-blob不太熟悉，但希望您能理解，还可以返回一个函数来运行清理逻辑。

const onDataObserevable=new Obserevable(obs=>{
ifstream.onData(data=>obs.next(data))
return ()=>{... you can add some clean up logic here like unsubcribe from source onData event}
});

更新

将整个链转换为可观察的，希望你能理解

from(RNFetchBlob.fs.readStream(
peripheral.name,
"utf8",
-1,
10
))
.pipe(mergeMap(ifstream => {
ifstream.open();
return new Observable(subscriber => {
ifstream.onData(chunk => {
// chunk will be a string when encoding is 'utf8'
logging.logWithTimestamp(`Received [${chunk}]`);
subscriber.next(chunk);
});
ifstream.onError(err => {
logging.logWithTimestamp(`oops [${err}]`);
subscriber.error(err);
});
ifstream.onEnd(() => {
subscriber.complete();
});
});),
mergeMap(value =>this.handleUpdatedValuesComingFromCSVFile(value)
)
)

Answer 2

以下是如何使用fromEventPattern:

this.emitter = fromEventPattern(
// tell fromEventPattern how to subscribe to the data
handler => ifstream.onData(handler),
// tell fromEventPattern how to unsubscribe from the data
handler => ifstream.onData(null)
)