/**
 * var connection = new Connection({ tag:        'groonga',
 *                                   host:       'localhost',
 *                                   port:       24224,
 *                                   listenPort: 10030 });
 *
 * // for REST APIs
 * connection.emitMessage(
 *   { command: 'search',
 *     query:   'foobar' },
 *   function callback(result) {
 *     // codes to handle results from the backend
 *   }
 * );
 *
 * // for socket.io APIs
 * connection.on('notification', function listener(messageBody) { ... });
 * // volatile message
 * connection.emitMessage({ command: 'reindex' });
 */

var EventEmitter = require('events').EventEmitter;
var socketIo = require('socket.io');
var fluent = require('fluent-logger');

var DEFAULT_FLUENT_TAG  = 'groonga';
var DEFAULT_FLUENT_HOST = 'localhost';
var DEFAULT_FLUENT_PORT = 24224;
var DEFAULT_LISTEN_PORT = 10030;

function Connection(params) {
  this._params = params || {};
  this._init();
}

Connection.prototype = new EventEmitter();

Connection.prototype._init = function() {
  this._callbacks = {};

  this._sender = (
    this._params.sender ||
    fluent.createFluentSender(this._params.tag || DEFAULT_FLUENT_TAG,
                              { host: this._params.host || DEFAULT_FLUENT_HOST,
                                port: this._params.port || DEFAULT_FLUENT_PORT })
  );
  this._socket = socketIo.listen(this._params.listenPort || DEFAULT_LISTEN_PORT);

  var self = this;
  this._socket.sockets.on('connection', function(socket) {
    socket.on((self._params.tag || DEFAULT_FLUENT_TAG)+ '.message',
              self._handleMessage.bind(self));
  });
  this._socket.set('log level', 1);
};

Connection.prototype._handleMessage = function(message) {
  var replyTo = message['reply_to'];
  if (replyTo) {
    this.emit('replyTo:' + replyTo, message.body);
  }
};

function createId() {
  return Date.now() + ':' + Math.round(Math.random() * 65536);
}

Connection.prototype.emitMessage = function(message, callback) {
  var id = createId();
  var message = {
    id:   id,
    body: message
  };
  if (callback) {
    this.once('replyTo:' + id, callback);
  }
  this._sender.emit('message', message);
};

export.Connection = Connection;

@@ -1077,16 +1077,113 @@
 s = %7B%7D;%0A
+  this._initSender();%0A  this._initReceiver();%0A%7D;%0A%0AConnection.prototype._initSender = function() %7B
 %0A  this.
@@ -1462,24 +1462,115 @@
  );%0A
-  this._socket =
+%7D;%0A%0AConnection.prototype._initReceiver = function() %7B%0A  this._receiver = (%0A    this._params.receiver %7C%7C%0A   
  soc
@@ -1633,29 +1633,14 @@
 ORT)
-;%0A
 %0A  
-var self = this;
+);%0A
 %0A  t
@@ -1644,22 +1644,24 @@
   this._
-socket
+receiver
 .sockets
@@ -1664,16 +1664,21 @@
 kets.on(
+%0A    
 'connect
@@ -1678,25 +1678,30 @@
 connection',
- 
+%0A    (
 function(soc
@@ -1707,24 +1707,26 @@
 cket) %7B%0A    
+  
 socket.on((s
@@ -1724,20 +1724,20 @@
 ket.on((
-self
+this
 ._params
@@ -1791,20 +1791,22 @@
         
-self
+  this
 ._handle
@@ -1822,19 +1822,36 @@
 ind(
-self
+this
 ));%0A  
-%7D
+  %7D).bind(this)%0A  
 );%0A 
@@ -1857,22 +1857,24 @@
   this._
-socket
+receiver
 .set('lo

Make receiver socket injectable