在简单的节点模块中尝试RxJS时，“ TypeError：this._subscribe不是函数”

我有以下节点代码

//index.js
var username = process.argv[2],
    password = process.argv[3],
    factories = require('./factories')(username, password);

factories.subscribe(function(data){
  console.log(data.length);
});
factories.refresh();

//factories.js
var Rx = require('rx');
var factories = function(username, password){
  var factories = [];
  var source = Rx.Observable.from(factories);
  var baseUrl = "<url>",
    factoriesOpt = {
      'uri': baseUrl+"<path>",
      'method': 'GET',
      'auth': {
        'user': username,
        'pass': password
      }
    },
    response="",
    request = require('request');

  var refresh = function(){
    request(factoriesOpt)
       .on('data', function(chunk){
         if(chunk)
           response += chunk;
       })
       .on('end', function(){
         factories = JSON.parse(response);
       })
  }
  return {
    subscribe : source.subscribe,
    factories : factories,
    refresh : refresh
  };
};
module.exports = factories;

但是当我运行它时，我得到了错误...

TypeError: this._subscribe is not a function
    at Object.Rx.Observable.observableProto.subscribe.observableProto.forEach [as subscribe] (/Users/<code location>/node_modules/rx/dist/rx.js:2034:19)
    at Object.<anonymous> (/Users/<code location>/index.js:5:11)
    at Module._compile (module.js:409:26)
    at Object.Module._extensions..js (module.js:416:10)
    at Module.load (module.js:343:32)
    at Function.Module._load (module.js:300:12)
    at Function.Module.runMain (module.js:441:10)
    at startup (node.js:139:18)
    at node.js:968:3

更新

如果我改变工厂使用...

var subscribe = function(callback){
  source.subscribe(callback);
};
...
return {
  subscribe : subscribe,
  factories : factories,
  refresh : refresh
};

它有效，但我想避免换行。