javascript - 我如何使这个 angular2 服务稍微同步？

Question

我有一个 angular2 服务，我希望它执行以下操作:

1. FTP 到远程服务器
2. 找到一个文件并从中读取一些行
3. 构建“结果”json 对象并返回到调用组件

所以 - 实际上我已经完成了步骤 1/2 - 但当然它都是“异步”的。所以发生的事情是在我的组件中，我正在对服务进行调用，其中 this.ftp 是我的服务的实例:

 this.servers = this.ftp.lookForServers();

现在，这可以正确调用我的 FTP 服务的lookForServers 方法，如下所示:

lookForServers(){
    var servers = [];
    var whereAreWe = 0;
    var possibles = ["/path/to/servers/"];
    for(var i=0;i<possibles.length;i++){
      whereAreWe = i;
      this.c.list(possibles[i],false,(err,list)=>{
        for(var p=0;p<list.length;p++){
          console.log(list[p]);
          var server_version = this.grabLog(possibles[whereAreWe]+list[p].name);
          servers.push({
            name: list[p].name,
            path: possibles[whereAreWe]+list[p].name,
            version: server_version
          });
        }
      });
    }
    return servers;
   }

现在 - this.grabLog(possibles[whereAreWe]+list[p].name); 函数调用最终会进一步调用 this.c - FTP 客户端，这当然是是异步的，因此该方法几乎立即返回 - 同时回调继续运行。这些回调下载一个文件，然后另一个回调函数处理该文件 - 再次逐行，异步挑选出我想要存储的各种详细信息。

在这条链的末尾 - 我在最后得到了我的所有详细信息:

lineReader.on('close', () => { function - but of course my `this.ftp.lookForServers();` function call has long gone....and the component is none the wiser.

那么我怎样才能让这项工作异步发生，并且在工作完成后仍然将我的结果 JSON 对象传递回组件呢？这可能是一个非常简单的问题，关于如何使服务调用成为组件回调......？

Answer 1

您不需要它同步运行。您应该让lookForServers(以及它正在使用的其他函数)使用可观察量，然后像这样订阅结果:

this.ftp.lookForServers().subscribe((data) => {  this.servers = data });

以下是实现:

const Client = require('ftp');
const fs = require('fs');
const readline = require('readline');

import { NextObserver } from 'rxjs/Observer';
import { Observable } from 'rxjs/Rx';

interface server {
    name: string;
    path: string;
    version: string;
    java_version: string;
}

export class FTPClient {
    username: string;
    password: string;
    host: string;
    port: number;
    c: any;

    constructor() {
    }

    init(username, password, host, port) {
        console.log("initiating FTP connection to:" + host + "on port:" + port);

        this.username = username;
        this.password = password;
        this.host = host;
        this.port = port;
        this.c = new Client();

        console.log("Client created");
    }

    connect() {
        console.log("About to start connection");

        this.c.on('ready', () => {
            this.c.list((err: any, list: any) => {
                if (err) throw err;

                console.dir(list);
                this.c.end();
            });
        });

        // connect to localhost:21 as anonymous
        var connectProps = {
            host : this.host,
            port : this.port,
            user : this.username,
            password : this.password
        };

        console.log("Connecting now...");
        this.c.connect(connectProps);
    }

    public lookForServers(name: string): Observable<any[]> {
        return Observable.create((observer: NextObserver <any[]>) => {
            let servers = [];
            let whereAreWe = 0;
            let possibles = [ "/path/to/servers/" ];

            for (var i = 0; i < possibles.length; i++) {
                whereAreWe = i;

                this.c.list(possibles[ i ], false, (err: any, list: any) => {
                    for (var p = 0; p < list.length; p++) {
                        this.grabMessagesLog(possibles[ whereAreWe ] + list[ p ].name)
                        .subscribe((data: any) => {
                                let server_version = data;

                                servers.push({
                                    name : list[ p ].name,
                                    path : possibles[ whereAreWe ] + list[ p ].name,
                                    version : server_version
                                });

                                observer.next(servers);
                                observer.complete();
                            }
                        );
                    }
                });
            }
        });
    }

    grabMessagesLog(path): Observable<any> {
        return Observable.create((observer: NextObserver <any>) => {
            let result = '';
            let unix = Math.round(+new Date() / 1000);
            this.c.binary(function(err) {
                console.log(err);
            });

            this.c.get(path + "/logs/messages.log", (err, stream) => {
                if (err) throw err;

                stream.once('close', () => {
                    this.c.end();
                    this.getServerMetadataFromMessagesLog(unix + "_messages.log")
                    .subscribe((data) => {
                        stream.pipe(fs.createWriteStream(unix + "_messages.log"));

                        observer.next(data);
                        observer.complete();
                    });
                });
            });
        });
    }

    getServerMetadataFromMessagesLog(path): Observable<any> {
        return Observable.create((observer: NextObserver <any>) => {
            let lineReader = readline.createInterface({
                input : fs.createReadStream(path)
            });

            let server_version = "";
            let java_version = "";
            let line_no = 0;

            lineReader.on('line', function(line) {
                line_no++;
                console.log("line number is:" + line_no);

                if (line.includes("STUFF") && line.includes("FLAG2") && line_no == 2) {
                    var first = line.split("FLAG2")[ 1 ];
                    var last = first.split(" (")[ 0 ];
                    var version = "FLAG2" + last;
                    this.server_version = version;
                    console.log("version is:" + version);
                }

                if (line.includes("java.version =")) {
                    var javav = line.split("java.version =")[ 1 ];
                    this.java_version = javav;
                    lineReader.close();
                }

                console.log('Line from file:', line);
            });

            lineReader.on('close', () => {
                var res = {
                    version : server_version,
                    java_version : java_version
                };

                alert("RES IS:" + JSON.stringify(res));

                observer.next(res);
                observer.complete();
            });
        });
    }
}