<commit_before>_                 = require 'underscore'
os                = require 'os'
amqp              = require 'amqp'
redis             = require 'redis'
Presence          = require 'koding-rabbit-presence'
{EventEmitter}    = require 'events'
{Scrubber, Store} = require 'koding-dnode-protocol'

class Kommunicator extends EventEmitter

  constructor:(@kite, @publicKey, @serviceGenericName, @serviceUniqueName, @api, @host='localhost', @port=6380)->
    @callbackid = 0
    @waitingReplies = {}
    @kiteList = {}
    @commands = 
      'kite.who': (options, callback)->
        callback false, @getChannelNameForKite()

      'kite.join': (options, callback)->
        channelName = opts.args[0]
        kiteName    = opts.args[1]
        if not @kiteList[channelName]
          @kiteList[channelName] = []
        if kiteName not in @kiteList[channelName] 
          @kiteList[channelName].push kiteName
        if callback
          callback false, true

      'kite.consume': (options, callback)=>
        @mergenData.lpop @getChannelNameForGenericName(), (err, val)=>
          if not val
            return
          try
            val = JSON.parse val
          catch E
            console.log "couldnt parse json", val
            return
          # console.log "!!!!!!!!!!!!! lpop lpop >>>>>>", val
          @receivedCommand val.from, val.id, val.cmd, val.args

  getChannelNameForKite:(genericName, uniqueName)->
    gn = genericName or @serviceGenericName
    kn = uniqueName or @serviceUniqueName
    "#{@publicKey}-kite-#{gn}-#{kn}"

  getChannelNameForGenericName:(genericName)->
    if genericName
      genericName = "kite-#{genericName}"
    "#{@publicKey}-kite-#{genericName or @serviceGenericName}"

  getChannelNameForAll:()->
    "#{@publicKey}-kite-all"

  getGroupName:(groupName)->
    "#{@publicKey}-kite-kite-#{groupName}"

  receivedCommand: (from, cmdid, cmd, args)->
    console.log "receivedCommand", arguments
    if cmd.indexOf('reply-') > -1
      cmdid = cmd.split('reply-')[1]
      #console.log ">>>> reply cmd", cmd, args
      @waitingReplies[cmdid]?.call @, false, {args:args, from:from}
      return

    fn = @commands[cmd]
    if fn
      ret = fn.call @, {args:args, from:from}, (err, ret)=>
        @sendCommand(from, "reply-#{cmdid}", [err, ret])
      return

    if @api.propertyIsEnumerable(cmd) and 'function' is typeof @api[cmd]
      fn = @api[cmd]
      fn.call @kite, {args:args, from:from}, (err, ret)=>
        @sendCommand(from, "reply-#{cmdid}", [err, ret])
      return

    @sendCommand(from, "reply-#{cmdid}", ['function not found', false])


  queueCommand:(channel, cmd, args, callback=null)->

    if not channel
      channel = @getChannelNameForGenericName()
    else
      channel = @getChannelNameForGenericName(channel)

    # sendCommand broadcasts, with this one, we first push to queue
    # then broadcast and one of them consumes
    @callbackid++
    data = {
      cmd: cmd,
      id: @callbackid,
      args: args,
      from: @getChannelNameForKite()
    }

    if cb
      @waitingReplies[@callbackid] = cb

    datastr = JSON.stringify(data)
    @mergenData.rpush channel, datastr, ()=>
      console.log "pushed to ", channel, datastr
      @sendCommand channel, "kite.consume", datastr

  sendCommand: (channel, cmd, args, callback=null)->
    # this broadcasts command to the said channel
    @callbackid++
    data = {
      cmd: cmd,
      id: @callbackid,
      args: args,
      from: @getChannelNameForKite()
    }
    if cb
      @waitingReplies[@callbackid] = cb
    @mergenData.publish channel, JSON.stringify(data)

  connectToMessageBus: (cb)=>

    # we have a second connection because, redis client in node doesnt send any commands
    # when it enters subscribe mode, though redis docs says you can publish if you are in subcribe
    # mode
    @mergenData = redis.createClient(@port, @host, {socket_nodelay:false})
    @mergenData.on "error", (err)->
      console.log "error", err

    data = {}
    data.pid = process.pid
    data.hostname = os.hostname()
    data.serviceGenericName = @serviceGenericName
    data.serviceUniqueName = @serviceUniqueName
    data.startedAt = Date.now()

    @mergenData.hset "#{@publicKey}-kites", "#{@serviceGenericName}-#{@serviceUniqueName}", data

    @sendCommand @getChannelNameForGenericName(), "kite.who", [], (err, reply)->
      @commands['kite.join'].call(@, {args:[@getChannelNameForGenericName(),reply.from], from:reply.from})

    @sendCommand @getChannelNameForGenericName(), "kite.join", [@getChannelNameForGenericName(), @getChannelNameForKite()]

    @kite.db = @mergenData

    @mergen = redis.createClient(@port, @host, {socket_nodelay:false})
    @mergen.on "error", (err)->
      console.log("Error ",err)

    subscribeToChannelNames = [@getChannelNameForAll(), @getChannelNameForGenericName(), @getChannelNameForKite()]

    for chan in subscribeToChannelNames
      @mergen.subscribe chan

    @mergen.on "message", (channel, message)=>
      message = JSON.parse message
      @receivedCommand message.from, message.id, message.cmd, message.args

    # we have 3 channels to subscribe on startup,
    # we have to wait for all before we emit ready
    @mergen.on "subscribe", (channel, count)->
      subscribeToChannelNames = _.without(subscribeToChannelNames, channel)
      console.log ">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> connected to channel", channel, subscribeToChannelNames
      

    startQueueChecker = ()=>
      startQueueCheckerInterval = setInterval ()=>
        @commands['kite.consume']()
      , 10

    subscribeCheck = setInterval ()=>
      if subscribeToChannelNames.length is 0
        clearInterval subscribeCheck
        startQueueChecker()
        cb()
    , 10
    @connected = true


module.exports = class Kite extends EventEmitter
  {parse, stringify} = JSON
  Object.defineProperty Object.prototype, 'throwError', value:(it)-> throw it

  @throwIt = (it) ->
    console.error it  if it?
    process.kill process.pid, 'SIGTERM'

  serializableError =(err)->
    message  : err.message
    stack    : err.stack

  group: (groupName, cmd, args, callback)->
    # console.log "group:::", groupName, cmd, args, callback
    # console.log "publishing to:", @communicator.getGroupName(groupName)
    @communicator.sendCommand @communicator.getGroupName(groupName), cmd, args, callback

  everyone: (cmd, args, callback)->
    @communicator.sendCommand @communicator.getChannelNameForAll(), cmd, args, callback

  all: (cmd, args, callback)->
    @communicator.sendCommand @communicator.getChannelNameForGenericName(), cmd, args, callback

  one: (chanName, cmd, args, callback)->
    channame = chanName
    @communicator.sendCommand channame, cmd, args, callback

  queue: (channel, cmd, args, callback)->
    @communicator.queueCommand channel, cmd, args, callback


  constructor:(@kiteName, @api, @publicKey)->
    if not @publicKey
      throw new Error("public key not set")
    [@api, @kiteName] = [kiteName, api] unless api
    @kiteName or= 'unknown'
    @localStore = new Store
    @remoteStore = new Store
    @channels = {}
    @clients = {} # a hash of clients by the secret channel ids they are using.
    @requesters = {}

  handleError:(routingKey, callbackId='uncaughtException', err)->
    err = serializableError(err)
    @fetchBrokerExchange (brokerExchange)->
      brokerExchange.publish routingKey, stringify(
        arguments : [err]
        method    : callbackId
        callbacks : {}
      ),{
        contentType: 'application/json'
      }

  handleRequest:(scrubbed, routingKey, {headers, deliveryInfo}, exchangeName = "broker")->
    authenticatedUsername = @clients[routingKey]
    return @handleResponse routingKey, 'pong', []  if scrubbed.method is 'ping'
    authenticatedUsername = if authenticatedUsername? then authenticatedUsername else routingKey

    @unscrubMessage routingKey, authenticatedUsername, scrubbed, (unscrubbed)=>
      [data, callback] = unscrubbed
      if data?
        data.withArgs?.username = authenticatedUsername
        {method, withArgs} = data
      withArgs or= {}
      if @api.propertyIsEnumerable(method) and 'function' is typeof @api[method]
        withArgs.authenticatedUser = authenticatedUsername
        fn = @api[method]
        if fn
            fn.call @, withArgs, (err, msg)=>
              @handleResponse routingKey, callback.id, [].slice.call(arguments), exchangeName
      else if callback = @localStore.get(scrubbed.method)
        callback unscrubbed...
      else
        @handleError routingKey, callback?.id, {
          message: "No such method! #{method}"
        }

  unscrubMessage:(routingKey, authenticatedUser, scrubbed, callback)->
    # duck-type for dnode protocol
    unless scrubbed.arguments? and scrubbed.callbacks?
      callback [scrubbed, ->] # a vanilla object; callback is noop
    else
      scrubber = new Scrubber @localStore
      unscrubbed = scrubber.unscrub scrubbed, (callbackId)=>
        storeKey = "#{authenticatedUser}#{callbackId}"
        unless @remoteStore.has(storeKey)
          @remoteStore.add storeKey, (args...)=>
            @handleResponse routingKey, callbackId, args
        @remoteStore.get storeKey
      callback unscrubbed

  handleResponse:(routingKey, callbackId, args, exchangeName = "broker")->
    scrubber = new Scrubber @localStore
    scrubber.scrub args, =>
      out = scrubber.toDnodeProtocol()
      out.method = callbackId
      out.pid = process.pid
      @fetchExchangeByName exchangeName, (exchange)->
        exchange.publish routingKey, stringify(out)

  handleMessage:(message, headers, deliveryInfo)->
    {routingKey} = deliveryInfo
    messageData = (try parse message.data) or message
    # catch e then @emit 'error', e
    switch routingKey
      when 'auth.join'  then @join messageData
      when 'auth.leave' then @leave messageData
      when 'auth.who' then @who messageData
      else
        if routingKey of @clients
          @handleRequest messageData, routingKey, {headers, deliveryInfo}
        else
          console.log "Unknown routing key: #{routingKey}.  Dropping message on the floor."

  getKiteId:=> "#{@publicKey}-#{process.pid}|#{os.hostname().replace /\./g, '_'}"

  getServiceType:-> "kite-#{@kiteName}"

  getRabbitMqResourceName:-> "#{@getServiceType()}-#{@getKiteId()}"

  getKiteExchangeOptions:->
    type        : 'fanout'
    autoDelete  : yes
    durable     : no

  fetchExchange:(name, options, callback)->
    exchangeName  = "#{name}Exchange"
    readyEvent    = "#{exchangeName}Ready"
    if @[exchangeName] is null
      @once readyEvent, => @fetchBrokerExchange callback
    else unless @brokerExchange?
      @[exchangeName] = null
      @connection.exchange name, options,
        (exchange)=>
          @[exchangeName] = exchange
          @emit readyEvent
          callback exchange
    else callback @[exchangeName]

  fetchAuthExchange:(callback)->
    @fetchExchange 'auth', {noDeclare:yes}, callback

  fetchBrokerExchange:(callback)->
    @fetchExchange 'broker', {noDeclare: yes}, callback

  fetchExchangeByName:(name, callback)->
    switch name
      when 'broker'     then @fetchBrokerExchange callback
      when 'auth'       then @fetchAuthExchange callback

  leave:(messageData)->
    {routingKey} = messageData
    delete @clients[routingKey]
    @api._leave? messageJson

  join:(messageData)->
    #console.log "auth.join", messageData
    {routingKey, username, correlationName, serviceGenericName} = messageData
    console.log "#{username}@#{correlationName} has joined to service #{serviceGenericName}"
    @clients[routingKey] = username
    @api._join? messageJson
    @handleResponse routingKey, 'ready', [@getRabbitMqResourceName()]

  createPresenceMemberKey:->
    serviceGenericName  = @getServiceType()
    serviceUniqueName   = @getRabbitMqResourceName()
    "serviceType.kite.serviceGenericName.#{serviceGenericName}.serviceUniqueName.#{serviceUniqueName}"

  establishPresence:->
    console.log "establishPresence", @createPresenceMemberKey()
    @presence = new Presence {
      @connection
      exchange    : 'services-presence'
      member      : @createPresenceMemberKey()
    }
    @presence.announce()

  registerSelf:(cb)=>
    {config} = this
    {host, port, protocol, login, password, vhost, heartbeat} = config.amqp

    if @registerSelfCalled
      throw new Error "double call for registerself"
    @registerSelfCalled = true

    port              ?= 5672
    protocol          ?= 'amqp:'
    options           = {host, port, protocol} # but not vhost, because of a bug in node-amqp
    options.login     = login     if login?
    options.password  = password  if password?
    options.vhost     = vhost     if vhost?
    # without heartbeat it crashes randomly, so
    options.heartbeat = heartbeat if heartbeat?
    @connection       = amqp.createConnection options
    @connection.on 'error', @emit.bind this, 'error'
    rabbitMqResourceName = @getRabbitMqResourceName()
    @connection.on 'ready', =>
      @establishPresence()
      @connection.exchange rabbitMqResourceName, @getKiteExchangeOptions(), (exchange)=>
        @connection.queue '', {exclusive: yes}, (queue)=>
          queue.bind exchange, ''
          queue.on 'queueBindOk', =>
            console.log "#{@kiteName}: Kite is connected to system"
            queue.subscribe (message, headers, deliveryInfo)=>
              @handleMessage message, headers, deliveryInfo
          cb()

  writePid:(pidPath)->
    if pidPath
      require("fs").writeFile pidPath,"#{process.pid}",(err)->
        if err?
          console.log "[WARN] Can't write pid to #{pidPath}:  monit can't watch this process. err:#{err.message}"

  run:(@config, @key, @manifest)->
    {pidPath, logPath, host, username, password, @apiUri} = config

    if @runSelfCalled
      throw new Error "double call for run"

    @runSelfCalled = true

    @writePid pidPath
    rabbitReady = false
    @registerSelf ()->
      rabbitReady = true

    hp = @config.messagebus.url.split(':')
    mergen_host = hp[0] 
    mergen_port = hp[1]

    @communicator = new Kommunicator(@, @key, @getServiceType(), 
        @getRabbitMqResourceName(), @api, mergen_host, mergen_port)

    @communicator.connectToMessageBus ()=>
      checkrabbit = setInterval ()=>
        if rabbitReady
          clearInterval checkrabbit
          @emit 'kite-ready'
      , 10

  tellKite:(kite, callback)=>
    args = [kite, callback]
    {method, kiteName} = kite
    if kiteName == @kiteName
      console.log 'Self-calling a kite is not recommended! Now dropping message'  
    else
      @request kiteName, method, args

  request :(kiteName, method, args)->
    @scrubKiteMessage method, args, (scrubbed) =>
      messageString = JSON.stringify scrubbed
      @sendMessageToAnotherKite kiteName, messageString
  
  sendMessageToAnotherKite:(kiteName, messsage)->
    @fetchExchangeByName 'kite-intra', (kite2kite)=>
      kite2kite.publish @kiteRoutingKeyPrefix + kiteName, messsage

  scrubKiteMessage:(method, args, callback) ->
    scrubber = new Scrubber @localStore
    scrubber.scrub args, =>
      scrubbed = scrubber.toDnodeProtocol()
      scrubbed.method or= method
      scrubbed.name or= @kiteName
      callback scrubbed

  receiveMessageFromAnotherKite:(message, headers, deliveryInfo)->
    messageData = (try message.data.toString()) or message
    scrubbed = JSON.parse messageData
    routingKey = @kiteRoutingKeyPrefix + scrubbed.name
    @handleRequest scrubbed, routingKey, {headers, deliveryInfo}, 'kite-intra'

  writePid:(pidPath)->
    if pidPath
      require("fs").writeFile pidPath,"#{process.pid}",(err)->
        if err?
          console.log "[WARN] Can't write pid to #{pidPath}:  monit can't watch this process. err:#{err.message}"
<commit_msg>camel case<commit_after>_                 = require 'underscore'
os                = require 'os'
amqp              = require 'amqp'
redis             = require 'redis'
Presence          = require 'koding-rabbit-presence'
{EventEmitter}    = require 'events'
{Scrubber, Store} = require 'koding-dnode-protocol'

class Kommunicator extends EventEmitter

  constructor:(@kite, @publicKey, @serviceGenericName, @serviceUniqueName, @api, @host='localhost', @port=6380)->
    @callbackid = 0
    @waitingReplies = {}
    @kiteList = {}
    @commands = 
      'kite.who': (options, callback)->
        callback false, @getChannelNameForKite()

      'kite.join': (options, callback)->
        channelName = opts.args[0]
        kiteName    = opts.args[1]
        if not @kiteList[channelName]
          @kiteList[channelName] = []
        if kiteName not in @kiteList[channelName] 
          @kiteList[channelName].push kiteName
        if callback
          callback false, true

      'kite.consume': (options, callback)=>
        @mergenData.lpop @getChannelNameForGenericName(), (err, val)=>
          if not val
            return
          try
            val = JSON.parse val
          catch E
            console.log "couldnt parse json", val
            return
          # console.log "!!!!!!!!!!!!! lpop lpop >>>>>>", val
          @receivedCommand val.from, val.id, val.cmd, val.args

  getChannelNameForKite:(genericName, uniqueName)->
    gn = genericName or @serviceGenericName
    kn = uniqueName or @serviceUniqueName
    "#{@publicKey}-kite-#{gn}-#{kn}"

  getChannelNameForGenericName:(genericName)->
    if genericName
      genericName = "kite-#{genericName}"
    "#{@publicKey}-kite-#{genericName or @serviceGenericName}"

  getChannelNameForAll:()->
    "#{@publicKey}-kite-all"

  getGroupName:(groupName)->
    "#{@publicKey}-kite-kite-#{groupName}"

  receivedCommand: (from, cmdId, cmd, args)->
    console.log "receivedCommand", arguments
    if cmd.indexOf('reply-') > -1
      cmdId = cmd.split('reply-')[1]
      #console.log ">>>> reply cmd", cmd, args
      @waitingReplies[cmdId]?.call @, false, {args:args, from:from}
      return

    fn = @commands[cmd]
    if fn
      ret = fn.call @, {args:args, from:from}, (err, ret)=>
        @sendCommand(from, "reply-#{cmdId}", [err, ret])
      return

    if @api.propertyIsEnumerable(cmd) and 'function' is typeof @api[cmd]
      fn = @api[cmd]
      fn.call @kite, {args:args, from:from}, (err, ret)=>
        @sendCommand(from, "reply-#{cmdId}", [err, ret])
      return

    @sendCommand(from, "reply-#{cmdId}", ['function not found', false])


  queueCommand:(channel, cmd, args, callback=null)->

    if not channel
      channel = @getChannelNameForGenericName()
    else
      channel = @getChannelNameForGenericName(channel)

    # sendCommand broadcasts, with this one, we first push to queue
    # then broadcast and one of them consumes
    @callbackid++
    data = {
      cmd: cmd,
      id: @callbackid,
      args: args,
      from: @getChannelNameForKite()
    }

    if cb
      @waitingReplies[@callbackid] = cb

    datastr = JSON.stringify(data)
    @mergenData.rpush channel, datastr, ()=>
      console.log "pushed to ", channel, datastr
      @sendCommand channel, "kite.consume", datastr

  sendCommand: (channel, cmd, args, callback=null)->
    # this broadcasts command to the said channel
    @callbackid++
    data = {
      cmd: cmd,
      id: @callbackid,
      args: args,
      from: @getChannelNameForKite()
    }
    if cb
      @waitingReplies[@callbackid] = cb
    @mergenData.publish channel, JSON.stringify(data)

  connectToMessageBus: (cb)=>

    # we have a second connection because, redis client in node doesnt send any commands
    # when it enters subscribe mode, though redis docs says you can publish if you are in subcribe
    # mode
    @mergenData = redis.createClient(@port, @host, {socket_nodelay:false})
    @mergenData.on "error", (err)->
      console.log "error", err

    data = {}
    data.pid = process.pid
    data.hostname = os.hostname()
    data.serviceGenericName = @serviceGenericName
    data.serviceUniqueName = @serviceUniqueName
    data.startedAt = Date.now()

    @mergenData.hset "#{@publicKey}-kites", "#{@serviceGenericName}-#{@serviceUniqueName}", data

    @sendCommand @getChannelNameForGenericName(), "kite.who", [], (err, reply)->
      @commands['kite.join'].call(@, {args:[@getChannelNameForGenericName(),reply.from], from:reply.from})

    @sendCommand @getChannelNameForGenericName(), "kite.join", [@getChannelNameForGenericName(), @getChannelNameForKite()]

    @kite.db = @mergenData

    @mergen = redis.createClient(@port, @host, {socket_nodelay:false})
    @mergen.on "error", (err)->
      console.log("Error ",err)

    subscribeToChannelNames = [@getChannelNameForAll(), @getChannelNameForGenericName(), @getChannelNameForKite()]

    for chan in subscribeToChannelNames
      @mergen.subscribe chan

    @mergen.on "message", (channel, message)=>
      message = JSON.parse message
      @receivedCommand message.from, message.id, message.cmd, message.args

    # we have 3 channels to subscribe on startup,
    # we have to wait for all before we emit ready
    @mergen.on "subscribe", (channel, count)->
      subscribeToChannelNames = _.without(subscribeToChannelNames, channel)
      console.log ">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> connected to channel", channel, subscribeToChannelNames
      

    startQueueChecker = ()=>
      startQueueCheckerInterval = setInterval ()=>
        @commands['kite.consume']()
      , 10

    subscribeCheck = setInterval ()=>
      if subscribeToChannelNames.length is 0
        clearInterval subscribeCheck
        startQueueChecker()
        cb()
    , 10
    @connected = true


module.exports = class Kite extends EventEmitter
  {parse, stringify} = JSON
  Object.defineProperty Object.prototype, 'throwError', value:(it)-> throw it

  @throwIt = (it) ->
    console.error it  if it?
    process.kill process.pid, 'SIGTERM'

  serializableError =(err)->
    message  : err.message
    stack    : err.stack

  group: (groupName, cmd, args, callback)->
    # console.log "group:::", groupName, cmd, args, callback
    # console.log "publishing to:", @communicator.getGroupName(groupName)
    @communicator.sendCommand @communicator.getGroupName(groupName), cmd, args, callback

  everyone: (cmd, args, callback)->
    @communicator.sendCommand @communicator.getChannelNameForAll(), cmd, args, callback

  all: (cmd, args, callback)->
    @communicator.sendCommand @communicator.getChannelNameForGenericName(), cmd, args, callback

  one: (chanName, cmd, args, callback)->
    channame = chanName
    @communicator.sendCommand channame, cmd, args, callback

  queue: (channel, cmd, args, callback)->
    @communicator.queueCommand channel, cmd, args, callback


  constructor:(@kiteName, @api, @publicKey)->
    if not @publicKey
      throw new Error("public key not set")
    [@api, @kiteName] = [kiteName, api] unless api
    @kiteName or= 'unknown'
    @localStore = new Store
    @remoteStore = new Store
    @channels = {}
    @clients = {} # a hash of clients by the secret channel ids they are using.
    @requesters = {}

  handleError:(routingKey, callbackId='uncaughtException', err)->
    err = serializableError(err)
    @fetchBrokerExchange (brokerExchange)->
      brokerExchange.publish routingKey, stringify(
        arguments : [err]
        method    : callbackId
        callbacks : {}
      ),{
        contentType: 'application/json'
      }

  handleRequest:(scrubbed, routingKey, {headers, deliveryInfo}, exchangeName = "broker")->
    authenticatedUsername = @clients[routingKey]
    return @handleResponse routingKey, 'pong', []  if scrubbed.method is 'ping'
    authenticatedUsername = if authenticatedUsername? then authenticatedUsername else routingKey

    @unscrubMessage routingKey, authenticatedUsername, scrubbed, (unscrubbed)=>
      [data, callback] = unscrubbed
      if data?
        data.withArgs?.username = authenticatedUsername
        {method, withArgs} = data
      withArgs or= {}
      if @api.propertyIsEnumerable(method) and 'function' is typeof @api[method]
        withArgs.authenticatedUser = authenticatedUsername
        fn = @api[method]
        if fn
            fn.call @, withArgs, (err, msg)=>
              @handleResponse routingKey, callback.id, [].slice.call(arguments), exchangeName
      else if callback = @localStore.get(scrubbed.method)
        callback unscrubbed...
      else
        @handleError routingKey, callback?.id, {
          message: "No such method! #{method}"
        }

  unscrubMessage:(routingKey, authenticatedUser, scrubbed, callback)->
    # duck-type for dnode protocol
    unless scrubbed.arguments? and scrubbed.callbacks?
      callback [scrubbed, ->] # a vanilla object; callback is noop
    else
      scrubber = new Scrubber @localStore
      unscrubbed = scrubber.unscrub scrubbed, (callbackId)=>
        storeKey = "#{authenticatedUser}#{callbackId}"
        unless @remoteStore.has(storeKey)
          @remoteStore.add storeKey, (args...)=>
            @handleResponse routingKey, callbackId, args
        @remoteStore.get storeKey
      callback unscrubbed

  handleResponse:(routingKey, callbackId, args, exchangeName = "broker")->
    scrubber = new Scrubber @localStore
    scrubber.scrub args, =>
      out = scrubber.toDnodeProtocol()
      out.method = callbackId
      out.pid = process.pid
      @fetchExchangeByName exchangeName, (exchange)->
        exchange.publish routingKey, stringify(out)

  handleMessage:(message, headers, deliveryInfo)->
    {routingKey} = deliveryInfo
    messageData = (try parse message.data) or message
    # catch e then @emit 'error', e
    switch routingKey
      when 'auth.join'  then @join messageData
      when 'auth.leave' then @leave messageData
      when 'auth.who' then @who messageData
      else
        if routingKey of @clients
          @handleRequest messageData, routingKey, {headers, deliveryInfo}
        else
          console.log "Unknown routing key: #{routingKey}.  Dropping message on the floor."

  getKiteId:=> "#{@publicKey}-#{process.pid}|#{os.hostname().replace /\./g, '_'}"

  getServiceType:-> "kite-#{@kiteName}"

  getRabbitMqResourceName:-> "#{@getServiceType()}-#{@getKiteId()}"

  getKiteExchangeOptions:->
    type        : 'fanout'
    autoDelete  : yes
    durable     : no

  fetchExchange:(name, options, callback)->
    exchangeName  = "#{name}Exchange"
    readyEvent    = "#{exchangeName}Ready"
    if @[exchangeName] is null
      @once readyEvent, => @fetchBrokerExchange callback
    else unless @brokerExchange?
      @[exchangeName] = null
      @connection.exchange name, options,
        (exchange)=>
          @[exchangeName] = exchange
          @emit readyEvent
          callback exchange
    else callback @[exchangeName]

  fetchAuthExchange:(callback)->
    @fetchExchange 'auth', {noDeclare:yes}, callback

  fetchBrokerExchange:(callback)->
    @fetchExchange 'broker', {noDeclare: yes}, callback

  fetchExchangeByName:(name, callback)->
    switch name
      when 'broker'     then @fetchBrokerExchange callback
      when 'auth'       then @fetchAuthExchange callback

  leave:(messageData)->
    {routingKey} = messageData
    delete @clients[routingKey]
    @api._leave? messageJson

  join:(messageData)->
    #console.log "auth.join", messageData
    {routingKey, username, correlationName, serviceGenericName} = messageData
    console.log "#{username}@#{correlationName} has joined to service #{serviceGenericName}"
    @clients[routingKey] = username
    @api._join? messageJson
    @handleResponse routingKey, 'ready', [@getRabbitMqResourceName()]

  createPresenceMemberKey:->
    serviceGenericName  = @getServiceType()
    serviceUniqueName   = @getRabbitMqResourceName()
    "serviceType.kite.serviceGenericName.#{serviceGenericName}.serviceUniqueName.#{serviceUniqueName}"

  establishPresence:->
    console.log "establishPresence", @createPresenceMemberKey()
    @presence = new Presence {
      @connection
      exchange    : 'services-presence'
      member      : @createPresenceMemberKey()
    }
    @presence.announce()

  registerSelf:(cb)=>
    {config} = this
    {host, port, protocol, login, password, vhost, heartbeat} = config.amqp

    if @registerSelfCalled
      throw new Error "double call for registerself"
    @registerSelfCalled = true

    port              ?= 5672
    protocol          ?= 'amqp:'
    options           = {host, port, protocol} # but not vhost, because of a bug in node-amqp
    options.login     = login     if login?
    options.password  = password  if password?
    options.vhost     = vhost     if vhost?
    # without heartbeat it crashes randomly, so
    options.heartbeat = heartbeat if heartbeat?
    @connection       = amqp.createConnection options
    @connection.on 'error', @emit.bind this, 'error'
    rabbitMqResourceName = @getRabbitMqResourceName()
    @connection.on 'ready', =>
      @establishPresence()
      @connection.exchange rabbitMqResourceName, @getKiteExchangeOptions(), (exchange)=>
        @connection.queue '', {exclusive: yes}, (queue)=>
          queue.bind exchange, ''
          queue.on 'queueBindOk', =>
            console.log "#{@kiteName}: Kite is connected to system"
            queue.subscribe (message, headers, deliveryInfo)=>
              @handleMessage message, headers, deliveryInfo
          cb()

  writePid:(pidPath)->
    if pidPath
      require("fs").writeFile pidPath,"#{process.pid}",(err)->
        if err?
          console.log "[WARN] Can't write pid to #{pidPath}:  monit can't watch this process. err:#{err.message}"

  run:(@config, @key, @manifest)->
    {pidPath, logPath, host, username, password, @apiUri} = config

    if @runSelfCalled
      throw new Error "double call for run"

    @runSelfCalled = true

    @writePid pidPath
    rabbitReady = false
    @registerSelf ()->
      rabbitReady = true

    hp = @config.messagebus.url.split(':')
    mergen_host = hp[0] 
    mergen_port = hp[1]

    @communicator = new Kommunicator(@, @key, @getServiceType(), 
        @getRabbitMqResourceName(), @api, mergen_host, mergen_port)

    @communicator.connectToMessageBus ()=>
      checkrabbit = setInterval ()=>
        if rabbitReady
          clearInterval checkrabbit
          @emit 'kite-ready'
      , 10

  tellKite:(kite, callback)=>
    args = [kite, callback]
    {method, kiteName} = kite
    if kiteName == @kiteName
      console.log 'Self-calling a kite is not recommended! Now dropping message'  
    else
      @request kiteName, method, args

  request :(kiteName, method, args)->
    @scrubKiteMessage method, args, (scrubbed) =>
      messageString = JSON.stringify scrubbed
      @sendMessageToAnotherKite kiteName, messageString
  
  sendMessageToAnotherKite:(kiteName, messsage)->
    @fetchExchangeByName 'kite-intra', (kite2kite)=>
      kite2kite.publish @kiteRoutingKeyPrefix + kiteName, messsage

  scrubKiteMessage:(method, args, callback) ->
    scrubber = new Scrubber @localStore
    scrubber.scrub args, =>
      scrubbed = scrubber.toDnodeProtocol()
      scrubbed.method or= method
      scrubbed.name or= @kiteName
      callback scrubbed

  receiveMessageFromAnotherKite:(message, headers, deliveryInfo)->
    messageData = (try message.data.toString()) or message
    scrubbed = JSON.parse messageData
    routingKey = @kiteRoutingKeyPrefix + scrubbed.name
    @handleRequest scrubbed, routingKey, {headers, deliveryInfo}, 'kite-intra'

  writePid:(pidPath)->
    if pidPath
      require("fs").writeFile pidPath,"#{process.pid}",(err)->
        if err?
          console.log "[WARN] Can't write pid to #{pidPath}:  monit can't watch this process. err:#{err.message}"

{'fraction_non_alphanumeric': 0.12214752390872076, 'fraction_numerical': 0.0015781333838336016, 'mean_word_length': 3.158005249343832, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 255, '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': '5625495', 'n_tokens_mistral': 9699, 'n_tokens_neox': 9360, 'n_words': 2357}