|
#--
|
|
# CyborgHood, a distributed system management software.
|
|
# Copyright (c) 2009-2010 Marc Dequènes (Duck) <Duck@DuckCorp.org>
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
#++
|
|
|
|
require 'yaml'
|
|
require 'cyborghood/cyborg/session'
|
|
|
|
|
|
module CyborgHood
|
|
|
|
class Protocol
|
|
VERSION = "0.1"
|
|
|
|
def initialize(conversation)
|
|
@conversation = conversation
|
|
end
|
|
|
|
def receive_announce_helo(parameters)
|
|
unless parameters[:bot_name] =~ Conversation::BOT_ID_PATTERN
|
|
return send_error_protocol "bad bot name"
|
|
end
|
|
unless parameters[:protocol_version] == VERSION
|
|
@conversation.set_fatal
|
|
return send_error_protocol "protocol version does not match"
|
|
end
|
|
@conversation.set_peer_info(parameters[:bot_name], parameters[:capabilities])
|
|
send_reply_ok
|
|
end
|
|
|
|
def send_error_protocol(msg = nil)
|
|
@conversation.set_reply("ERROR PROTO", { :error => msg || "" })
|
|
end
|
|
|
|
def send_reply_ack
|
|
@conversation.set_reply("REPLY ACK")
|
|
end
|
|
end
|
|
|
|
class Conversation < EventMachine::Protocols::LineAndTextProtocol
|
|
private_class_method :new
|
|
|
|
# don't rely on EventMachine's default, it may change one day
|
|
MaxLineLength = 16*1024
|
|
|
|
EOD = "\033[0J"
|
|
BOT_ID_PATTERN = "[a-zA-Z0-9]+"
|
|
ACTION_PATTERN = "^(#{BOT_ID_PATTERN})(\d{4})-(\d{4})([+]?) ([a-zA-Z0-9 ]+)$"
|
|
MAXIMUM_ERROR_COUNT = 3
|
|
MAXIMUM_LINES = 1024
|
|
|
|
attr_reader :interface
|
|
|
|
def initialize(interface)
|
|
@interface = interface
|
|
|
|
super
|
|
|
|
@config = Config.instance
|
|
@split_data_mode = false
|
|
@split_data_action = nil
|
|
@split_data = []
|
|
# one session for each conversation thread
|
|
@sessions = {}
|
|
@error_count = 0
|
|
@fatal_error = false
|
|
@actions_wip = {}
|
|
@peer_id = nil
|
|
@peer_capabilities = []
|
|
|
|
@protocol = Protocol.new(this)
|
|
end
|
|
|
|
def send_line(msg)
|
|
send_data "#{msg}\n"
|
|
logger.debug "Sent data [#{identifier}]: #{msg}"
|
|
end
|
|
|
|
def post_init
|
|
logger.debug "New conversation with #{identifier}"
|
|
end
|
|
|
|
def receive_line(data)
|
|
return if data.empty?
|
|
|
|
@reply_action = "Internal error"
|
|
@reply_parameters = {}
|
|
|
|
if data == EOD
|
|
logger.debug "Received EOD [#{identifier}]"
|
|
exit_split_mode
|
|
else
|
|
if @split_data_mode
|
|
logger.debug "Received data (split mode) [#{identifier}]: #{data}"
|
|
|
|
if @split_data.size > MAXIMUM_LINES
|
|
reply_fatal_error "overflow"
|
|
else
|
|
@split_data << data
|
|
end
|
|
else
|
|
logger.debug "Received data [#{identifier}]: #{data}"
|
|
|
|
if data =~ Regexp.new(ACTION_PATTERN)
|
|
@thread_id = $1
|
|
@action_id = $2
|
|
flags = $3 || ""
|
|
action = $4
|
|
|
|
if @actions_wip[@thread_id].nil? or not @actions_wip[@thread_id].include?(@action_id)
|
|
@sessions[@thread_id] ||= Session.new
|
|
if flags.index '+'
|
|
enter_split_mode(action)
|
|
else
|
|
receive_action(action)
|
|
end
|
|
else
|
|
reply_fatal_error "wip action id reused"
|
|
end
|
|
else
|
|
logger.error "Error [#{identifier}]: syntax error"
|
|
@error_count += 1
|
|
|
|
if @error_count >= MAXIMUM_ERROR_COUNT
|
|
reply_fatal_error "too much errors, terminating"
|
|
else
|
|
reply_syntax_error "bad action"
|
|
end
|
|
end
|
|
end
|
|
end
|
|
|
|
send_reply
|
|
close_connection_after_writing if @fatal_error
|
|
end
|
|
|
|
def receive_action(action, data = nil)
|
|
logger.debug "Executing action '#{action}' [#{identifier}]"
|
|
|
|
unless data.nil?
|
|
parameters = YAML.parse(data)
|
|
reply_syntax_error "bad parameters"
|
|
return
|
|
end
|
|
|
|
# use a Protocol object and convert back result to the conversation level protocol
|
|
# we need to use EM.defer or EM.spawn+Deferrable, preferably after parsing, to avoid spawning threads for nothing
|
|
#send_line @interface.call(@session, action, data)
|
|
end
|
|
|
|
def receive_error(msg)
|
|
logger.error "Error [#{identifier}]: #{msg}"
|
|
end
|
|
|
|
def unbind
|
|
logger.debug "Conversation finished with #{identifier}"
|
|
@sessions.each_values {|s| s.clear }
|
|
end
|
|
|
|
def set_peer_info(id, capabilities)
|
|
@peer_id = id
|
|
@peer_capabilities = capabilities || []
|
|
end
|
|
|
|
def set_reply(action, parameters = nil)
|
|
@reply_action = action
|
|
@reply_parameters = parameters
|
|
end
|
|
|
|
def set_fatal
|
|
@fatal_error = true
|
|
end
|
|
|
|
def current_session
|
|
@sessions[@thread_id]
|
|
end
|
|
|
|
protected
|
|
|
|
def reply_syntax_error(msg = nil)
|
|
msg = "syntax error" + (msg ? ": " + msg : "")
|
|
@protocol.send_error_protocol(msg)
|
|
end
|
|
|
|
def reply_fatal_error(msg = nil)
|
|
msg = "fatal error" + (msg ? ": " + msg : "")
|
|
@protocol.send_error_protocol(msg)
|
|
set_fatal
|
|
end
|
|
|
|
def send_reply
|
|
send_line "#{@config.bot_id}-#{@thread_id}-#{@action_id}" + (@reply_parameters.nil? : "" : "+") + " #{@reply_action}"
|
|
unless @reply_parameters.nil?
|
|
@reply_parameters.to_yaml.split.each {|l| send_line l }
|
|
send_line EOD
|
|
end
|
|
end
|
|
|
|
def enter_split_mode(action)
|
|
if @split_data_mode
|
|
logger.error "Error [#{identifier}]: already in split mode"
|
|
send_line "551 protocol error"
|
|
@split_data_mode = false
|
|
@split_data_action = nil
|
|
else
|
|
logger.debug "Entered split mode for action '#{action}' [#{identifier}]"
|
|
@split_data_mode = true
|
|
@split_data_action = action
|
|
end
|
|
@split_data = []
|
|
end
|
|
|
|
def exit_split_mode
|
|
if @split_data_mode
|
|
logger.debug "Quit split mode for action '#{@split_data_action}' [#{identifier}]"
|
|
receive_action(@split_data_action, @split_data.join("\n"))
|
|
else
|
|
logger.error "Error [#{identifier}]: not in split mode"
|
|
send_line "551 protocol error"
|
|
end
|
|
@split_data_mode = false
|
|
@split_data_action = nil
|
|
@split_data = []
|
|
end
|
|
end
|
|
|
|
class ConversationUNIXSocket < Conversation
|
|
public_class_method :new
|
|
|
|
def identifier
|
|
"unix_socket/#{@signature}"
|
|
end
|
|
end
|
|
end
|