<commit_before>{EventEmitter} = require 'microemitter'

module.exports = class Broker extends EventEmitter

  [NOTREADY, READY, CLOSED] = [0,1,3]

  Channel = require './channel'

  createId = require 'hat'

  constructor:(ws, options)->
    @sockURL = ws
    {@autoReconnect, @authChannelName, @overlapDuration} = options
    @overlapDuration ?= 3000
    @authChannelName ?= 'auth'
    @readyState = NOTREADY
    @channels = {}
    @namespacedEvents = {}
    @subscriptions = []
    @listOfSubscriptions = {}
    @initBackoff options  if @autoReconnect
    @connect()

  bound: require 'koding-bound'

  onopen:->
    if @readyState is CLOSED
      @clearBackoffTimeout()
      # replay any existing subscriptions after reconnecting.
      oldSubscriptions = @subscriptions.slice()
      @listOfSubscriptions.length = 0
      @subscriptions.length = 0
      @subscribe args...  for args in (oldSubscriptions or [])

    @readyState = READY

    @emit 'ready'
    @emit 'connected'

  onclose:->
    @readyState = CLOSED
    @emit "disconnected", Object.keys @channels
    @channels = {}
    if @autoReconnect
      process.nextTick => @setBackoffTimeout @connect.bind this

  connect:->
    @ws = new SockJS @sockURL
    @ws.addEventListener 'open', @bound 'onopen'
    @ws.addEventListener 'close', @bound 'onclose'
    @ws.addEventListener 'message', @bound 'handleMessageEvent'

  disconnect:(reconnect)->
    @autoReconnect = !!reconnect  if reconnect?
    @ws.close()

  initBackoff:(options)->
    backoff = options.backoff ? {}
    totalReconnectAttempts = 0
    initalDelayMs = backoff.initialDelayMs ? 700
    multiplyFactor = backoff.multiplyFactor ? 1.4
    maxDelayMs = backoff.maxDelayMs ? 1000 * 15 # 15 seconds
    maxReconnectAttempts = backoff.maxReconnectAttempts ? 10

    @clearBackoffTimeout =->
      totalReconnectAttempts = 0

    @setBackoffTimeout = (fn)=>
      if totalReconnectAttempts < maxReconnectAttempts
        timeout = Math.min initalDelayMs * Math.pow(
          multiplyFactor, totalReconnectAttempts
        ), maxDelayMs
        setTimeout fn, timeout
        totalReconnectAttempts++
      else
        @emit "connectionFailed"

  createRoutingKeyPrefix:(name, options = {})->
    {isReadOnly, suffix} = options
    name += suffix or ''
    if isReadOnly then name
    else "client.#{name}"

  wrapPrivateChannel:(channel)->
    channel.on 'cycle', => @authenticate channel
    channel.on 'setSecretName', (secretName)=>

      {isReadOnly} = channel
      
      channel.setSecretName secretName
      channel.close() # close the channel—no more messages from upstream
      
      privateChannel = @subscribe secretName, {isReadOnly, isSecret: yes}
      privateChannel.setAuthenticationInfo {serviceType: 'secret'}
      
      channel.privateChannel = privateChannel
      
      privateChannel.on 'cycleChannel', ->
        channel.oldPrivateChannel = channel.privateChannel
        channel.cycle()
      
      @swapPrivateSourceChannel channel
      
      channel.emit 'ready'

  swapPrivateSourceChannel:(channel)->
    {privateChannel, oldPrivateChannel} = channel
    if oldPrivateChannel?
      setTimeout =>

        oldPrivateChannel.close().off()
        delete channel.oldPrivateChannel

        privateChannel.pipe channel

      , @overlapDuration
    else
      privateChannel.pipe channel

  registerNamespacedEvent:(name)->
    register = @namespacedEvents
    register[name] ?= 0
    register[name] += 1
    return register[name] is 1

  createChannel:(name, {isReadOnly, isSecret, isExclusive, suffix})->
    return @channels[name]  if @channels[name]?

    suffix ?= if isExclusive then ".#{createId 32}" else ''

    routingKeyPrefix = @createRoutingKeyPrefix name, {suffix, isReadOnly}
    channel = new Channel name, routingKeyPrefix, {
      isReadOnly, isSecret, isExclusive
    }

    # handle authentication once the broker is subscribed.
    @on 'broker.subscribed', handler = (routingKeyPrefix1)=>
      if routingKeyPrefix is routingKeyPrefix1
        @authenticate channel
        @off 'broker.subscribed', handler
        channel.emit 'broker.subscribed', channel.routingKeyPrefix
    # messages that are routed to the bare routingKey will be interpretted
    # as "message" events
    @on routingKeyPrefix, (rest...)-> channel.emit 'message', rest...

    channel.on 'newListener', (event, listener)=>
      if channel.privateChannel?
        channel.trackListener event, listener
      unless event is 'broker.subscribed'
        namespacedEvent = "#{routingKeyPrefix}.#{event}"
        needsToBeRegistered = @registerNamespacedEvent namespacedEvent
        if needsToBeRegistered
          # add the namespaced listener, but forward the event only
          @on namespacedEvent, (rest...)-> channel.emit event, rest...

    unless isSecret
      channel.on 'authOk', -> channel.isAuthenticated = yes

    channel.on 'error', channel.bound 'close'
    channel.on 'close', => @unsubscribe channel.name
    @wrapPrivateChannel channel  if isExclusive

    # when the channel emits a publish event, we want to actually publish the
    # message to the broker.
    unless isReadOnly
      channel.on 'publish', (payload)=>
        exchange = channel.exchange ? channel.name
        @publish {exchange, routingKey: channel.name}, payload

    @channels[name] = channel
    return channel

  authenticate:(channel)->
    authInfo = {}
    authInfo[key] = val  for own key, val of channel.getAuthenticationInfo()
    authInfo.routingKey = channel.routingKeyPrefix
    @publish @authChannelName, authInfo

  handleMessageEvent:(event)->
    message = event.data
    @emit 'rawMessage', message
    @emit message.routingKey, message.payload  if message.routingKey
    return

  ready:(listener)->
    if @readyState is READY then listener()
    else @on 'ready', listener

  send:(data)->
    @ready => @ws._transport.doSend JSON.stringify data
    return this

  publish:(options, payload)->
    if 'string' is typeof options
      routingKey = exchange = options
    else
      {routingKey, exchange} = options
    routingKey = @createRoutingKeyPrefix routingKey
    payload = JSON.stringify payload  unless 'string' is typeof payload
    @send {
      action: 'publish'
      exchange
      routingKey
      payload
    }
    return this

  subscribe:(name, options={}, callback)->
    return @channels[name]  if @channels[name]
    @subscriptions.push arguments
    isReadOnly  = !!options.isReadOnly
    isSecret    = !!options.isSecret
    isExclusive = !!options.isExclusive
    {suffix}    = options
    routingKeyPrefix = @createRoutingKeyPrefix name, {isReadOnly}
    channel = @channels[name] = @createChannel name, {
      isReadOnly, isSecret, isExclusive, suffix
    }
    @send {
      action: 'subscribe'
      routingKeyPrefix: channel.routingKeyPrefix
    }
    if callback?
      @on 'broker.subscribed', handler = (prefix)=>
        if prefix is routingKeyPrefix
          @off 'broker.subscribed', handler
          callback prefix
    @listOfSubscriptions[name] = name
    return channel

  unsubscribe:(name)->
    @send {
      action: 'unsubscribe'
      routingKeyPrefix: @createRoutingKeyPrefix name
    }
    return this

  ping:(callback)->
    @send { action: "ping" }
    @once "broker.pong", callback  if callback?
<commit_msg>broker-client: allow delegate channel constructors.<commit_after>{EventEmitter} = require 'microemitter'

module.exports = class Broker extends EventEmitter

  [NOTREADY, READY, CLOSED] = [0,1,3]

  Channel = require './channel'

  createId = require 'hat'

  constructor:(ws, options)->
    @sockURL = ws
    {@autoReconnect, @authChannelName, @overlapDuration} = options
    @overlapDuration ?= 3000
    @authChannelName ?= 'auth'
    @readyState = NOTREADY
    @channels = {}
    @namespacedEvents = {}
    @subscriptions = []
    @listOfSubscriptions = {}
    @initBackoff options  if @autoReconnect
    @connect()

  bound: require 'koding-bound'

  onopen:->
    if @readyState is CLOSED
      @clearBackoffTimeout()
      # replay any existing subscriptions after reconnecting.
      oldSubscriptions = @subscriptions.slice()
      @listOfSubscriptions.length = 0
      @subscriptions.length = 0
      @subscribe args...  for args in (oldSubscriptions or [])

    @readyState = READY

    @emit 'ready'
    @emit 'connected'

  onclose:->
    @readyState = CLOSED
    @emit "disconnected", Object.keys @channels
    @channels = {}
    if @autoReconnect
      process.nextTick => @setBackoffTimeout @connect.bind this

  connect:->
    @ws = new SockJS @sockURL
    @ws.addEventListener 'open', @bound 'onopen'
    @ws.addEventListener 'close', @bound 'onclose'
    @ws.addEventListener 'message', @bound 'handleMessageEvent'

  disconnect:(reconnect)->
    @autoReconnect = !!reconnect  if reconnect?
    @ws.close()

  initBackoff:(options)->
    backoff = options.backoff ? {}
    totalReconnectAttempts = 0
    initalDelayMs = backoff.initialDelayMs ? 700
    multiplyFactor = backoff.multiplyFactor ? 1.4
    maxDelayMs = backoff.maxDelayMs ? 1000 * 15 # 15 seconds
    maxReconnectAttempts = backoff.maxReconnectAttempts ? 10

    @clearBackoffTimeout =->
      totalReconnectAttempts = 0

    @setBackoffTimeout = (fn)=>
      if totalReconnectAttempts < maxReconnectAttempts
        timeout = Math.min initalDelayMs * Math.pow(
          multiplyFactor, totalReconnectAttempts
        ), maxDelayMs
        setTimeout fn, timeout
        totalReconnectAttempts++
      else
        @emit "connectionFailed"

  createRoutingKeyPrefix:(name, options = {})->
    {isReadOnly, suffix} = options
    name += suffix or ''
    if isReadOnly then name
    else "client.#{name}"

  wrapPrivateChannel:(channel)->
    channel.on 'cycle', => @authenticate channel
    channel.on 'setSecretName', (secretName)=>

      {isReadOnly} = channel
      
      channel.setSecretName secretName
      channel.close() # close the channel—no more messages from upstream
      
      privateChannel = @subscribe secretName, {isReadOnly, isSecret: yes}
      privateChannel.setAuthenticationInfo {serviceType: 'secret'}
      
      channel.privateChannel = privateChannel
      
      privateChannel.on 'cycleChannel', ->
        channel.oldPrivateChannel = channel.privateChannel
        channel.cycle()
      
      @swapPrivateSourceChannel channel
      
      channel.emit 'ready'

  swapPrivateSourceChannel:(channel)->
    {privateChannel, oldPrivateChannel} = channel
    if oldPrivateChannel?
      setTimeout =>

        oldPrivateChannel.close().off()
        delete channel.oldPrivateChannel

        privateChannel.pipe channel

      , @overlapDuration
    else
      privateChannel.pipe channel

  registerNamespacedEvent:(name)->
    register = @namespacedEvents
    register[name] ?= 0
    register[name] += 1
    return register[name] is 1

  createChannel:(name, options)->
    return @channels[name]  if @channels[name]?

    { isReadOnly, isSecret, isExclusive, suffix, konstructor } = options

    suffix ?= if isExclusive then ".#{createId 32}" else ''

    routingKeyPrefix = @createRoutingKeyPrefix name, { suffix, isReadOnly }
    channel = new (konstructor ? Channel) name, routingKeyPrefix, {
      isReadOnly, isSecret, isExclusive
    }

    # handle authentication once the broker is subscribed.
    @on 'broker.subscribed', handler = (routingKeyPrefix1)=>
      if routingKeyPrefix is routingKeyPrefix1
        @authenticate channel
        @off 'broker.subscribed', handler
        channel.emit 'broker.subscribed', channel.routingKeyPrefix
    # messages that are routed to the bare routingKey will be interpretted
    # as "message" events
    @on routingKeyPrefix, (rest...)-> channel.emit 'message', rest...

    channel.on 'newListener', (event, listener)=>
      if channel.privateChannel?
        channel.trackListener event, listener
      unless event is 'broker.subscribed'
        namespacedEvent = "#{routingKeyPrefix}.#{event}"
        needsToBeRegistered = @registerNamespacedEvent namespacedEvent
        if needsToBeRegistered
          # add the namespaced listener, but forward the event only
          @on namespacedEvent, (rest...)-> channel.emit event, rest...

    unless isSecret
      channel.on 'authOk', -> channel.isAuthenticated = yes

    channel.on 'error', channel.bound 'close'
    channel.on 'close', => @unsubscribe channel.name
    @wrapPrivateChannel channel  if isExclusive

    # when the channel emits a publish event, we want to actually publish the
    # message to the broker.
    unless isReadOnly
      channel.on 'publish', (payload)=>
        exchange = channel.exchange ? channel.name
        @publish {exchange, routingKey: channel.name}, payload

    @channels[name] = channel
    return channel

  authenticate:(channel)->
    authInfo = {}
    authInfo[key] = val  for own key, val of channel.getAuthenticationInfo()
    authInfo.routingKey = channel.routingKeyPrefix
    @publish @authChannelName, authInfo

  handleMessageEvent:(event)->
    message = event.data
    @emit 'rawMessage', message
    @emit message.routingKey, message.payload  if message.routingKey
    return

  ready:(listener)->
    if @readyState is READY then listener()
    else @on 'ready', listener

  send:(data)->
    @ready => @ws._transport.doSend JSON.stringify data
    return this

  publish:(options, payload)->
    if 'string' is typeof options
      routingKey = exchange = options
    else
      {routingKey, exchange} = options
    routingKey = @createRoutingKeyPrefix routingKey
    payload = JSON.stringify payload  unless 'string' is typeof payload
    @send {
      action: 'publish'
      exchange
      routingKey
      payload
    }
    return this

  subscribe:(name, options={}, callback)->
    return @channels[name]  if @channels[name]
    @subscriptions.push arguments
    isReadOnly  = !!options.isReadOnly
    isSecret    = !!options.isSecret
    isExclusive = !!options.isExclusive
    {suffix}    = options
    routingKeyPrefix = @createRoutingKeyPrefix name, {isReadOnly}
    channel = @channels[name] = @createChannel name, {
      isReadOnly, isSecret, isExclusive, suffix
    }
    @send {
      action: 'subscribe'
      routingKeyPrefix: channel.routingKeyPrefix
    }
    if callback?
      @on 'broker.subscribed', handler = (prefix)=>
        if prefix is routingKeyPrefix
          @off 'broker.subscribed', handler
          callback prefix
    @listOfSubscriptions[name] = name
    return channel

  unsubscribe:(name)->
    @send {
      action: 'unsubscribe'
      routingKeyPrefix: @createRoutingKeyPrefix name
    }
    return this

  ping:(callback)->
    @send { action: "ping" }
    @once "broker.pong", callback  if callback?

{'fraction_non_alphanumeric': 0.09948495527243155, 'fraction_numerical': 0.0044727568446733534, 'mean_word_length': 3.081028761061947, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 73, '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': '1920227', 'n_tokens_mistral': 4287, 'n_tokens_neox': 4110, 'n_words': 1094}