<commit_before>Meteor.Stream = function Stream(name, remote, callback) {
  if (remote instanceof Function) {
    callback = remote;
    remote = null;
  }

  EV.call(this);

  var self = this;
  var streamName = 'stream-' + name;
  var collection;
  var subscription;
  var subscriptionId;

  var connected = false;
  var pendingEvents = [];

  if (remote)
		collection = new Meteor.Collection(streamName, remote);
	else
		collection = new Meteor.Collection(streamName);

  self._emit = self.emit;

  collection.find({}).observe({
    "added": function(item) {
      if(item.type == 'subscriptionId') {
        subscriptionId = item._id;
        connected = true;
        pendingEvents.forEach(function(args) {
          self.emit.apply(self, args);
        });
        pendingEvents = [];
      } else {
        var context = {};
        context.subscriptionId = item.subscriptionId;
        context.userId = item.userId;
        self._emit.apply(context, item.args);    
      }
    }
  });

  if (remote)
		subscription = remote.subscribe(streamName, callback);
	else
		subscription = Meteor.subscribe(streamName, callback);

  self.emit = function emit() {
    if(connected) {
      Meteor.call(streamName, subscriptionId, arguments);
    } else {
      pendingEvents.push(arguments);
    }
  };

  self.close = function close() {
    subscription.stop();
  };
}

_.extend(Meteor.Stream.prototype, EV.prototype);
<commit_msg>enabled to close and reopen<commit_after>var collections = {};

Meteor.Stream = function Stream(name, remote, callback) {
  if (remote instanceof Function) {
    callback = remote;
    remote = null;
  }

  EV.call(this);

  var self = this;
  var streamName = 'stream-' + name;
  var collection;
  var subscription;
  var subscriptionId;

  var connected = false;
  var pendingEvents = [];

  if (collections[streamName] == null)
    if (remote)
			collection = new Meteor.Collection(streamName, remote);
		else
			collection = new Meteor.Collection(streamName);
  else
    collection = collections[streamName];

  self._emit = self.emit;

  collection.find({}).observe({
    "added": function(item) {
      if(item.type == 'subscriptionId') {
        subscriptionId = item._id;
        connected = true;
        pendingEvents.forEach(function(args) {
          self.emit.apply(self, args);
        });
        pendingEvents = [];
      } else {
        var context = {};
        context.subscriptionId = item.subscriptionId;
        context.userId = item.userId;
        self._emit.apply(context, item.args);    
      }
    }
  });

  if (remote)
		subscription = remote.subscribe(streamName, callback);
	else
		subscription = Meteor.subscribe(streamName, callback);

  self.emit = function emit() {
    if(connected) {
      Meteor.call(streamName, subscriptionId, arguments);
    } else {
      pendingEvents.push(arguments);
    }
  };

  self.close = function close() {
    subscription.stop();
  };
}

_.extend(Meteor.Stream.prototype, EV.prototype);

{'fraction_non_alphanumeric': 0.1278447121820616, 'fraction_numerical': 0.0, 'mean_word_length': 2.8369704749679077, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, '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': '1144774', 'n_tokens_mistral': 910, 'n_tokens_neox': 883, 'n_words': 192}