Контекст в Highland.js

Question

Я люблю Highland.js и стиль реактивного программирования в целом. Я борюсь с потерей контекста, и я пытаюсь определить, как элегантно обрабатывать контекст в модели, где целью является отказ от состояния.

В качестве примера у меня есть массив моих учетных записей в Amazon Web Services.

var accounts = [{accessId:"12345","secretKey":"abc123","account":"foo"}, 
       {accessId:"34512","secretKey":"def456","account":"bar"}]; 

Моя цель - создать таблицу всех моих экземпляров EC2, работающих в регионе. Что-то вроде этого.

Account | Instance Size 
--------- | ------------- 
foo  | m3.xlarge 
foo  | c3.medium 
bar  | t2.small 

Общий рабочий будет

1. Пройдите через каждый счет
2. ec2DescribeInstaces Вызов
3. Как-карту каждый ec2DescribeInstances перезвоним имя учетной записи для окончательного вывода

В обычный JavaScript, мы будем делать цикл здесь, поэтому, когда мы делаем каждый вызов ec2Des cribeInstances, контекст будет существовать

for (account in accounts) { 
    var instances = ec2DescribeInstaces(account); 
    for (instance in instances) { 
    results.push({account:account.name, instanceSize: instance.size}); 
    } 
} 

Из того, что я понимаю в реактивном программировании, я хотел бы сделать что-то вроде этого

_(accounts) 
.map(ec2DescribeInstaces) 
.parallel(2) 
.each(function(result) { 
    results.push(result); 
}); 

Любой инструктивный ??? Итак, в конце этой цепи у меня есть экземпляры из Amazon. Но я не уверен, как связать их с аккаунтами, чтобы получить имя. Я знаю, что могу взломать это, чтобы получить ценность, но я ищу лучшие практики и что-то элегантное.

Итак, @Bergi, как показано ниже ??? Это потребовало бы возврата «контекстного» объекта как с данными, которые мне нужны в дополнение к данным, которые возвращались из Amazon. Моя единственная забота об этом заключается в том, что если я передаю контекст по всей цепочке, мы будем много собирать данные, набивать и обертывать для звонков.

ec2DescribeInstances = _.wrapCallback(function(accountData, callback) { 
    // we remove the extraneous account name using a pick 
    var ec2 = new AWS.EC2(lodash.pick(accountData,'accessKeyId','secretAccessKey')); 
    ec2.describeInstances({ Filters: [{Name:'instance-state-name', Values:['running']}] }, function(err,data) { 
    if (err) return callback(err); 
    // here I create an object that wraps the AWS response 
    callback(null, {"account":accountData.alias, "data": data}) 
    }); 
}); 

Answer 1

Вы можете использовать _.zip(), которая принимает два потока и возвращает поток пара. Итак, используя ваш поток, который генерировал ec2Instances в качестве примера:

var ec2InstanceStream = _(accounts) 
    .map(ec2DescribeInstaces) 
    .parallel(2); 

_(acounts).zip(ec2InstanceStream); 
    .each(function(result) { 
     //Each result: [{ /* original account */ }, { /* ec2 instance */ }] 
     results.push(result); 
    }); 

Итак, теперь у вас есть все вместе. Конечно, вы можете сделать это лучше. Например, если вы хотите объединить каждую пару в один объект, вы можете добавить .map(_.extend) до этапа .each(), чтобы объединить объекты вместе.

Answer 2

Я бы гнездятся операции, чтобы получить некоторые возможности:

const h = require("highland"); 
const ec2DescribeInstaces = h.wrapCallback(function(account, cb) { 
    return cb(null, [{ 
    size: account.size 
    }, { 
    size: ++account.size 
    }]); 
}); 

const accounts = [{ 
    size: 1, name: 1 
}, { 
    size: 2, name: 2 
}, { 
    size: 3, name: 3 
}]; 

var results = []; 
h(accounts) 
    .map(account => { 
    return ec2DescribeInstaces(account) 
     .flatten() 
     .map(instance => ({ 
     account: account.name, 
     instanceSize: instance.size, 
     })); 
    }) 
    .parallel(2) 
    .tap(results.push.bind(results)) 
    .collect() 
    .toCallback(h.log) 

Таким образом, вы еще знаете об индивидуальном account, что каждый instance связан с в то время как вы работаете с этим экземпляром, и потоки по-прежнему будет работать параллельно, потому что результатом map является поток.

Кроме того, я считаю, что это широко применимый подход. Неважно, сколько экземпляров вы вернетесь или сколько вещей вы хотите сделать для каждой учетной записи.Например, вы могли бы вернуть и parallel(2) два потока из map, если вам нужно было нажать два значения до results для каждого экземпляра в учетной записи, и вам не нужно будет полностью перенастраивать ваш подход, просто измените поведение.