/* Copyright (c) 2014 Richard Rodger, MIT License */
'use strict';


var _    = require('underscore')
var amqp = require('amqplib/callback_api')

module.exports = function( options ) {
  var seneca = this
  var plugin = 'rabbitmq-transport'

  var so = seneca.options()

  options = seneca.util.deepextend(
    {
      rabbitmq: {
        type: 'rabbitmq',
        url: 'amqp://localhost',
        socketoptions: {}
      }
    },
    so.transport,
    options)

  var tu = seneca.export('transport/utils')

  seneca.add({role:'transport',hook:'listen',type:'rabbitmq'}, hook_listen_rabbitmq)
  seneca.add({role:'transport',hook:'client',type:'rabbitmq'}, hook_client_rabbitmq)


  function hook_listen_rabbitmq( args, done ) {
    var seneca         = this
    var type           = args.type
    var listen_options = seneca.util.clean(_.extend({},options[type],args))

    amqp.connect(listen_options.url, listen_options.socketoptions, function (error, connection) {
      if (error) return done(error)

      connection.createChannel(function (error, channel) {
        if (error) return done(error);

        channel.on('error', done);

        tu.listen_topics( seneca, args, listen_options, function ( topic ) {
          var acttopic = topic+'_act'

          seneca.log.debug('listen', 'subscribe', acttopic, listen_options, seneca)

          channel.assertQueue(acttopic)

          // Subscribe
          channel.consume(acttopic, on_message);

          var asserted = {}

          function on_message ( message ) {
            var content = message.content ? message.content.toString() : undefined
            var data = tu.parseJSON( seneca, 'listen-'+type, content )

            channel.ack(message)

            // Publish
            tu.handle_request( seneca, data, listen_options, function(out){
              if( null === out ) return;
              var outstr = tu.stringifyJSON( seneca, 'listen-'+type, out )

              var restopic = topic+'_res'+'/'+data.origin

              if (!asserted[restopic]) {
                channel.assertQueue(restopic)
                asserted[restopic] = true
              }

              channel.sendToQueue(restopic, new Buffer(outstr));
            })
          }
        });


        seneca.add('role:seneca,cmd:close',function( close_args, done ) {
          var closer = this
          channel.close();
          connection.close();
          closer.prior(close_args,done)
        })


        seneca.log.info('listen', 'open', listen_options, seneca)

        done()
      })
    })
  }


  function hook_client_rabbitmq( args, client_done ) {
    var seneca         = this
    var type           = args.type
    var client_options = seneca.util.clean(_.extend({},options[type],args))

    amqp.connect(client_options.url, client_options.socketoptions, function (error, connection) {
      if (error) return client_done(error)

      connection.createChannel(function (error, channel) {
        if (error) return client_done(error);

        tu.make_client( seneca, make_send, client_options, client_done )

        function make_send( spec, topic, send_done ) {
          var acttopic = topic+'_act'
          var restopic = topic+'_res'+'/'+seneca.id

          channel.on('error', send_done)

          channel.assertQueue(acttopic)
          channel.assertQueue(restopic)

          seneca.log.debug('client', 'subscribe', restopic, client_options, seneca)

          // Subscribe
          channel.consume(restopic, function ( message ) {
            var content = message.content ? message.content.toString() : undefined
            var input = tu.parseJSON(seneca,'client-'+type,content)
            
            channel.ack(message)
            
            tu.handle_response( seneca, input, client_options )
          })

          // Publish
          send_done( null, function ( args, done ) {
            var outmsg = tu.prepare_request( this, args, done )
            var outstr = tu.stringifyJSON( seneca, 'client-rabbitmq', outmsg )
            channel.sendToQueue(acttopic, new Buffer(outstr));
          })

          seneca.add('role:seneca,cmd:close',function( close_args, done ) {
            var closer = this
            channel.close();
            connection.close();
            closer.prior(close_args,done)
          })

        }
      })
    })
  }

  return {
    name: plugin,
  }
}

@@ -1247,16 +1247,54 @@
 c+'_act'
+%0A          var restopic = topic+'_res'
 %0A%0A      
@@ -1407,24 +1407,64 @@
 ue(acttopic)
+%0A          channel.assertQueue(restopic)
 %0A%0A          
@@ -1530,37 +1530,8 @@
 );%0A%0A
-          var asserted = %7B%7D%0A%0A
     
@@ -1978,214 +1978,8 @@
 t )%0A
-%0A              var restopic = topic+'_res'+'/'+data.origin%0A%0A              if (!asserted%5Brestopic%5D) %7B%0A                channel.assertQueue(restopic)%0A                asserted%5Brestopic%5D = true%0A              %7D%0A%0A
     
@@ -3049,22 +3049,8 @@
 res'
-+'/'+seneca.id
 %0A%0A  

Revert "use a different response queue for each client and topic"