<commit_before>"use strict";

var Redis = require('redis');
var _ = require('underscore');
var util = require('util');
var Node = require('./node');

var Topology = module.exports = function(opts){

  var self = this;

  if(typeof opts === 'undefined'){
    opts = {};           
  }

  if(!opts.hasOwnProperty('redis')){
    opts.redis = {
      host: '127.0.0.1',
      port: 6379
    };
  }

  if(!opts.redis.hasOwnProperty('prefix')){
    opts.redis.prefix = '';
  } else {
    opts.redis.prefix = ':' + opts.redis.prefix;
  }

  var keys = {
    queue: opts.redis.prefix + 'queue'
  };

  var redis = Redis.createClient();

  this.running = false;

  // our topology
  this.nodes = {};

  // ids of all nodes (keys of this.nodes)
  this.ids = [];

  // nodes that listen to a given node
  this.sources = {};

  // nodes a given node listens to
  this.targets = {};

  this.keys = function(){
    return this.ids;
  };

  // add a node to the topology
  this.add = function(opts, methods, sources, done){
    var self = this;
    opts.id = String(opts.id);
    if(typeof methods === 'function'){
      done = methods;
      methods = {};
      sources = [];
    }

    if(typeof sources === 'function'){
      done = sources;
      sources = [];
    }
    this.ids.push(opts.id);
    this.nodes[opts.id] = new Node(opts, methods, function(err){
      _.each(sources, function(id){
        if(!self.targets.hasOwnProperty(id)){
          self.targets[id] = [];
        }
        if(!self.sources.hasOwnProperty(opts.id)){
          self.sources[opts.id] = [];
        }
        self.targets[id].push(opts.id);
        self.sources[opts.id].push(id);
      });
      return done();
    });
  };

  // remove a node from the topology
  this.remove = function(id, done){
    var self = this;

    if(typeof this.nodes[id] === 'undefined'){
      return done();
    }

    this.nodes[id].stop(function(){

      _.each(self.sources[id], function(source_id){
        self.targets[source_id] = _.without(self.targets[source_id], id);
      });
      delete self.sources[id];
      self.ids = _.without(self.ids, id);
      delete self.nodes[id];

      return done();
    });
  };

  this.set = function(id, key, value){

    if(!this.nodes.hasOwnProperty(id)){
      return;
    }

    if(!key){
      return;
    }

    this.nodes[id].set(key, value);

  };

  this.get = function(id, key){

    if(!this.nodes.hasOwnProperty(id)){
      return;
    }

    if(!key){
      return;
    }

    return this.nodes[id].get(key);

  };

  // return map of topology structure
  this.inspect = function(){
    var nodes = {};
    for(var id in this.nodes){
      nodes[id] = {
        id: id,
        sources: self.sources[id]<commit_msg>Return whole Node<commit_after>"use strict";

var Redis = require('redis');
var _ = require('underscore');
var util = require('util');
var Node = require('./node');

var Topology = module.exports = function(opts){

  var self = this;

  if(typeof opts === 'undefined'){
    opts = {};           
  }

  if(!opts.hasOwnProperty('redis')){
    opts.redis = {
      host: '127.0.0.1',
      port: 6379
    };
  }

  if(!opts.redis.hasOwnProperty('prefix')){
    opts.redis.prefix = '';
  } else {
    opts.redis.prefix = ':' + opts.redis.prefix;
  }

  var keys = {
    queue: opts.redis.prefix + 'queue'
  };

  var redis = Redis.createClient();

  this.running = false;

  // our topology
  this.nodes = {};

  // ids of all nodes (keys of this.nodes)
  this.ids = [];

  // nodes that listen to a given node
  this.sources = {};

  // nodes a given node listens to
  this.targets = {};

  this.keys = function(){
    return this.ids;
  };

  // add a node to the topology
  this.add = function(opts, methods, sources, done){
    var self = this;
    opts.id = String(opts.id);
    if(typeof methods === 'function'){
      done = methods;
      methods = {};
      sources = [];
    }

    if(typeof sources === 'function'){
      done = sources;
      sources = [];
    }
    this.ids.push(opts.id);
    this.nodes[opts.id] = new Node(opts, methods, function(err){
      _.each(sources, function(id){
        if(!self.targets.hasOwnProperty(id)){
          self.targets[id] = [];
        }
        if(!self.sources.hasOwnProperty(opts.id)){
          self.sources[opts.id] = [];
        }
        self.targets[id].push(opts.id);
        self.sources[opts.id].push(id);
      });
      return done();
    });
  };

  // remove a node from the topology
  this.remove = function(id, done){
    var self = this;

    if(typeof this.nodes[id] === 'undefined'){
      return done();
    }

    this.nodes[id].stop(function(){

      _.each(self.sources[id], function(source_id){
        self.targets[source_id] = _.without(self.targets[source_id], id);
      });
      delete self.sources[id];
      self.ids = _.without(self.ids, id);
      delete self.nodes[id];

      return done();
    });
  };

  this.set = function(id, key, value){

    if(!this.nodes.hasOwnProperty(id)){
      return;
    }

    if(!key){
      return;
    }

    this.nodes[id].set(key, value);

  };

  this.get = function(id, key){

    if(!this.nodes.hasOwnProperty(id)){
      return;
    }

    if(!key){
      return this.nodes[id];
    }

    return this.nodes[id].get(key);

  };

  // return map of topology structure
  this.inspect = function(){
    var nodes = {};
    for(var id in this.nodes){
      nodes[id] = {
        id: id,
        sources: self.sources[id]
{'fraction_non_alphanumeric': 0.1718093846996531, 'fraction_numerical': 0.0036516341062625525, 'mean_word_length': 2.2510385756676556, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2433341', 'n_tokens_mistral': 1881, 'n_tokens_neox': 1800, 'n_words': 380}