function EventStream(channel, options) {
  // call parent class constructor
  EventEmitter.call(this);
  var self = this;
  // set amqp channel
  this.channel = channel;
  // set event stream options
  this.options = {
    // library options
    encoding: 'utf8',
    // channel options
    prefetch: false,
    // exchange/queue options
    exchange: '',
    exclusive: false,
    durable: false,
    autoDelete: false,
    // message options
    persistent: false,
    expiration: undefined,
    noAck: true
  };
  extend(this.options, options);
  // by default generate a new queue
  this.queue = '';
  this.exchange = this.options.exchange;
  // set channel options
  function setChannelOptions(ch) {
    ch.on('close', bindEvent(self, 'close'));
    ch.on('error', bindEvent(self, 'error'));
    if (self.options.prefetch !== false) {
      ch.prefetch(options.prefetch);
    }
    // @todo set options of the channel
    return ch;
  }

  channel.then(setChannelOptions, bindEvent(self, 'error'));
}
/**
 * Base event stream class
 *
 * @param channel AMQP channel
 * @param options
 * @constructor
 */
Base event stream class
@param channel AMQP channel
@param options
@constructor
Base event stream class @param channel AMQP channel @param options @constructor
function EventStream(channel, options) {
  EventEmitter.call(this);
  var self = this;
  this.channel = channel;
  this.options = {
    encoding: 'utf8',
    prefetch: false,
    exchange: '',
    exclusive: false,
    durable: false,
    autoDelete: false,
    persistent: false,
    expiration: undefined,
    noAck: true
  };
  extend(this.options, options);
  this.queue = '';
  this.exchange = this.options.exchange;
  function setChannelOptions(ch) {
    ch.on('close', bindEvent(self, 'close'));
    ch.on('error', bindEvent(self, 'error'));
    if (self.options.prefetch !== false) {
      ch.prefetch(options.prefetch);
    }
    return ch;
  }
  channel.then(setChannelOptions, bindEvent(self, 'error'));
}
function EventStream ( channel , options ) { EventEmitter . call ( this ) ; var self = this ; this . channel = channel ; this . options = { encoding : 'utf8' , prefetch : false , exchange : '' , exclusive : false , durable : false , autoDelete : false , persistent : false , expiration : undefined , noAck : true } ; extend ( this . options , options ) ; this . queue = '' ; this . exchange = this . options . exchange ; function setChannelOptions ( ch ) { ch . on ( 'close' , bindEvent ( self , 'close' ) ) ; ch . on ( 'error' , bindEvent ( self , 'error' ) ) ; if ( self . options . prefetch !== false ) { ch . prefetch ( options . prefetch ) ; } return ch ; } channel . then ( setChannelOptions , bindEvent ( self , 'error' ) ) ; }
Base event stream class
 @param channel AMQP channel
 @param options
 @constructor
Base event stream class @param channel AMQP channel @param options @constructor
// call parent class constructor // set amqp channel // set event stream options // library options // channel options // exchange/queue options // message options // by default generate a new queue // set channel options // @todo set options of the channel