-
Notifications
You must be signed in to change notification settings - Fork 39
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
cleanup. Add ConnectionStream object to funnel all messages #48
Changes from 4 commits
86eb77d
b3bb738
e6378e1
0bb1b58
9fd37fb
fd6581c
02a43ca
c00ec4d
ccfc342
67b2375
bda63cc
7adc0d2
0841ff0
f52cb5b
7c37d61
2253d0c
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,132 @@ | ||
local bind = require('utils').bind | ||
local timer = require('timer') | ||
local Emitter = require('core').Emitter | ||
local Object = require('core').Object | ||
local misc = require('../util/misc') | ||
local logging = require('logging') | ||
local loggingUtil = require ('../util/logging') | ||
local table = require('table') | ||
local os = require('os') | ||
local Scheduler = require('../schedule').Scheduler | ||
|
||
local fmt = require('string').format | ||
|
||
function SECONDS(n, jitter) | ||
return misc.calcJitter(n * 1000, jitter * 1000) | ||
end | ||
|
||
-- State | ||
|
||
local State = Emitter:extend() | ||
|
||
function State:initialize(name, messages) | ||
self._name = name | ||
self._messages = messages | ||
self._log = loggingUtil.makeLogger(fmt('State(%s)', self._name)) | ||
end | ||
|
||
function State:getName() | ||
return self._name | ||
end | ||
|
||
-- RegisterCheckState State | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Can you describe what a RegisterCheckState is in a few words here? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Added in 02a43ca. |
||
|
||
local RegisterCheckState = State:extend() | ||
function RegisterCheckState:initialize(messages) | ||
State.initialize(self, 'RegisterCheckState', messages) | ||
self._timeout = SECONDS(20, 10) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Can you pull these values out to the top of the file and describe why they were chosen? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I removed this code, because we don't use the timeout in this context. c00ec4d |
||
self._lastFetchTime = 0 | ||
end | ||
|
||
function RegisterCheckState:_scheduleManifest(client, manifest) | ||
local checks = client:_createChecks(manifest) | ||
self._scheduler = Scheduler:new('scheduler.state', checks, function() | ||
self._scheduler:start() | ||
end) | ||
self._scheduler:on('check', function(check, checkResults) | ||
local client = self._messages:getStream():getClient() | ||
if client then | ||
client.protocol:sendMetrics(check, checkResults) | ||
end | ||
end) | ||
end | ||
|
||
function RegisterCheckState:onHandshake() | ||
|
||
function run() | ||
local client = self._messages:getStream():getClient() | ||
if client then | ||
client.protocol:getManifest(function(err, manifest) | ||
if err then | ||
-- TODO Abort connection? | ||
client:getLog()(logging.ERROR, 'Error while retrieving manifest: ' .. err.message) | ||
else | ||
self:_scheduleManifest(client, manifest) | ||
end | ||
end) | ||
end | ||
end | ||
|
||
-- TODO at some point we want to add logic to update the manifest | ||
if self._lastFetchTime == 0 then | ||
if self._timer then | ||
timer.clearTimer(self._timer) | ||
end | ||
self._log(logging.DEBUG, fmt('registering timeout %d', self._timeout)) | ||
self._timer = process.nextTick(run) | ||
self._lastFetchTime = os.time() | ||
end | ||
end | ||
|
||
-- Connection Messages | ||
|
||
local ConnectionMessages = Emitter:extend() | ||
function ConnectionMessages:initialize(connectionStream) | ||
self._connectionStream = connectionStream | ||
self._states = {} | ||
self:_addState(RegisterCheckState:new(self)) | ||
self:on('handshake_success', bind(ConnectionMessages.onHandshake, self)) | ||
self:on('client_end', bind(ConnectionMessages.onClientEnd, self)) | ||
self:on('message', bind(ConnectionMessages.onMessage, self)) | ||
end | ||
|
||
function ConnectionMessages:_addState(state) | ||
table.insert(self._states, state) | ||
end | ||
|
||
function ConnectionMessages:getStream() | ||
return self._connectionStream | ||
end | ||
|
||
function ConnectionMessages:onClientEnd(client) | ||
client:getLog()(logging.INFO, 'Detected client disconnect') | ||
for i in ipairs(self._states) do | ||
if self._states[i].onClientEnd then | ||
self._states[i]:onClientEnd(client) | ||
end | ||
end | ||
end | ||
|
||
function ConnectionMessages:onHandshake(client) | ||
client:getLog()(logging.DEBUG, '(onHandshake)') | ||
for i in ipairs(self._states) do | ||
if self._states[i].onHandshake then | ||
self._states[i]:onHandshake(client) | ||
end | ||
end | ||
end | ||
|
||
function ConnectionMessages:onMessage(client, msg) | ||
client:getLog()(logging.DEBUG, '(onMessage)') | ||
for i in ipairs(self._states) do | ||
if self._states[i].onMessage then | ||
self._states[i]:onMessage(client, msg) | ||
end | ||
end | ||
client.protocol:processMessage(msg) | ||
end | ||
|
||
local exports = {} | ||
exports.State = State | ||
exports.ConnectionMessages = ConnectionMessages | ||
return exports |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Why not just have it call self.log(...)? the c:getLog()("STUFF") is sort of ugly.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Changed in ccfc342
client:log(PRIORITY, message, ...)
Good call... I think this looks cleaner.