Module: AutomateEm::DeviceConnection
- Included in:
- DatagramBase, AutomateEm::Device::Base
- Defined in:
- lib/automate-em/device/device_connection.rb
Instance Method Summary collapse
- #add_to_queue(command) ⇒ Object
-
#call_connected(*args) ⇒ Object
Connection state.
- #config=(options) ⇒ Object
- #default_send_options=(options) ⇒ Object
- #do_process_response(response, command) ⇒ Object
-
#do_receive_data(data) ⇒ Object
Data received Allow modules to set message delimiters for auto-buffering Default max buffer length == 1mb (setting can be overwritten) NOTE: The buffer cannot be defered otherwise there are concurrency issues.
-
#do_send_command(data, options = {}, *args, &block) ⇒ Object
Processes sends in strict order.
- #initialize(parent) ⇒ Object
-
#logger ⇒ Object
—————————————————————- Everything below here is called from a deferred thread.
- #process_next_send(command) ⇒ Object
-
#process_response(response, command) ⇒ Object
Caled from recieve.
- #process_response_complete ⇒ Object
- #process_result(result) ⇒ Object
-
#process_send(command) ⇒ Object
this is on the reactor thread.
- #received_lock ⇒ Object
- #sending_timeout ⇒ Object
- #shutdown(system) ⇒ Object
Instance Method Details
#add_to_queue(command) ⇒ Object
544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 |
# File 'lib/automate-em/device/device_connection.rb', line 544 def add_to_queue(command) begin if @connected || @make_break if @com_paused && !@make_break # We are calling from connected function (and we are connected) command[:priority] -= (2 * @config[:priority_bonus]) # Double bonus elsif @make_break if !@connected && !@connecting EM.next_tick do do_connect end elsif @connected && @disconnecting EM.next_tick do add_to_queue(command) end return # Don't add to queue yet end end add = true if command[:name].present? name = command[:name] if @named_commands[name].nil? @named_commands[name] = [[command[:priority]], command] #TODO:: we need to deal with the old commands emit values! elsif @named_commands[name][0][-1] > command[:priority] @named_commands[name][0].push(command[:priority]) @named_commands[name][1] = command #TODO:: we need to deal with the old commands emit values! else @named_commands[name][1] = command #TODO:: we need to deal with the old commands emit values! add = false end end @send_queue.push(command, command[:priority]) if add end rescue => e EM.defer do AutomateEm.print_error(logger, e, { :message => "module #{@parent.class} in device_connection.rb, send : something went terribly wrong to get here", :level => Logger::ERROR }) end end end |
#call_connected(*args) ⇒ Object
Connection state
594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 |
# File 'lib/automate-em/device/device_connection.rb', line 594 def call_connected(*args) # Called from a deferred thread # # NOTE:: Same as add parent in device module!!! # TODO:: Should break into a module and include it # @task_queue.push lambda { @parent[:connected] = true begin @send_monitor.mon_synchronize { # Any sends in here are high priority (no emits as this function must return) @parent.connected(*args) if @parent.respond_to?(:connected) } rescue => e # # save from bad user code (don't want to deplete thread pool) # AutomateEm.print_error(logger, e, { :message => "module #{@parent.class} error whilst calling: connect", :level => Logger::ERROR }) ensure EM.schedule do # # First connect if no commands pushed then we disconnect asap # if @make_break && @first_connect && @send_queue.size == 0 close_connection_after_writing @disconnecting = true @com_paused = true @first_connect = false elsif @com_paused @com_paused = false @wait_queue.push(nil) else EM.defer do logger.info "Reconnected, communications not paused." end end end end } end |
#config=(options) ⇒ Object
645 646 647 648 649 |
# File 'lib/automate-em/device/device_connection.rb', line 645 def config= () EM.schedule do @config.merge!() end end |
#default_send_options=(options) ⇒ Object
639 640 641 642 643 |
# File 'lib/automate-em/device/device_connection.rb', line 639 def () @status_lock.synchronize { @default_send_options.merge!() } end |
#do_process_response(response, command) ⇒ Object
308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 |
# File 'lib/automate-em/device/device_connection.rb', line 308 def do_process_response(response, command) return if @shutting_down.value @received_lock.synchronize { # This lock protects the send queue lock when we are emiting status @send_monitor.mon_synchronize { result = :abort begin if @parent.respond_to?(:received) if command.present? @parent.mark_emit_start(command[:emit]) if command[:emit].present? if command[:callback].present? result = command[:callback].call(response, command) # # The data may still be usefull # if [nil, :ignore].include?(result) @parent.received(response, nil) end else result = @parent.received(response, command) end else # logger.debug "Out of order response received for: #{@parent.class}" result = @parent.received(response, nil) end else if command.present? @parent.mark_emit_start(command[:emit]) if command[:emit].present? if command[:callback].present? result = command[:callback].call(response, command) else result = true end else result = true end end rescue => e # # save from bad user code (don't want to deplete thread pool) # This error should be logged in some consistent manner # AutomateEm.print_error(logger, e, { :message => "module #{@parent.class} error whilst calling: received", :level => Logger::ERROR }) ensure if command.present? @parent.mark_emit_end if command[:emit].present? end ActiveRecord::Base.clear_active_connections! end if command.present? && command[:wait] EM.schedule do process_result(result) end end } } end |
#do_receive_data(data) ⇒ Object
Data received Allow modules to set message delimiters for auto-buffering Default max buffer length == 1mb (setting can be overwritten) NOTE: The buffer cannot be defered otherwise there are concurrency issues
251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 |
# File 'lib/automate-em/device/device_connection.rb', line 251 def do_receive_data(data) @last_recieve_at = Time.now.to_f if @parent.respond_to?(:response_delimiter) begin if @buf.nil? del = @parent.response_delimiter if del.class == Array del = array_to_str(del) elsif del.class == Fixnum del = "" << del #array_to_str([del & 0xFF]) end @buf = BufferedTokenizer.new(del, @config[:max_buffer]) # Call back for character end data = @buf.extract(data) rescue => e @buf = nil # clear the buffer EM.defer do # Error in a thread AutomateEm.print_error(logger, e, { :message => "module #{@parent.class} error whilst setting delimiter", :level => Logger::ERROR }) end data = [data] end else data = [data] end if @waiting && data.length > 0 if @processing @receive_queue.push(*data) else @processing = true @timeout.cancel process_response(data.shift, @command) if data.length > 0 @receive_queue.push(*data) end end else data.each do |result| process_response(result, nil) end end end |
#do_send_command(data, options = {}, *args, &block) ⇒ Object
Processes sends in strict order
476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 |
# File 'lib/automate-em/device/device_connection.rb', line 476 def do_send_command(data, = {}, *args, &block) begin @status_lock.synchronize { = @default_send_options.merge() } # # Make sure we are sending appropriately formatted data # if data.class == Array data = array_to_str(data) elsif [:hex_string] == true data = hex_to_byte(data) end [:data] = data [:retries] = 0 if [:wait] == false if [:callback].nil? && (args.length > 0 || block.present?) [:callback] = args[0] unless args.empty? || args[0].class != Proc [:callback] = block unless block.nil? end if [:name].present? [:name] = [:name].to_sym end rescue => e AutomateEm.print_error(logger, e, { :message => "module #{@parent.class} in device_connection.rb, send : possible bad data or options hash", :level => Logger::ERROR }) return true end # # Use a monitor here to allow for re-entrant locking # This allows for a priority queue and we guarentee order of operations # bonus = false begin @send_monitor.mon_exit @send_monitor.mon_enter bonus = true rescue end EM.schedule do if bonus [:priority] -= @config[:priority_bonus] end add_to_queue() end return false rescue => e # # Save from a fatal error # AutomateEm.print_error(logger, e, { :message => "module #{@parent.class} in device_connection.rb, send : something went terribly wrong to get here", :level => Logger::ERROR }) return true end |
#initialize(parent) ⇒ Object
9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 |
# File 'lib/automate-em/device/device_connection.rb', line 9 def initialize( parent ) super @default_send_options = { :wait => true, # Wait for response :delay => 0, # Delay next send by x.y seconds :delay_on_recieve => 0, # Delay next send after a recieve by x.y seconds (only works when we are waiting for responses) #:emit :max_waits => 3, :callback => nil, # Alternative to the received function :retries => 2, :hex_string => false, :timeout => 5, # Timeout in seconds :priority => 50, :retry_on_disconnect => true, :force_disconnect => false # part of make and break options } @config = { :max_buffer => 524288, # 512kb :clear_queue_on_disconnect => false, :flush_buffer_on_disconnect => false, :priority_bonus => 20 } # # Queues # @task_queue = EM::Queue.new # basically we add tasks here that we want to run in a strict order (connect, disconnect) @receive_queue = EM::Queue.new # So we can process responses in different ways @wait_queue = EM::Queue.new @send_queue = EM::PriorityQueue.new(:fifo => true) {|x,y| x < y} # regular priority # # Named commands # Allowing for state control # @named_commands = {} # # Locks # @received_lock = Mutex.new @task_lock = Mutex.new @status_lock = Mutex.new @send_monitor = Object.new.extend(MonitorMixin) # # State # @connected = false @connecting = false @disconnecting = false @com_paused = true @command = nil @waiting = false @processing = false @last_sent_at = 0.0 @last_recieve_at = 0.0 @timeout = nil # # Configure links between objects (This is a very loose tie) # Relies on serial loading of modules # @parent = parent @parent.setbase(self) @tls_enabled = @parent.secure_connection if @parent.makebreak_connection @make_break = true @first_connect = true else @make_break = false end @make_occured = false @shutting_down = Atomic.new(false) # # Task event loop # @task_queue_proc = Proc.new do |task| if !@shutting_down.value EM.defer do begin @task_lock.synchronize { task.call } rescue => e AutomateEm.print_error(logger, e, { :message => "module #{@parent.class} in device_connection.rb, base : error in task loop", :level => Logger::ERROR }) ensure ActiveRecord::Base.clear_active_connections! @task_queue.pop &@task_queue_proc end end end end @task_queue.pop &@task_queue_proc # First task is ready # # send loop # @wait_queue_proc = Proc.new do |ignore| if ignore != :shutdown @send_queue.pop {|command| if command != :shutdown begin process = true if command[:name].present? begin name = command[:name] @named_commands[name][0].pop # Extract the command data command = @named_commands[name][1] if @named_commands[name][0].empty? # See if any more of these commands are queued @named_commands.delete(name) # Delete if there are not else @named_commands[name][1] = nil # Reset if there are end if command.nil? # decide if to continue or not command = {} process = false end rescue # # Retry (pop empty, lets let it have it) # end end if process if command[:delay] > 0.0 delay = @last_sent_at + command[:delay] - Time.now.to_f if delay > 0.0 EM.add_timer delay do process_send(command) end else process_send(command) end else process_send(command) end else process_next_send(command) end rescue => e EM.defer do AutomateEm.print_error(logger, e, { :message => "module #{@parent.class} in device_connection.rb, base : error in send loop", :level => Logger::ERROR }) end ensure ActiveRecord::Base.clear_active_connections! @wait_queue.pop &@wait_queue_proc end end } end end #@wait_queue.push(nil) Start paused @wait_queue.pop &@wait_queue_proc end |
#logger ⇒ Object
Everything below here is called from a deferred thread
464 465 466 |
# File 'lib/automate-em/device/device_connection.rb', line 464 def logger @parent.logger end |
#process_next_send(command) ⇒ Object
233 234 235 236 237 238 239 240 241 242 243 |
# File 'lib/automate-em/device/device_connection.rb', line 233 def process_next_send(command) if command[:force_disconnect] # Allow connection control close_connection_after_writing @disconnecting = true @com_paused = true else EM.next_tick do @wait_queue.push(nil) # Allows next response to process end end end |
#process_response(response, command) ⇒ Object
Caled from recieve
302 303 304 305 306 |
# File 'lib/automate-em/device/device_connection.rb', line 302 def process_response(response, command) EM.defer do do_process_response(response, command) end end |
#process_response_complete ⇒ Object
440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 |
# File 'lib/automate-em/device/device_connection.rb', line 440 def process_response_complete if (@make_break && @send_queue.empty?) || @command[:force_disconnect] if @connected close_connection_after_writing @disconnecting = true end @com_paused = true else EM.next_tick do @wait_queue.push(nil) end end @command = nil # free memory end |
#process_result(result) ⇒ Object
393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 |
# File 'lib/automate-em/device/device_connection.rb', line 393 def process_result(result) if [nil, :ignore].include?(result) && @command[:max_waits] > 0 @command[:max_waits] -= 1 if @receive_queue.size() > 0 @receive_queue.pop { |response| process_response(response, @command) } else @timeout = EM::Timer.new(@command[:timeout]) { sending_timeout } @processing = false end else if [false, :failed].include?(result) && @command[:retries] > 0 # assume command failed, we need to retry @command[:retries] -= 1 @send_queue.push(@command, @command[:priority] - @config[:priority_bonus]) end #else result == :abort || result == :success || result == true || waits and retries exceeded @receive_queue.size().times do @receive_queue.pop { |response| process_response(response, nil) } end @processing = false @waiting = false if @command[:delay_on_recieve] > 0.0 delay_for = (@last_recieve_at + @command[:delay_on_recieve] - Time.now.to_f) if delay_for > 0.0 EM.add_timer delay_for do process_response_complete end else process_response_complete end else process_response_complete end end end |
#process_send(command) ⇒ Object
this is on the reactor thread
189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 |
# File 'lib/automate-em/device/device_connection.rb', line 189 def process_send(command) # this is on the reactor thread begin if !error? && @connected do_send_data(command[:data]) @last_sent_at = Time.now.to_f @waiting = command[:wait] if @waiting @command = command @timeout = EM::Timer.new(command[:timeout]) { sending_timeout } else process_next_send(command) end else if @connected process_next_send(command) else if command[:retry_on_disconnect] || @make_break @send_queue.push(command, command[:priority] - (2 * @config[:priority_bonus])) # Double bonus end @com_paused = true end end rescue => e # # Save the thread in case of bad data in that send # EM.defer do AutomateEm.print_error(logger, e, { :message => "module #{@parent.class} in device_connection.rb, process_send : possible bad data", :level => Logger::ERROR }) end if @connected process_next_send(command) else @com_paused = true end end end |
#received_lock ⇒ Object
468 469 470 |
# File 'lib/automate-em/device/device_connection.rb', line 468 def received_lock @send_monitor # for monitor use end |
#sending_timeout ⇒ Object
372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 |
# File 'lib/automate-em/device/device_connection.rb', line 372 def sending_timeout @timeout = true if !@processing && @connected && @command.present? # Probably not needed... @processing = true # Ensure responses go into the queue command = @command[:data] process_result(:failed) EM.defer do logger.info "module #{@parent.class} timeout" logger.info "A response was not received for the command: #{command}" unless command.nil? end elsif !@connected && @command.present? && @command[:wait] if @command[:retry_on_disconnect] || @make_break @send_queue.push(@command, @command[:priority] - (2 * @config[:priority_bonus])) # Double bonus end @com_paused = true end end |
#shutdown(system) ⇒ Object
654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 |
# File 'lib/automate-em/device/device_connection.rb', line 654 def shutdown(system) if @parent.leave_system(system) == 0 @shutting_down.value = true close_connection @wait_queue.push(:shutdown) @send_queue.push(:shutdown, -32768) @task_queue.push(nil) EM.defer do begin @parent[:connected] = false # Communicator off at this point if @parent.respond_to?(:disconnected) @task_lock.synchronize { @parent.disconnected } end rescue => e # # save from bad user code (don't want to deplete thread pool) # AutomateEm.print_error(logger, e, { :message => "module #{@parent.class} error whilst calling: disconnected on shutdown", :level => Logger::ERROR }) ensure @parent.clear_active_timers ActiveRecord::Base.clear_active_connections! end end end end |