<commit_before>var util = require('util');
var PluginAbstract = require('./abstract.js');
var serviceLocator = require('../../service-locator');
var JanusError = require('../error');
var Promise = require('bluebird');

function PluginStreaming() {
  PluginStreaming.super_.apply(this, arguments);
  this.stream = null;
}

util.inherits(PluginStreaming, PluginAbstract);

/**
 * @param {Object} message
 * @returns {Promise}
 */
PluginStreaming.prototype.processMessage = function(message) {
  var janusMessage = message['janus'];

  if ('webrtcup' === janusMessage) {
    return this.subscribe(message);
  }
  return PluginStreaming.super_.prototype.processMessage.call(this, message);
};

/**
 * @param {Object} message
 * @returns {Promise}
 */
PluginStreaming.prototype.subscribe = function(message) {
  var plugin = this;
  return Promise.resolve(message)
    .then(function() {
      return serviceLocator.get('cm-api-client').subscribe(plugin.stream)
    })
    .then(function() {
      serviceLocator.get('streams').add(plugin.stream);
      serviceLocator.get('logger').info('Storing ' + plugin.stream + ' for ' + plugin);
      return message;
    })
    .catch(JanusError.Error, function(error) {
      serviceLocator.get('http-client').detach(plugin);
      throw new JanusError.Error('Cannot subscribe: ' + plugin.stream + ' error: ' + error.message, 490, message['transaction']);
    });
};

/**
 * @param {Object} message
 * @returns {Promise}
 */
PluginStreaming.prototype.publish = function(message) {
  var plugin = this;
  return Promise.resolve(message)
    .then(function() {
      return serviceLocator.get('cm-api-client').publish(plugin.stream)
    })
    .then(function() {
      serviceLocator.get('streams').add(plugin.stream);
      serviceLocator.get('logger').info('Storing ' + plugin.stream + ' for ' + plugin);
      return message;
    })
    .catch(JanusError.Error, function(error) {
      serviceLocator.get('http-client').detach(plugin);
      throw new JanusError.Error('Cannot publish: ' + plugin.stream + ' error: ' + error.message, 490, message['transaction']);
    });
};

PluginStreaming.prototype.removeStream = function() {
  if (this.stream) {
    serviceLocator.get('cm-api-client').removeStream(this.stream);
    var streams = serviceLocator.get('streams');
    if (streams.has(this.stream.id)) {
      streams.remove(this.stream);
    }
    this.stream = null;
  }
};

PluginStreaming.prototype.onRemove = function() {
  this.removeStream();
  PluginStreaming.super_.prototype.onRemove.call(this);
};

module.exports = PluginStreaming;
<commit_msg>Introduce getStream in Streaming plugin to unify no-stream exceptions<commit_after>var util = require('util');
var PluginAbstract = require('./abstract.js');
var serviceLocator = require('../../service-locator');
var JanusError = require('../error');
var Promise = require('bluebird');

function PluginStreaming() {
  PluginStreaming.super_.apply(this, arguments);
  this.stream = null;
}

util.inherits(PluginStreaming, PluginAbstract);

/**
 * @param {Object} message
 * @returns {Promise}
 */
PluginStreaming.prototype.processMessage = function(message) {
  var janusMessage = message['janus'];

  if ('webrtcup' === janusMessage) {
    return this.subscribe(message);
  }
  return PluginStreaming.super_.prototype.processMessage.call(this, message);
};

/**
 * @param {Object} message
 * @returns {Promise}
 */
PluginStreaming.prototype.subscribe = function(message) {
  var plugin = this;
  return Promise.resolve()
    .then(function() {
      return serviceLocator.get('cm-api-client').subscribe(plugin.getStream())
    })
    .then(function() {
      var stream = plugin.getStream();
      serviceLocator.get('streams').add(stream);
      serviceLocator.get('logger').info('Storing ' + stream + ' for ' + plugin);
      return message;
    })
    .catch(JanusError.Error, function(error) {
      serviceLocator.get('http-client').detach(plugin);
      throw new JanusError.Error('Cannot subscribe: ' + plugin.stream + ' for ' + plugin + 'error: ' + error.message, 490, message['transaction']);
    });
};

/**
 * @param {Object} message
 * @returns {Promise}
 */
PluginStreaming.prototype.publish = function(message) {
  var plugin = this;
  return Promise.resolve()
    .then(function() {
      return serviceLocator.get('cm-api-client').publish(plugin.getStream())
    })
    .then(function() {
      var stream = plugin.getStream();
      serviceLocator.get('streams').add(stream);
      serviceLocator.get('logger').info('Storing ' + stream + ' for ' + plugin);
      return message;
    })
    .catch(JanusError.Error, function(error) {
      serviceLocator.get('http-client').detach(plugin);
      throw new JanusError.Error('Cannot publish: ' + plugin.stream + ' for ' + plugin + ' error: ' + error.message, 490, message['transaction']);
    });
};

PluginStreaming.prototype.getStream = function() {
  if (null === this.stream) {
    throw new JanusError.Warning('No Stream found for ' + this);
  }
  return this.stream;
};

PluginStreaming.prototype.removeStream = function() {
  if (this.stream) {
    serviceLocator.get('cm-api-client').removeStream(this.stream);
    var streams = serviceLocator.get('streams');
    if (streams.has(this.stream.id)) {
      streams.remove(this.stream);
    }
    this.stream = null;
  }
};

PluginStreaming.prototype.onRemove = function() {
  this.removeStream();
  PluginStreaming.super_.prototype.onRemove.call(this);
};

module.exports = PluginStreaming;

{'fraction_non_alphanumeric': 0.16161249319048485, 'fraction_numerical': 0.002179044852006537, 'mean_word_length': 4.0118289353958145, 'pattern_counts': {'":': 0, '<': 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': '7365047', 'n_tokens_mistral': 1681, 'n_tokens_neox': 1652, 'n_words': 292}