<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 this, false, {args:args, from:from}
      return

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

    if @api.propertyIsEnumerable(cmd) and 'function' is typeof @api[cmd]
      fn = @api[cmd]
      fn.call @kite, {args:args, from:from}, (err, result)=>
        @sendCommand(from, "reply-#{cmdId}", [err, result])
      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 callback
      @waitingReplies[@callbackId] = callback

    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 callback
      @waitingReplies[@callbackId] = callback
    @mergenData.publish channel, JSON.stringify(data)

  connectToMessageBus: (callback)=>

    # 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 = 
      pid: process.pid
      hostname: os.hostname()
      serviceGenericName: @serviceGenericName
      serviceUniqueName: @serviceUniqueName
      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)

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

    for channel in channelNames
      @mergen.subscribe channel

    @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)->
      channelNames = _.without(subscribeToChannelNames, channel)

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

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


module.exports = class Kite extends EventEmitter
  {parse, stringify} = JSON
  Object.defineProperty Object.prototype, 'throwError', value:(it)-> throw it<commit_msg>old habits<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 this, false, {args:args, from:from}
      return

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

    if @api.propertyIsEnumerable(cmd) and 'function' is typeof @api[cmd]
      fn = @api[cmd]
      fn.call @kite, {args:args, from:from}, (err, result)=>
        @sendCommand(from, "reply-#{cmdId}", [err, result])
      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 callback
      @waitingReplies[@callbackId] = callback

    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 callback
      @waitingReplies[@callbackId] = callback
    @mergenData.publish channel, JSON.stringify(data)

  connectToMessageBus: (callback)=>

    # 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 = 
      pid: process.pid
      hostname: os.hostname()
      serviceGenericName: @serviceGenericName
      serviceUniqueName: @serviceUniqueName
      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)

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

    for channel in channelNames
      @mergen.subscribe channel

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

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

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

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


module.exports = class Kite extends EventEmitter
  {parse, stringify} = JSON
  Object.defineProperty Object.prototype, 'throwError', value:(it)-> throw it
{'fraction_non_alphanumeric': 0.13036402569593147, 'fraction_numerical': 0.002569593147751606, 'mean_word_length': 3.0401384083044984, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 75, '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': '5104126', 'n_tokens_mistral': 3599, 'n_tokens_neox': 3520, 'n_words': 877}