Revision e26c015f
Added by Marc Dequènes about 14 years ago
- ID e26c015fb860b80b174f901b037240d5f1c5826c
| lib/cyborghood/cyborg/conversation.rb | ||
|---|---|---|
|
require 'yaml'
|
||
|
require 'cyborghood/cyborg/session'
|
||
|
|
||
|
# Note: Event machine core is mono-thread, so this code is not threadsafe.
|
||
|
# Only long tasks will be run in threads, which do not care about all this.
|
||
|
|
||
|
module CyborgHood
|
||
|
|
||
|
class Protocol
|
||
|
module BotProtocol
|
||
|
VERSION = "0.1"
|
||
|
|
||
|
def initialize(conversation)
|
||
|
@conversation = conversation
|
||
|
module_function
|
||
|
|
||
|
def process_received_message(message)
|
||
|
end
|
||
|
|
||
|
def receive(action, parameters)
|
||
|
method = "receive_" + action.downcase.tr(" ", "_")
|
||
|
if respond_to? method
|
||
|
send(method, parameters)
|
||
|
else
|
||
|
send_error_protocol("unknown action")
|
||
|
end
|
||
|
end
|
||
|
|
||
|
def receive_announce_helo(parameters)
|
||
| ... | ... | |
|
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"
|
||
|
# TODO: send QUIT DECLINE "protocol version does not match"
|
||
|
return
|
||
|
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 || "" })
|
||
|
@conversation.set_error_status
|
||
|
@conversation.send_peer("ERROR PROTO", { :error => msg || "" })
|
||
|
end
|
||
|
|
||
|
def send_reply_ack
|
||
|
@conversation.set_reply("REPLY ACK")
|
||
|
@conversation.send_peer("REPLY ACK")
|
||
|
end
|
||
|
|
||
|
def send_quit(reason = nil)
|
||
|
action = "QUIT " + (reason.nil? ? "LEAVING" : "DECLINE")
|
||
|
parameters = (reason.nil? ? nil : { :reason => reason })
|
||
|
@conversation.send_peer(action, parameters)
|
||
|
end
|
||
|
end
|
||
|
|
||
|
class ConversationThread
|
||
|
attr_reader :conversation, :name, :id, :session
|
||
|
|
||
|
@@next_id = 0
|
||
|
|
||
|
def initialized(conversation, name)
|
||
|
@conversation = conversation
|
||
|
@name = name
|
||
|
|
||
|
@id = @@next_id
|
||
|
@@next_id += 1
|
||
|
@session = Session.new
|
||
|
@next_action_id = 0
|
||
|
end
|
||
|
|
||
|
def new_message(action_code, parameters = nil, action_id = nil)
|
||
|
Message.new(this, action_code, parameters)
|
||
|
end
|
||
|
|
||
|
def next_action_id
|
||
|
id = @next_action_id
|
||
|
@next_action_id +=1
|
||
|
id
|
||
|
end
|
||
|
|
||
|
def close
|
||
|
@session.clear
|
||
|
# TODO: enforce destruction ???
|
||
|
end
|
||
|
end
|
||
|
|
||
|
class Message
|
||
|
private_class_method :new
|
||
|
|
||
|
attr_reader :conv_thread, :action_code, :action_parameters, :action_id
|
||
|
|
||
|
def initialized(conv_thread, action_code, action_parameters = nil, action_id = nil)
|
||
|
@conv_thread = conv_thread
|
||
|
@action_code = action_code
|
||
|
@action_parameters = action_parameters
|
||
|
@action_id = action_id
|
||
|
end
|
||
|
|
||
|
def new?
|
||
|
@action_id.nil?
|
||
|
end
|
||
|
|
||
|
def send
|
||
|
raise CyberError.new(:unrecoverable, "bot/conversation", "Not sending twice the same message") unless self.new?
|
||
|
@action_id = @conv_thread.next_action_id
|
||
|
@conv_thread.conversation.send_message(this)
|
||
|
end
|
||
|
|
||
|
def create_reply(action_code, parameters)
|
||
|
raise CyberError.new(:unrecoverable, "bot/conversation", "Cannot reply to a newly created message") if self.new?
|
||
|
new(this, action_code, parameters)
|
||
|
end
|
||
|
end
|
||
|
|
||
| ... | ... | |
|
|
||
|
EOD = "\033[0J"
|
||
|
BOT_ID_PATTERN = "[a-zA-Z0-9]+"
|
||
|
ACTION_PATTERN = "^(#{BOT_ID_PATTERN})(\d{4})-(\d{4})([+]?) ([a-zA-Z0-9 ]+)$"
|
||
|
ACTION_WORD_PATTERN = "[a-zA-Z0-9]+"
|
||
|
ACTION_PATTERN = "^(#{BOT_ID_PATTERN})(\d{4})-(\d{4})([+]?) (#{ACTION_WORD_PATTERN}( #{ACTION_WORD_PATTERN})*)$"
|
||
|
MAXIMUM_ERROR_COUNT = 3
|
||
|
MAXIMUM_LINES = 1024
|
||
|
|
||
| ... | ... | |
|
super
|
||
|
|
||
|
@config = Config.instance
|
||
|
|
||
|
@receive_error_count = 0
|
||
|
@split_data_mode = false
|
||
|
@split_data_action = nil
|
||
|
@split_data_message = nil
|
||
|
@split_data = []
|
||
|
# one session for each conversation thread
|
||
|
@sessions = {}
|
||
|
@error_count = 0
|
||
|
@fatal_error = false
|
||
|
@actions_wip = {}
|
||
|
|
||
|
# associated conversation threads
|
||
|
@conv_threads = {}
|
||
|
# thread 0 is reserved
|
||
|
@system_thread = self.thread('system')
|
||
|
|
||
|
# post-negociation peer info
|
||
|
@peer_id = nil
|
||
|
@peer_capabilities = []
|
||
|
end
|
||
|
|
||
|
@protocol = Protocol.new(this)
|
||
|
def clear_receive_info
|
||
|
@receive_error = false
|
||
|
@receive_fatal_error = false
|
||
|
end
|
||
|
|
||
|
def send_line(msg)
|
||
| ... | ... | |
|
def receive_line(data)
|
||
|
return if data.empty?
|
||
|
|
||
|
@reply_action = "Internal error"
|
||
|
@reply_parameters = {}
|
||
|
clear_receive_info
|
||
|
|
||
|
if data == EOD
|
||
|
logger.debug "Received EOD [#{identifier}]"
|
||
| ... | ... | |
|
logger.debug "Received data [#{identifier}]: #{data}"
|
||
|
|
||
|
if data =~ Regexp.new(ACTION_PATTERN)
|
||
|
@thread_id = $1
|
||
|
@action_id = $2
|
||
|
conv_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
|
||
|
action_code = $4
|
||
|
|
||
|
if @error_count >= MAXIMUM_ERROR_COUNT
|
||
|
reply_fatal_error "too much errors, terminating"
|
||
|
conv_thread = self.thread_by_id(conv_thread_id)
|
||
|
message = conv_thread.new_message(action_code, nil, action_id)
|
||
|
|
||
|
if flags.index '+'
|
||
|
enter_split_mode(message)
|
||
|
else
|
||
|
reply_syntax_error "bad action"
|
||
|
receive_message(message)
|
||
|
end
|
||
|
else
|
||
|
reply_syntax_error "bad action format"
|
||
|
end
|
||
|
end
|
||
|
end
|
||
|
|
||
|
send_reply
|
||
|
|
||
|
# TODO: properly QUIT
|
||
|
@error_count += 1 if @receive_error
|
||
|
if @error_count >= MAXIMUM_ERROR_COUNT
|
||
|
reply_fatal_error "too much errors, terminating"
|
||
|
end
|
||
|
close_connection_after_writing if @fatal_error
|
||
|
end
|
||
|
|
||
|
def receive_action(action, data = nil)
|
||
|
logger.debug "Executing action '#{action}' [#{identifier}]"
|
||
|
def receive_message(message)
|
||
|
logger.debug "Received message '#{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)
|
||
|
BotProtocol.process_received_message(message)
|
||
|
end
|
||
|
|
||
|
def receive_error(msg)
|
||
| ... | ... | |
|
|
||
|
def unbind
|
||
|
logger.debug "Conversation finished with #{identifier}"
|
||
|
@sessions.each_values {|s| s.clear }
|
||
|
@conv_threads.each_values {|s| s.close }
|
||
|
end
|
||
|
|
||
|
def set_peer_info(id, capabilities)
|
||
| ... | ... | |
|
@peer_capabilities = capabilities || []
|
||
|
end
|
||
|
|
||
|
def set_reply(action, parameters = nil)
|
||
|
@reply_action = action
|
||
|
@reply_parameters = parameters
|
||
|
def set_error_status(fatal = false)
|
||
|
# fatal status is conservative, it cannot be canceled
|
||
|
@reveive_fatal_error = @reveive_fatal_error || fatal
|
||
|
@receive_error = true
|
||
|
end
|
||
|
|
||
|
def thread(name = 'default')
|
||
|
th = @conv_threads[name] || ConversationThread.new(this, name)
|
||
|
# allow searching by id too
|
||
|
@conv_threads_index[th.id] = name
|
||
|
th
|
||
|
end
|
||
|
|
||
|
def set_fatal
|
||
|
@fatal_error = true
|
||
|
def thread_by_id(id)
|
||
|
name = @conv_threads_index[id]
|
||
|
name.nil? ? ConversationThread.new(this, "noname/#{index}") : @conv_threads[name]
|
||
|
end
|
||
|
|
||
|
def current_session
|
||
|
@sessions[@thread_id]
|
||
|
def send_message(message)
|
||
|
raise CyberError.new(:unrecoverable, "bot/conversation", "Cannot send message without action id") if message.action_id.nil?
|
||
|
|
||
|
send_line "#{@config.bot_id}-#{message.conv_thread.id}-#{message.action_id}" + (message.action_parameters.nil? : "" : "+") + " #{message.action_code}"
|
||
|
unless message.action_parameters.nil?
|
||
|
message.action_parameters.to_yaml.split.each {|l| send_line l }
|
||
|
send_line EOD
|
||
|
end
|
||
|
|
||
|
action_id
|
||
|
end
|
||
|
|
||
|
protected
|
||
|
|
||
|
def reply_syntax_error(msg = nil)
|
||
|
logger.error "Protocol error [#{identifier}]: syntax error (#{msg})"
|
||
|
set_error_status
|
||
|
|
||
|
msg = "syntax error" + (msg ? ": " + msg : "")
|
||
|
@protocol.send_error_protocol(msg)
|
||
|
end
|
||
|
|
||
|
def reply_fatal_error(msg = nil)
|
||
|
logger.error "Protocol error [#{identifier}]: fatal error (#{msg})"
|
||
|
set_error_status(true)
|
||
|
|
||
|
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)
|
||
|
def enter_split_mode(message)
|
||
|
if @split_data_mode
|
||
|
logger.error "Error [#{identifier}]: already in split mode"
|
||
|
send_line "551 protocol error"
|
||
|
reply_fatal_error "already in split mode"
|
||
|
@split_data_mode = false
|
||
|
@split_data_action = nil
|
||
|
@split_data_message = nil
|
||
|
else
|
||
|
logger.debug "Entered split mode for action '#{action}' [#{identifier}]"
|
||
|
logger.debug "Protocol info [#{identifier}]: entered split mode for action '#{message.action_id}'"
|
||
|
@split_data_mode = true
|
||
|
@split_data_action = action
|
||
|
@split_data_message = message
|
||
|
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"))
|
||
|
logger.debug "Protocol info [#{identifier}]: quit split mode for action '#{@split_data_message.action_id}'"
|
||
|
|
||
|
parameters = YAML.parse(@split_data.join("\n"))
|
||
|
reply_syntax_error("bad parameters format") if parameters.nil?
|
||
|
|
||
|
message = @split_data_message.conv_thread.new_message(@split_data_message.action_code, parameters, @split_data_message.action_id)
|
||
|
receive_message(message)
|
||
|
else
|
||
|
logger.error "Error [#{identifier}]: not in split mode"
|
||
|
send_line "551 protocol error"
|
||
|
reply_fatal_error "not in split mode"
|
||
|
end
|
||
|
@split_data_mode = false
|
||
|
@split_data_action = nil
|
||
|
@split_data_message = nil
|
||
|
@split_data = []
|
||
|
end
|
||
|
end
|
||
Also available in: Unified diff
[evol] conversation/bot protocol rework §2 (refs #30)