<commit_before>module.exports = Membership;

var EventEmitter = require('events').EventEmitter;
var util = require('util');
var Swim = require('happn-swim');
var Promise = require('bluebird');
var dface = require('dface');

var property = require('./property');
var getAddress = require('./get-address');

function Membership(opts) {
  property(this, 'log', opts.logger.createLogger('Membership'));
}

util.inherits(Membership, EventEmitter);

Membership.prototype.initialize = function(config, callback) {
  property(this, 'happn', this.happn);
  this.config = config;
  this.__defaults(callback);
};

Membership.prototype.bootstrap = Promise.promisify(function(callback) {
  var config, protocol, address, happnAddress, wait, _this = this;

  this.log.info('listening at %s', this.swimAddress);
  this.log.info('joining cluster \'%s\'', this.config.clusterName);

  config = this.config;
  protocol = this.happn.config.transport.mode;
  address = this.happn.server.address();
  if (address.address == '0.0.0.0') {
    happnAddress = getAddress();
  } else {
    happnAddress = address.address;
  }
  happnAddress += ':' + address.port;
  happnAddress = protocol + '://' + happnAddress;

  if (this.swim) this.swim.leave();

  property(this, 'swim', new Swim({
    local: {
      host: this.swimAddress,
      meta: {
        cluster: config.clusterName,
        happn: happnAddress
      }
    },
    disseminationFactor: config.disseminationFactor,
    interval: config.pingInterval,
    joinTimeout: config.joinTimeout,
    pingTimeout: config.pingTimeout,
    pingReqTimeout: config.pingReqTimeout,
    pingReqGroupSize: config.pingReqGroupSize,
    udp: config.udp
  }));

  wait = config.seed ? 0 : config.seedWait;

  setTimeout(function() {
    _this.swim.bootstrap(_this.config.hosts, function(error) {
      if (error) {
        if (error.name == 'JoinFailedError' && !_this.config.seed) {
          return callback(error);
          // seed member accepts failure to join cluster
        }
      }

      var doLog = false;

      function logMemberChange() {
        if (!doLog) return;
        _this.log.info('has %d other members', Object.keys(_this.members).length);
      }

      function addMember(member) {
        if (_this.members[member.host]) return;
        _this.members[member.host] = {
          happn: member.meta.happn
        };
        logMemberChange();
        _this.emit('add', {
          memberId: member.host,
          happnUrl: member.meta.happn
        });
      }

      function removeMember(member) {
        if (!_this.members[member.host]) return;
        delete _this.members[member.host];
        logMemberChange();
        _this.emit('remove', {
          memberId: member.host
        });
      }

      function onUpdate(member) {
        if (member.host == _this.swimAddress) return;
        if (member.meta.cluster !== config.clusterName) return;
        if (member.state == 0) return addMember(member);
        if (member.state == 1) return;
        if (member.state == 2) return removeMember(member);
      }

      var members = _this.swim.members();
      _this.log.info('has %d other members', members.length);

      members.forEach(addMember);

      doLog = true;

      _this.swim.on(Swim.EventType.Update, onUpdate);
      _this.swim.on(Swim.EventType.Change, onUpdate);

      callback();
    });
  }, wait);
});

Membership.prototype.stop = function(options, callback) {
  if (typeof options == 'function') callback = options;
  this.swim.leave();<commit_msg>clear listeners on stop<commit_after>module.exports = Membership;

var EventEmitter = require('events').EventEmitter;
var util = require('util');
var Swim = require('happn-swim');
var Promise = require('bluebird');
var dface = require('dface');

var property = require('./property');
var getAddress = require('./get-address');

function Membership(opts) {
  property(this, 'log', opts.logger.createLogger('Membership'));
}

util.inherits(Membership, EventEmitter);

Membership.prototype.initialize = function(config, callback) {
  property(this, 'happn', this.happn);
  this.config = config;
  this.__defaults(callback);
};

Membership.prototype.bootstrap = Promise.promisify(function(callback) {
  var config, protocol, address, happnAddress, wait, _this = this;

  this.log.info('listening at %s', this.swimAddress);
  this.log.info('joining cluster \'%s\'', this.config.clusterName);

  config = this.config;
  protocol = this.happn.config.transport.mode;
  address = this.happn.server.address();
  if (address.address == '0.0.0.0') {
    happnAddress = getAddress();
  } else {
    happnAddress = address.address;
  }
  happnAddress += ':' + address.port;
  happnAddress = protocol + '://' + happnAddress;

  if (this.swim) {
    this.swim.leave();
    this.swim.removeAllListeners();
  }

  property(this, 'swim', new Swim({
    local: {
      host: this.swimAddress,
      meta: {
        cluster: config.clusterName,
        happn: happnAddress
      }
    },
    disseminationFactor: config.disseminationFactor,
    interval: config.pingInterval,
    joinTimeout: config.joinTimeout,
    pingTimeout: config.pingTimeout,
    pingReqTimeout: config.pingReqTimeout,
    pingReqGroupSize: config.pingReqGroupSize,
    udp: config.udp
  }));

  wait = config.seed ? 0 : config.seedWait;

  setTimeout(function() {
    _this.swim.bootstrap(_this.config.hosts, function(error) {
      if (error) {
        if (error.name == 'JoinFailedError' && !_this.config.seed) {
          return callback(error);
          // seed member accepts failure to join cluster
        }
      }

      var doLog = false;

      function logMemberChange() {
        if (!doLog) return;
        _this.log.info('has %d other members', Object.keys(_this.members).length);
      }

      function addMember(member) {
        if (_this.members[member.host]) return;
        _this.members[member.host] = {
          happn: member.meta.happn
        };
        logMemberChange();
        _this.emit('add', {
          memberId: member.host,
          happnUrl: member.meta.happn
        });
      }

      function removeMember(member) {
        if (!_this.members[member.host]) return;
        delete _this.members[member.host];
        logMemberChange();
        _this.emit('remove', {
          memberId: member.host
        });
      }

      function onUpdate(member) {
        if (member.host == _this.swimAddress) return;
        if (member.meta.cluster !== config.clusterName) return;
        if (member.state == 0) return addMember(member);
        if (member.state == 1) return;
        if (member.state == 2) return removeMember(member);
      }

      var members = _this.swim.members();
      _this.log.info('has %d other members', members.length);

      members.forEach(addMember);

      doLog = true;

      _this.swim.on(Swim.EventType.Update, onUpdate);
      _this.swim.on(Swim.EventType.Change, onUpdate);

      callback();
    });
  }, wait);
});

Membership.prototype.stop = function(options, callback) {
  if (typeof options == 'function') callback = options;
  if (this.swim) {
    this.swim.leave();
    this.swim.removeAllListeners();
  }
{'fraction_non_alphanumeric': 0.14138171667829727, 'fraction_numerical': 0.0022330774598743894, 'mean_word_length': 2.9308831596269886, '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': '7345368', 'n_tokens_mistral': 2247, 'n_tokens_neox': 2192, 'n_words': 438}