<commit_before>--[[
Copyright 2012 Rackspace

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS-IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
--]]

local Emitter = require('core').Emitter
local math = require('math')
local timer = require('timer')
local fmt = require('string').format

local async = require('async')
local dns = require('dns')

local ConnectionMessages = require('./connection_messages').ConnectionMessages
local UpgradePollEmitter = require('./upgrade').UpgradePollEmitter

local Scheduler = require('../schedule').Scheduler
local AgentClient = require('./client').AgentClient
local logging = require('logging')
local consts = require('../util/constants')
local misc = require('../util/misc')
local vtime = require('virgo-time')
local path = require('path')
local utils = require('utils')
local version = require('../util/version')
local request = require('../protocol/request')

local ConnectionStream = Emitter:extend()
function ConnectionStream:initialize(id, token, guid, options)
  self._id = id
  self._token = token
  self._guid = guid
  self._channel = nil
  self._clients = {}
  self._unauthedClients = {}
  self._delays = {}
  self._activeTimeSyncClient = nil
  self._options = options or {}
  self._scheduler = Scheduler:new()
  self._scheduler:on('check.completed', function(check, checkResult)
    self:_sendMetrics(check, checkResult)
  end)

  local _event_names = {
    'bundle_upgrade.success',
    'binary_upgrade.success',
    'bundle_upgrade.already_downloaded',
    'binary_upgrade.already_downloaded',
    'bundle_upgrade.error',
    'binary_upgrade.error'
  }
  self._messages = ConnectionMessages:new(self)
  misc.propagateEvents(self._messages, self, _event_names)

  self._upgrade = UpgradePollEmitter:new()
   self._upgrade:on('upgrade', utils.bind(ConnectionStream._onUpgrade, self))
end

function ConnectionStream:getUpgrade()
  return self._upgrade
end

function ConnectionStream:setChannel(channel)
  self._channel = channel or consts.DEFAULT_CHANNEL
end

function ConnectionStream:_onUpgrade()
  local client = self:getClient()
  local bundleVersion = version.bundle
  local processVersion = version.process
  local uri_path, options

  if not client then
    return
  end

  options = {
    method = 'GET',
    host = client._host,
    port = client._port,
    tls = client._tls_options
  }

  self._channel = 'master'
  uri_path = fmt('/upgrades/%s/VERSION', self._channel)
  options = misc.merge({ path = uri_path, }, options)
  request.makeRequest(options, function(err, result, version)
    if err then
      client:log(logging.ERROR, 'Error on upgrade: ' .. tostring(err))
      return
    end
    version = misc.trim(version)
    client:log(logging.DEBUG, fmt('(upgrade) -> Current Version: %s', bundleVersion))
    client:log(logging.DEBUG, fmt('(upgrade) -> Upstream Version: %s', version))
      self._messages:getUpgrade(version, client)
    if misc.compareVersions(version, bundleVersion) > 0 then
      client:log(logging.INFO, fmt('(upgrade) -> found: %s', version))
      self._messages:getUpgrade(version, client)
    end
  end)
end
<commit_msg>remove my debug getUpgrade<commit_after>--[[
Copyright 2012 Rackspace

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS-IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
--]]

local Emitter = require('core').Emitter
local math = require('math')
local timer = require('timer')
local fmt = require('string').format

local async = require('async')
local dns = require('dns')

local ConnectionMessages = require('./connection_messages').ConnectionMessages
local UpgradePollEmitter = require('./upgrade').UpgradePollEmitter

local Scheduler = require('../schedule').Scheduler
local AgentClient = require('./client').AgentClient
local logging = require('logging')
local consts = require('../util/constants')
local misc = require('../util/misc')
local vtime = require('virgo-time')
local path = require('path')
local utils = require('utils')
local version = require('../util/version')
local request = require('../protocol/request')

local ConnectionStream = Emitter:extend()
function ConnectionStream:initialize(id, token, guid, options)
  self._id = id
  self._token = token
  self._guid = guid
  self._channel = nil
  self._clients = {}
  self._unauthedClients = {}
  self._delays = {}
  self._activeTimeSyncClient = nil
  self._options = options or {}
  self._scheduler = Scheduler:new()
  self._scheduler:on('check.completed', function(check, checkResult)
    self:_sendMetrics(check, checkResult)
  end)

  local _event_names = {
    'bundle_upgrade.success',
    'binary_upgrade.success',
    'bundle_upgrade.already_downloaded',
    'binary_upgrade.already_downloaded',
    'bundle_upgrade.error',
    'binary_upgrade.error'
  }
  self._messages = ConnectionMessages:new(self)
  misc.propagateEvents(self._messages, self, _event_names)

  self._upgrade = UpgradePollEmitter:new()
   self._upgrade:on('upgrade', utils.bind(ConnectionStream._onUpgrade, self))
end

function ConnectionStream:getUpgrade()
  return self._upgrade
end

function ConnectionStream:setChannel(channel)
  self._channel = channel or consts.DEFAULT_CHANNEL
end

function ConnectionStream:_onUpgrade()
  local client = self:getClient()
  local bundleVersion = version.bundle
  local processVersion = version.process
  local uri_path, options

  if not client then
    return
  end

  options = {
    method = 'GET',
    host = client._host,
    port = client._port,
    tls = client._tls_options
  }

  self._channel = 'master'
  uri_path = fmt('/upgrades/%s/VERSION', self._channel)
  options = misc.merge({ path = uri_path, }, options)
  request.makeRequest(options, function(err, result, version)
    if err then
      client:log(logging.ERROR, 'Error on upgrade: ' .. tostring(err))
      return
    end
    version = misc.trim(version)
    client:log(logging.DEBUG, fmt('(upgrade) -> Current Version: %s', bundleVersion))
    client:log(logging.DEBUG, fmt('(upgrade) -> Upstream Version: %s', version))
    if misc.compareVersions(version, bundleVersion) > 0 then
      client:log(logging.INFO, fmt('(upgrade) -> found: %s', version))
      self._messages:getUpgrade(version, client)
    end
  end)
end

{'fraction_non_alphanumeric': 0.11743822777619994, 'fraction_numerical': 0.0025560920193126954, 'mean_word_length': 4.532600157109191, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4436963', 'n_tokens_mistral': 2076, 'n_tokens_neox': 2003, 'n_words': 565}