angular - 如何使异步数据可观察？

Question

Observables 的新手。我正在使用 ssh2 获取服务器上的文件/文件夹列表。由于所有在线示例都使用 http 而不是第 3 方模块，因此我无法弄清楚如何将我的数据作为可观察对象取回。

如何设置此服务，以便我可以将列表作为可观察对象取回？

import {Injectable, NgZone} from '@angular/core';
import {Observable} from 'rxjs';
var Client = require('ssh2').Client;
var user = require('credentials.json')
@Injectable()
export class ConnectionService {
    public connSettings:any;

    constructor(private zone: NgZone){
        this.connSettings = {
            host: user.url,
            username: user.username,
            password: user.password
        };
        this.openConnection();

    }
    openConnection() {
        let remotePathToList = '/home/user';
        var conn = new Client();

        conn.on('ready', ()=>{
            conn.sftp((err:any, sftp:any)=>{
                if (err) throw err;
                sftp.readdir(remotePathToList, (err:any, list:any)=>{
                    conn.end();

                    //return observable data here

                    return list;

                })
            })
        }).connect(this.connSettings);

    }


}

Answer 1

如果您希望通用属性随时访问列表(与openConnection() 方法无关，您可以使用ReplaySubject:

@Injectable()
export class ConnectionService {
    public yourList$: ReplaySubject<any> = new ReplaySubject<any>(1);
    public connSettings:any;

    constructor(private zone: NgZone){
        this.connSettings = {
            host: user.url,
            username: user.username,
            password: user.password
        };
        this.openConnection();

    }
    openConnection() {
        let remotePathToList = '/home/user';
        var conn = new Client();

        conn.on('ready', ()=>{
            conn.sftp((err:any, sftp:any)=>{
                if (err) throw err;
                sftp.readdir(remotePathToList, (err:any, list:any)=>{
                    conn.end();

                    //return observable data here
                    yourList$.next(list);
                    return list;

                })
            })
        }).connect(this.connSettings);
    }
}

---

如果您只想将它​​作为 openConnection 的返回 Observable，您可以创建自定义 Observable:

@Injectable()
export class ConnectionService {
    public yourList$: ReplaySubject<any> = new ReplaySubject<any>(1);
    public connSettings:any;

    constructor(private zone: NgZone){
        this.connSettings = {
            host: user.url,
            username: user.username,
            password: user.password
        };
        this.openConnection();

    }
    openConnection(): Observable<any> {
        return Observable.create(obs => {
            let remotePathToList = '/home/user';
            var conn = new Client();

            conn.on('ready', ()=>{
                conn.sftp((err:any, sftp:any)=>{
                    if (err) throw err;
                    sftp.readdir(remotePathToList, (err:any, list:any)=>{
                        conn.end();

                        //return observable data here
                        obs.next(list);
                        obs.complete(); // required, otherwise your subscription would never close
                    })
                })
            }).connect(this.connSettings);
        });
    }
}

请注意，对于第二种解决方案，您必须在 openConnection 上调用 subscribe 否则什么也不会发生，因为它现在被包裹在一个 observable 中:

connectionService
    .openConnection()
    .subscribe(list => console.log(list));