使用 Baconjs 永无止境地流式传输

Question

我有一个简单的脚本，它只从 Redis 列表中获取所有值并将它们打印到控制台。

var redis = require("redis"),
    client = redis.createClient();
    Bacon = require('baconjs');

Bacon.repeat(function(){
   return Bacon.fromNodeCallback(
    client, 'lpop', ['errors']
  );
})
.takeWhile(function(val) {return val !== null;})
.fold(
    [],
    function(acc, next) {
      acc.push(next); return acc;
    }
).onValue(console.log);

程序打印了正确的列表，但从未完成。我该如何解决这个问题？为什么会这样？

Answer 1

一个简单的解决方案是在 onValue 处理程序中调用 process.exit。

).onValue(function(value) {
  console.log(value)
  process.exit(0)
})

编辑：您可以编写一个自定义的 redit 源，它会在不再需要时关闭连接。

Bacon.fromBinder(function(sink) {
  var client = redis.createClient()
  sink(new Bacon.Next(client))
  return function unsubscribe() {
    client.quit()
  }
}).flatMap(function(client) {
  return Bacon.repeat(function(){
    return Bacon.fromNodeCallback(
      client, 'lpop', ['errors']
    );
  })
})

使用 Baconjs 永无止境地流式传输

Never ending stream with Baconjs

javascript

node.js

redis

reactive-programming

bacon.js