将普通的string []转换为Observable< string []>.并将其连接到另一个Observable< string []>使用RxJS 5

问题描述

我正在尝试将普通的 string [] 转换为 Observable< string []> 并将其连接到现有的 Observable< string []> .

I am trying to convert a plain string[] into a Observable<string[]> and concat it to an existing Observable<string[]>.

然后我将使用angular2 async 管道以显示 Observable .

I will then use an angular2 async pipe in order to display the Observable.

这是我的代码:

import {Injectable} from "angular2/core";
import {Observable} from "rxjs/Observable";
import 'rxjs/Rx';


@Injectable()
export class AppService {

    constructor() {
        console.log('constructor', 'appService');
        this.constructSomeObservable();
    }

    someObservable$:Observable <string[]>;

    constructSomeObservable() {
        this.someObservable$ = Observable.create(observer => {
            const eventSource = new EventSource('/interval-sse-observable');
            eventSource.onmessage = x => observer.next(JSON.parse(x.data));
            eventSource.onerror = x => observer.error(console.log('EventSource failed'));
            return () => {
                eventSource.close();
            };
        });

        this.someObservable$.subscribe(
            theStrings=> {
                console.log(theStrings);
                //Somehow convert the plain array of strings to an observable and concat it to this.someObservable$ observable...
            },
            error=>console.log(error)
        );
    }
}

任何人都可以帮忙吗?

此外，我想确保服务实例 Observable< string []> 不断更新，因为EventSource被重复调用.我的订阅逻辑在正确的位置吗?

Furthermore, I want to ensure that the service instance Observable<string[]> is continuously updated as the EventSource is called repeatedly. Is my subscribe logic in the right place?

编辑1 :我尝试如下使用RxJS concat 运算符:

edit 1: I tried to use RxJS concat operator as follows:

    this.someObservable$.subscribe(
        theStrings=> {
            console.log(theStrings);
            this.someObservable$ = this.someObservable$.concat(Observable.create(theStrings));
        },
        error=>console.log(error)
    );
 }

与angular2 async 管道一起

together with the angular2 async pipe:

<ul>
    <li *ngFor="#s of appService.someObservable$ | async">
       a string: {{ s }}
    </li>
</ul>

，页面上什么也没有显示；字符串刚刚显示在控制台上...

and nothing gets displayed on the page; the strings just get displayed on the console...

我出了什么问题?

编辑2 :该应用程序位于github 此处

edit 2: The app is available on github here

编辑3 :我考虑了Thierry的建议，特别是使用 async 管道进行订阅以及使用扫描运算符的方式.

edit 3: I have taken into account Thierry's advice, especially the use of the async pipe in order to subscribe as well as the usage of the scan operator.

现在唯一剩下的问题是我需要单击路由器链接以使字符串在模板上呈现...模板不会自动更新...

The only remaining issue now is that I need to click on the router link in order for the strings to render on the template... The template does not update automatically...

请参阅github上的项目和相关标签: https://github.com.com/balteo/demo-angular2-rxjs/tree/36864628/536299

See project on github and relevant tag: https://github.com/balteo/demo-angular2-rxjs/tree/36864628/536299

推荐答案

我会利用 scan 运算符来做到这一点.这是一个示例:

I would leverage the scan operator to do that. Here is a sample:

@Component({
  selector: 'app'
  template: `
    <div>
      <div *ngFor="#elt of someObservable$  | async">{{elt.name}</div>
    </div>
  `
})
export class App {
  constructor() {
    this.someObservable$ = Observable.create((observer) => {
      const eventSource = new EventSource('/interval-sse-observable');
      eventSource.onmessage = x => observer.next(JSON.parse(x.data));
      eventSource.onerror = x => observer.error(console.log('EventSource failed'));
      return () => {
        eventSource.close();
      };
    })
    .startWith([])
    .scan((acc,value) => acc.concat(value));
  }
}

以下是相应的插件: https://plnkr.co/edit/St7LozX3bnOBcoHaG4uM?p=预览.

有关更多详细信息，请参见此问题:

See this question for more details:

• Angular 2 AsynPipe不适用于Observable

这篇关于将普通的string []转换为Observable< string []>.并将其连接到另一个Observable< string []>使用RxJS 5的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！