代码之家 › 专栏 › 技术社区 › Chris

如何将RxJs observable的结果作为节点的rest响应返回。js公司

nio observable rxjs rest node.js

Chris · 技术社区 · 8 年前

脚本: 来自多个rest调用的数据必须聚合到一个对象中,并作为通过节点服务的初始请求的rest响应返回。js。

问题: rest响应不会等到可观察到的完成,因此在调度rest响应后实现突变(聚合)。

//teamsController class invoked via GET /teams 

import * as Rx from 'rxjs/Rx'
import http from 'axios'
import Teams from '../models/teams'

const teamsAPI = "http://localhost:8081/api/v1/teams/players/";
const usersAPI = "http://localhost:8082/api/v1/users/";

exports.getTeamByPlayer = function (req, res) {

let username= req.get("username");

    Rx.Observable.fromPromise(fetchTeam(username))
        .map(team => {
            Rx.Observable.from(team.players).subscribe(player => {
                console.log(`Player name is ${player.username}`);
                Rx.Observable.fromPromise(fetchUser(player.username))
                    .map(avatar => avatar.avatar)
                    .subscribe(avatar => {
                        player.avatar = avatar;
                        console.log(player)
                    })
            });
            return team;
         })
        .subscribe(result => {
            console.log(`result is ${JSON.stingify(result)}`);
            res.json(result);

        })
}


/**
 * Fetch a team by a player
 *
 * @param name The name of the team
 * @returns {Promise.<Teams>}
 */
function fetchTeam(name) {
    return http.get(teamsAPI + name)
        .then(response => new Teams(response.data.data))
        .catch(error => {
            throw  new Error("todo: fill error message");
        })
}

/**
 * Fetch a user given its username
 *
 * @param username The username of the player
 * @returns {Promise.<TResult>}
 */
function fetchUser(username) {
    return new Promise(function (resolve, reject) {
        console.log(`fetching user: ${username}`);
        resolve();
    }).then(() => {
        return {
            "avatar": {
                "flagColor": "dummyValue",
                "flagCrest": "dummyValue"
            }
        }
    });

日志结果:

Player name is username1
fetching user: username1
Player name is username2
fetching user: username2
result is {"id":"5a1c2a4030c39e5d88aed087","name":null,"avatar":{"flagColor":"abcdefg","flagCrest":"hijklmn"},"description":"string","motto":"string","players":[{"userId":"59b94a7b8b68ef0a048e85c1","username":"username1","status":"ACTIVE","role":"ADMIN","dateJoined":1511795264314,"dateInvited":null,"score":0},{"userId":"59b94a7b8b68ef0a048e85c1","username":"username2","status":"ACTIVE","role":"MEMBER","dateJoined":1511795264314,"dateInvited":null,"score":0}],"score":0,"type":"TEAM","open":true,"location":{"longitude":0,"latitude":0,"country":"string"},"owner":"username1","dateCreated":1511795264314}
{ userId: '59b94a7b8b68ef0a048e85c1',
  username: 'username1',
  status: 'ACTIVE',
  role: 'ADMIN',
  dateJoined: 1511795264314,
  dateInvited: null,
  score: 0,
  avatar: { flagColor: 'dummyValue', flagCrest: 'dummyValue' } }
{ userId: '59b94a7b8b68ef0a048e85c1',
  username: 'username2',
  status: 'ACTIVE',
  role: 'MEMBER',
  dateJoined: 1511795264314,
  dateInvited: null,
  score: 0,
  avatar: { flagColor: 'dummyValue', flagCrest: 'dummyValue' } }

AppServer:节点。JS v8.7.0,中间件:expess 4.16.2,Libs:RxJs 5.0.0-beta。12,axios 0.17.1

1 回复 | 直到 8 年前

bryan60 8 年前

function fetchTeam(name) {
    return http.get(teamsAPI + name).map(res => new Teams(res.json().data));
}

function fetchUser(username) {
    return Rx.Observable.of({
        "avatar": {
            "flagColor": "dummyValue",
            "flagCrest": "dummyValue"
        }
    });
}

如果一些外部消费者需要基于承诺的api,那么可以使用公共函数将这些私有函数封装在承诺中。

其次,map是一个同步操作符,您不能在其中执行异步操作,您需要使用一个异步操作符来为您完成订阅工作。如果你发现自己在一个可观察的流中订阅,那么你就做错了。

let username= req.get("username");

fetchTeam(username)
    .switchMap(team => { // switchMap will subscribe to inner observables
        //build observables for fetching each avatar, pay close attention, I'm mixing the array map operator and the rx map operator in here
        let players$ = team.players.map(player => fetchUser(player.username).map(avatar => avatar.avatar)); 

        return Rx.Observable.forkJoin(players$); // use forkjoin to executre requests
     }, (team, playerAvatars) => { // use second switchMap argument to combine results
         team.players.forEach((player, idx) => player.avatar = playerAvatars[idx]);
         return team;
     })
    .subscribe(result => {
        console.log(`result is ${JSON.stingify(result)}`);
        res.json(result);

    });