'use strict'

var Promise = require('./promise')
var defer = require('./defer')
var amqplib = require('amqplib')
var inherits = require('util').inherits
var EventEmitter = require('events').EventEmitter

module.exports = Connection

inherits(Connection, EventEmitter)
/**
 * Create a connection to a RabbitMQ server.
 *
 * Once connected, it will emit a `connect` event.
 *
 * If the connection process fails, an `error` event will be emitted.
 *
 * @class
 * @param {String} url - The URL of the RabbitMQ Server. E.g. `amqp://127.0.0.1:5672`
 * @param {Object} options
 * @param {String} [options.exchange = 'loke-queue'] - Name of the exchange for scoping event names.
 */
function Connection (url, options) {
  EventEmitter.call(this)

  options = options || {}

  this.amqp = amqplib.connect(url)

  this.exchange = options.exchange || 'loke-queue'

  var self = this

  defer(
    this.amqp
    .then(
      function (client) {
        self.emit('connect')
      },
      function (err) {
        self.emit('error', err)
      }
    )
  )
}

/**
 * Get a channel and assert an exchange.
 * @private
 * @return {Promise}
 */
Connection.prototype._getPublisher = function () {
  if (this._publishChannel) return this._publishChannel
  var exchange = this.exchange
  var channelPromise = this.amqp
  .then(function (client) {
    return client.createChannel()
  })
  .then(function (ch) {
    return ch.assertExchange(exchange, 'topic', {
      durable: false,
      autoDelete: false
    })
    .then(function () {
      return ch
    })
  })

  return (this._publishChannel = channelPromise)
}

/**
 * Get or create the channel shared by all queues
 * @private
 * @return {Promise<amqplib.Channel>}
 */
Connection.prototype._getSharedChannel = function () {
  if (this._sharedChannel) return this._sharedChannel

  var channelPromise = this.amqp
  .then(function (client) {
    return client.createChannel()
  })

  return (this._sharedChannel = channelPromise)
}

/**
 * Publish an event to the topic exchange.
 *
 * The routing key is a dot separated name of the event.
 * Events can be subscribed to using wildcards at any part of the routing key.
 *
 *
 * @param  {String | String[]} routingKey - The event name. E.g. `user.create`. Also supports using ['user', 'create'] which is sometimes more convenient.
 * @param  {*} body - The data to send. This contains arbitrary JSON-serializable message data.
 * @param  {Object} [opts]
 * @param  {Object<String, String>} [opts.headers] - Metadata to send along with the message.
 * @return {Promise} - a promise resolved after send confirmation
 */
Connection.prototype.publish = function (routingKey, body, opts) {
  opts = opts || {}
  if (Array.isArray(routingKey)) {
    routingKey = routingKey.join('.')
  }

  var content = new Buffer(JSON.stringify(body))
  var exchange = this.exchange

  var options = {
    correlationId: process.domain && process.domain.cid || null,
    contentType: 'application/json',
    contentEncoding: 'utf-8',
    headers: opts.headers || {}
  }

  return this._getPublisher()
  .then(function (ch) {
    return ch.publish(
      exchange,
      routingKey,
      content,
      options
    )
  })
  .then(function () {
    return
  })
}

/**
 * Receive messages from a queue.
 *
 * For every {@link Connection.Message}, it will call `handler(message)`. Each message expects an acknowledgement
 * which is done by returning a value or a promise. If you throw an error or return a rejecting promise,
 * the message will be sent back into the queue.
 *
 * This method does not return a promise because it is asynchronous. Instead, it emits an 'error' event upon failure.
 *
 * @param  {String} routingKey - The event names to receive.
 * @param  {Object} [options]
 * @param  {String} [options.queueName = auto] - The name of the queue to receive events through. If not provided, then a new queue is created. If you want to share a queue between processes, simply use the same string here.
 * @param  {Boolean} [options.durable = true] - If true, the queue will survive RabbiqMQ broker restarts.
 * @param  {Boolean} [options.exclusive = false] - If true, scopes the queue to the connection.
 * @param  {Boolean} [options.autoDelete = false] - Automatically delete the queue when there are no consumers.
 * @param  {Number} [options.maxConcurrent = 100] - Maximum number of messages to receive which can be waiting acknowledgement. This is the maximum number of times that handler() will be called simultaneously.
 * @param  {Function} handler - The consumer function called with every {@link Connection.Message}.
 */
Connection.prototype.subscribe = function (routingKey, options, handler) {
  var channelPromise = this._getSharedChannel()
  var connection = this

  if (typeof options === 'function') {
    handler = options
    options = {}
  }

  options = options || {}

  var queueName = options.queueName
  var exchange = this.exchange

  var promise = channelPromise
  .then(function (ch) {
    return ch.checkQueue(queueName)
    .then(function () {
      ch.prefetch(options.maxConcurrent || 100)
      return ch.assertQueue(queueName, {
        exclusive: options.exclusive || false,
        autoDelete: options.autoDelete || false,
        durable: options.durable || true
      })
    })
    .then(function (ok) {
      return ok.queue
    })
    .then(function (queue) {
      return subscribe(connection, ch, exchange, queue, queueName, routingKey, handler)
    })
  })

  defer(promise)
}

/**
 * @typedef {Object} Connection.Message
 * @property {*} body
 * @property {Object<String, String>} headers
 * @property {String} routingKey
 * @property {String} exchange
 */
function Message (msg) {
  this.routingKey = msg.fields.routingKey
  this.exchange = msg.fields.exchange
  this.body = JSON.parse(msg.content)
  this.headers = msg.properties.headers || {}
}

function subscribe (connection, channel, exchange, queue, queueName, routingKey, handler) {
  function handleMessage (msg) {
    var message = new Message(msg)

    return Promise.resolve()
    .then(function () {
      return handler(message)
    })
    .then(
      function () {
        return channel.ack(msg)
      },
      function (err) {
        channel.reject(msg, true)
        throw err
      }
    )
  }

  return channel.bindQueue(queue, exchange, routingKey)
  .then(function () {
    var options = {noAck: false}
    return channel.consume(queue, function (msg) {
      defer(
        handleMessage(msg)
        .then(null, function (err) {
          connection.emit('error', err)
        })
      )
    }, options)
  })
}

@@ -4994,31 +4994,24 @@
 urn 
-ch.checkQueue(queueName
+Promise.resolve(
 )%0A  
@@ -5080,16 +5080,63 @@
 %7C%7C 100)%0A
+      // Create the queue if it doesn't exist:%0A
       re

Remove debug line