commit e26c015fb860b80b174f901b037240d5f1c5826c
Author: Marc Dequènes (Duck) <Duck@DuckCorp.org>
Date:   Mon Nov 1 18:42:14 2010 +0100

    [evol] conversation/bot protocol rework §2 (refs #30)

diff --git a/lib/cyborghood/cyborg/conversation.rb b/lib/cyborghood/cyborg/conversation.rb
index 9bdd594..6b46122 100644
--- a/lib/cyborghood/cyborg/conversation.rb
+++ b/lib/cyborghood/cyborg/conversation.rb
@@ -19,14 +19,26 @@
 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)
@@ -34,19 +46,85 @@ module CyborgHood
         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
 
@@ -58,7 +136,8 @@ module CyborgHood
 
     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
 
@@ -70,18 +149,25 @@ module CyborgHood
       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)
@@ -96,8 +182,7 @@ module CyborgHood
     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}]"
@@ -115,50 +200,39 @@ module CyborgHood
           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)
@@ -167,7 +241,7 @@ module CyborgHood
 
     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)
@@ -175,64 +249,81 @@ module CyborgHood
       @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
