Project

General

Profile

« Previous | Next » 

Revision 45333094

Added by Marc Dequènes almost 11 years ago

  • ID 4533309490bb30fb5b2827d0f1140afa339799bf

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

View differences:

bin/mapmaker_client
38 38
      def start_work
39 39
        contact_peer("MapMaker") do |conv|
40 40
          if conv
41
            logger.info "Yo ! Conversation ready with MapMaker (#{conv.identifier}) !"
41
            logger.info "Yo ! Conversation ready with MapMaker (#{conv.peer_name}) !"
42 42
          else
43 43
            logger.error "Could not connect to MapMaker"
44
            stop
44 45
          end
45 46
        end
46 47
      end
lib/cyborghood/cyborg.rb
25 25
    include I18nTranslation
26 26
    bindtextdomain("cyborghood", {:path => Config::L10N_DIR, :charset => "UTF-8"})
27 27

  
28
    attr_reader :name
29

  
28 30
    def initialize
31
      @name = self.class.name.split("::").last
32

  
29 33
      @config = Config.instance
30
      @config.bot_id = self.class.name
31 34

  
32 35
      # setup logs
33 36
      unless @config.log.nil?
......
55 58
      EventMachine.run do
56 59
        start_work
57 60
      end
61
    rescue EventMachine::ConnectionNotBound
62
      logger.fatal "Internal error (EventMachine::ConnectionNotBound)"
58 63
    end
59 64

  
60 65
    def start_work
......
63 68

  
64 69
    def ask_to_stop
65 70
      logger.info "Bot was asked to stop..."
71
      stop
72
    end
73

  
74
    def stop
75
      logger.info "Bot stopping"
66 76
      EventMachine.stop
67 77
    end
68 78

  
lib/cyborghood/cyborg/botnet.rb
28 28
  end
29 29

  
30 30
  module BotNetClientUNIXSocket
31
    def identifier_prefix
32
      "unix_socket"
33
    end
34

  
31 35
    def peer_socket(peer)
32 36
      File.join(Config::RUN_DIR, peer.downcase + ".sock")
33 37
    end
......
37 41
        block.call @comm_list[peer]
38 42
      else
39 43
        begin
40
          EventMachine.connect_unix_domain(peer_socket(peer), Conversation, self, &block)
44
          EventMachine.connect_unix_domain(peer_socket(peer), Conversation, self, block)
41 45
        rescue
42 46
          block.call false
43 47
        end
lib/cyborghood/cyborg/botnet_server.rb
40 40
    def capabilities
41 41
      super + []
42 42
    end
43

  
44
    # TODO: should be common to the future BotClientUNIXSocket
45
    def identifier_prefix
46
      "unix_socket"
47
    end
48 43
  end
49 44

  
50 45
  module BotNetServer
lib/cyborghood/cyborg/conversation.rb
35 35
    end
36 36

  
37 37
    def new_message(action_code, parameters = nil, action_id = nil)
38
      Message.new(self, action_code, parameters)
38
      Message.new(self, action_code, parameters, action_id)
39 39
    end
40 40

  
41 41
    def next_action_id
......
50 50
  end
51 51

  
52 52
  class Message
53
    private_class_method :new
54

  
55
    attr_reader :conv_thread, :action_code, :action_parameters, :action_id
53
    attr_reader :conv_thread, :action_code, :action_parameters, :action_id, :reply_to_action_id
56 54

  
57 55
    def initialize(conv_thread, action_code, action_parameters = nil, action_id = nil)
58 56
      @conv_thread = conv_thread
59 57
      @action_code = action_code
60 58
      @action_parameters = action_parameters
61
      @action_id = action_id
59
      @reply_to_action_id = action_id
60

  
61
      @action_id = nil
62 62
    end
63 63

  
64 64
    def new?
65
      @reply_to_action_id.nil?
66
    end
67

  
68
    def sent?
65 69
      @action_id.nil?
66 70
    end
67 71

  
68 72
    def send
69
      raise CyberError.new(:unrecoverable, "bot/conversation", "Not sending twice the same message") unless self.new?
73
      raise CyberError.new(:unrecoverable, "bot/conversation", "Not sending twice the same message") unless self.sent?
70 74
      @action_id = @conv_thread.next_action_id
71 75
      @conv_thread.conversation.send_message(self)
72 76
    end
73 77

  
74
    def create_reply(action_code, parameters)
78
    def create_reply(action_code, parameters = nil)
75 79
      raise CyberError.new(:unrecoverable, "bot/conversation", "Cannot reply to a newly created message") if self.new?
76
      new(self, action_code, parameters, @action_id)
80
      self.class.new(@conv_thread, action_code, parameters, @action_id)
77 81
    end
78 82
  end
79 83

  
80 84
  class Conversation < EventMachine::Protocols::LineAndTextProtocol
81
    private_class_method :new
82

  
83 85
    # don't rely on EventMachine's default, it may change one day
84 86
    MaxLineLength = 16*1024
85 87

  
86 88
    EOD = "\033[0J"
87
    BOT_ID_PATTERN = "[a-zA-Z0-9]+"
89
    BOT_NAME_PATTERN = "[A-Z][a-zA-Z0-9]+"
88 90
    ACTION_WORD_PATTERN = "[a-zA-Z0-9]+"
89
    ACTION_PATTERN = "^(#{BOT_ID_PATTERN})(\d{4})-(\d{4})([+]?) (#{ACTION_WORD_PATTERN}( #{ACTION_WORD_PATTERN})*)$"
91
    ACTION_PATTERN = "^(#{BOT_NAME_PATTERN})-([0-9]{4})-([0-9]{4})([+]?) (#{ACTION_WORD_PATTERN}( #{ACTION_WORD_PATTERN})*)$"
90 92
    MAXIMUM_ERROR_COUNT = 3
91 93
    MAXIMUM_LINES = 1024
92 94

  
93
    attr_reader :bot
95
    attr_reader :bot, :peer_name, :peer_capabilities
94 96

  
95
    def initialize(bot, &block)
97
    def initialize(bot, block = nil)
96 98
      @bot = bot
97 99
      @comm_logic_block = block
98 100

  
......
100 102

  
101 103
      @config = Config.instance
102 104

  
103
      @receive_error_count = 0
105
      @error_count = 0
104 106
      @split_data_mode = false
105 107
      @split_data_message = nil
106 108
      @split_data = []
107 109

  
108 110
      # associated conversation threads
109 111
      @conv_threads = {}
112
      @conv_threads_index = {}
110 113
      # thread 0 is reserved
111 114
      @next_thread_id = 0
112 115
      @system_thread = self.thread('system')
113 116

  
114 117
      # post-negociation peer info
115
      @peer_id = nil
118
      @peer_name = nil
116 119
      @peer_capabilities = []
117 120

  
121
      @comm_stop = false
122

  
118 123
      @protocol = BotProtocol.new(self)
119 124
    end
120 125

  
121 126
    def post_init
122
      logger.debug "New conversation with #{identifier}"
127
      logger.info "New conversation with #{identifier}"
128
      @protocol.send_announce_helo unless @comm_logic_block.nil?
129
    rescue
130
      logger.error "Conversation post_init: " + $!
131
      connection_close
123 132
    end
124 133

  
125 134
    def receive_line(data)
126
      return if data.empty?
135
      return if @comm_stop or data.empty?
127 136

  
128 137
      clear_receive_info
129 138

  
......
143 152
          logger.debug "Received data [#{identifier}]: #{data}"
144 153

  
145 154
          if data =~ Regexp.new(ACTION_PATTERN)
146
            conv_thread_id = $1
147
            action_id = $2
148
            flags = $3 || ""
149
            action_code = $4
155
            # TODO: pass peer_name into Message and use it in Protocol to check replies match
156
            peer_name = $1
157
            conv_thread_id = $2.to_i
158
            action_id = $3.to_i
159
            flags = $4 || ""
160
            action_code = $5
150 161

  
151 162
            conv_thread = self.thread_by_id(conv_thread_id)
152 163
            message = conv_thread.new_message(action_code, nil, action_id)
......
162 173
        end
163 174
      end
164 175

  
165
      send_reply
166

  
167 176
      check_errors
168 177
    end
169 178

  
170 179
    def receive_message(message)
171
      logger.debug "Received message '#{action}' [#{identifier}]"
180
      logger.debug "Received message '#{message.action_code}' [#{identifier}]"
172 181

  
173 182
      @protocol.process_received_message(message)
174 183
    end
......
179 188
    end
180 189

  
181 190
    def unbind
182
      logger.debug "Conversation finished with #{identifier}"
183
      @bot.unregister_communication @peer_id unless @peer_id.nil?
184
      @conv_threads.each_values {|s| s.close }
191
      logger.info "Conversation finished with #{identifier} (#{@peer_name})"
192
      @bot.unregister_communication @peer_name unless @peer_name.nil?
193
      @conv_threads.each_value {|s| s.close }
194
      @comm_logic_block.call false unless @comm_logic_block.nil? or @protocol.negociation_ok?
185 195
    end
186 196

  
187 197
    def bye
188
      send_quit_leaving
198
      @protocol.send_quit_leaving
189 199
    end
190 200

  
191
    def set_peer_info(id, capabilities)
192
      @peer_id = id
201
    def set_peer_info(name, capabilities)
202
      @peer_name = name
193 203
      @peer_capabilities = capabilities || []
194
      @bot.register_communication @peer_id, self
204
      logger.info "Peer name for #{identifier}: #{@peer_name}"
205
      logger.info "Peer capabilities for #{identifier}: #{@peer_capabilities.join(", ")}"
206
      @bot.register_communication @peer_name, self
195 207
    end
196 208

  
197 209
    def set_comm_ready
198
      @comm_logic_block.call self
210
      logger.info "Protocol negociation with '#{@peer_name}' on #{identifier} succeeded"
211
      @comm_logic_block.call self unless @comm_logic_block.nil?
212
    end
213

  
214
    def set_comm_stop
215
      @comm_stop = true
216
      yield if block_given?
217
      close_connection_after_writing
199 218
    end
200 219

  
201 220
    def set_error_status(fatal = false)
......
216 235
    def send_message(message)
217 236
      raise CyberError.new(:unrecoverable, "bot/conversation", "Cannot send message without action id") if message.action_id.nil?
218 237

  
219
      send_line "#{@config.bot_id}-#{message.conv_thread.id}-#{message.action_id}" + (message.action_parameters.nil? ? "" : "+") + " #{message.action_code}"
238
      flags = ""
239
      flags += "+" unless message.action_parameters.nil?
240

  
241
      send_line sprintf("%s-%04d-%04d%s %s", @bot.name, message.conv_thread.id, message.action_id, flags, message.action_code)
220 242
      unless message.action_parameters.nil?
221
        message.action_parameters.to_yaml.split.each {|l| send_line l }
243
        message.action_parameters.to_yaml.each_line {|l| send_line l }
222 244
        send_line EOD
223 245
      end
224

  
225
      action_id
226 246
    end
227 247

  
228 248
    protected
......
249 269
        msg_quit = "previous fatal error"
250 270
      end
251 271

  
252
      unless msg_quit.nil?
253
        send_quit_decline msg_quit
254
        @comm_logic_block.call false
255
        close_connection_after_writing
256
      end
272
      @protocol.send_quit_decline msg_quit unless msg_quit.nil?
257 273
    end
258 274

  
259 275
    def new_thread(name, id = nil)
......
298 314
      if @split_data_mode
299 315
        logger.debug "Protocol info [#{identifier}]: quit split mode for action '#{@split_data_message.action_id}'"
300 316

  
301
        parameters = YAML.parse(@split_data.join("\n"))
317
        parameters = YAML.load(@split_data.join("\n"))
302 318
        reply_syntax_error("bad parameters format") if parameters.nil?
303 319

  
304
        message = @split_data_message.conv_thread.new_message(@split_data_message.action_code, parameters, @split_data_message.action_id)
320
        message = @split_data_message.conv_thread.new_message(@split_data_message.action_code, parameters, @split_data_message.reply_to_action_id)
305 321
        receive_message(message)
306 322
      else
307 323
        reply_fatal_error "not in split mode"
lib/cyborghood/cyborg/protocol.rb
21 21

  
22 22

  
23 23
module CyborgHood
24
  module BotProtocol
24
  class BotProtocol
25 25
    VERSION = "0.1"
26 26
    CAPABILITIES = []
27 27

  
......
63 63
      unless message.conv_thread.id == 0
64 64
        return send_quit_decline "bad negociation"
65 65
      end
66
      unless message.parameters[:bot_name] =~ Conversation::BOT_ID_PATTERN
66
      if message.action_parameters.nil?
67
        return send_quit_decline "missing parameters"
68
      end
69
      unless message.action_parameters[:bot_name] =~ Regexp.new(Conversation::BOT_NAME_PATTERN)
67 70
        return send_quit_decline "bad bot name"
68 71
      end
69
      unless message.parameters[:protocol_version] == VERSION
72
      unless message.action_parameters[:protocol_version] == VERSION
70 73
        return send_quit_decline "protocol version does not match"
71 74
      end
72 75
      @negociation_received = true
73
      @conversation.set_peer_info(message.parameters[:bot_name], message.parameters[:capabilities])
76
      @conversation.set_peer_info(message.action_parameters[:bot_name], message.action_parameters[:capabilities])
74 77

  
75 78
      if @negociation_sent
76 79
        send_announce_ok(message)
......
78 81
        @conversation.set_comm_ready
79 82
      else
80 83
        send_announce_helo(message)
81
        @negociation_sent = true
82 84
      end
83 85
    end
84 86

  
......
94 96
    end
95 97

  
96 98
    def receive_request_call(message)
97
      unless @conversation.bot.interface.is_node? message.parameters[:node]
99
      if message.action_parameters.nil?
100
        return send_error_action(message, "missing parameters")
101
      end
102
      unless @conversation.bot.interface.is_node? message.action_parameters[:node]
98 103
        return send_error_action(message, "bad node")
99 104
      end
100 105
      send_reply_ack(message)
......
104 109
        }
105 110
        begin
106 111
          result[:action_result] = @conversation.bot.interface.call(message.conv_thread.session,
107
                                                                    message.parameters[:node],
108
                                                                    message.parameters[:data])
112
                                                                    message.action_parameters[:node],
113
                                                                    message.action_parameters[:data])
109 114
        rescue
110 115
          result[:error] = $!
111 116
        end
......
113 118
    end
114 119

  
115 120
    def receive_request_exists(message)
116
      unless @conversation.bot.interface.is_node? message.parameters[:node]
121
      if message.action_parameters.nil?
122
        return send_error_action(message, "missing parameters")
123
      end
124
      unless @conversation.bot.interface.is_node? message.action_parameters[:node]
117 125
        return send_error_action(message, "bad node")
118 126
      end
119 127
      send_reply_ack(message)
120 128
      @conversation.bot.schedule_task(@@request_callback) do
121 129
        {
122 130
          :reply_message => message,
123
          :action_result => @conversation.bot.interface.has_node?(message.parameters[:node])
131
          :action_result => @conversation.bot.interface.has_node?(message.action_parameters[:node])
124 132
        }
125 133
      end
126 134
    end
......
132 140

  
133 141
    def send_announce_helo(recv_message = nil)
134 142
      action_code = "ANNOUNCE HELO"
135
      message = (recv_message.nil? ? @conversation.thread('system').new_message(action_code) :
136
                 recv_message.create_reply(action_code))
143
      action_parameters = {
144
        :bot_name         => @conversation.bot.name,
145
        :protocol_version => VERSION
146
      }
147

  
148
      message = (recv_message.nil? ? @conversation.thread('system').new_message(action_code, action_parameters) :
149
                 recv_message.create_reply(action_code, action_parameters))
137 150
      message.send
151
      @negociation_sent = true
138 152
    end
139 153

  
140 154
    def send_announce_ok(recv_message)
......
179 193
    end
180 194

  
181 195
    def send_quit_decline(reason)
182
      @conversation.thread('system').new_message("QUIT LEAVING", { :reason => reason }).send
196
      @conversation.set_comm_stop do
197
        @conversation.thread('system').new_message("QUIT DECLINE", { :reason => reason }).send
198
      end
183 199
    end
184 200

  
185 201
    def send_quit_leaving
186
      @conversation.thread('system').new_message("QUIT LEAVING").send
202
      @conversation.set_comm_stop do
203
        @conversation.thread('system').new_message("QUIT LEAVING").send
204
      end
187 205
    end
188 206
  end
189 207
end

Also available in: Unified diff