Module: SDN::CLI::MQTT::Write

Defined in:
lib/sdn/cli/mqtt/write.rb

Overview

Writer loop that drains the outbound queue and manages retries and follow-up timing.

Instance Method Summary collapse

Instance Method Details

#write ⇒ void

This method returns an undefined value.

Continuously drains the prioritized queue and transmits SDN messages.



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
 
# File 'lib/sdn/cli/mqtt/write.rb', line 11

def write
  last_write_at = Process.clock_gettime(Process::CLOCK_MONOTONIC)

  loop do
    message_and_retries = nil
    @mutex.synchronize do
      # got woken up early by another command getting queued; spin
      while @response_pending
        remaining_wait = @response_pending - Process.clock_gettime(Process::CLOCK_MONOTONIC)
        if remaining_wait.negative?
          if @prior_message
            SDN.logger.info "Timed out waiting on response to #{@prior_message.message.inspect}; " \
                            "retries left: #{@prior_message.remaining_retries}"
          else
            SDN.logger.info "Timed out waiting on response"
          end
          @response_pending = nil
          @broadcast_pending = nil
          if @prior_message
            if @prior_message.remaining_retries.zero?
              handle_timeout_fallback(@prior_message.message)
            elsif Message.group_address?(@prior_message.message.src) && !@pending_group_motors.empty?
              SDN.logger.debug "Re-targetting group message to individual motors"
              @pending_group_motors.each do |addr|
                new_message = @prior_message.message.dup
                new_message.src = [0, 0, 1]
                new_message.dest = Message.parse_address(addr)
                @queue.push(MessageAndRetries.new(new_message,
                                                  @prior_message.remaining_retries,
                                                  @prior_message.priority))
              end
              @pending_group_motors = []
            else
              @queue.push(@prior_message)
            end
            @prior_message = nil
          end
        else
          @cond.wait(@mutex, remaining_wait)
        end
      end

      message_and_retries = @queue.shift
      if message_and_retries && (
        message_and_retries.message.ack_requested ||
        message_and_retries.message.class.name =~ /^SDN::Message::Get/)
        @response_pending = Process.clock_gettime(Process::CLOCK_MONOTONIC) + WAIT_TIME
        @pending_group_motors = if Message.group_address?(message_and_retries.message.src)
                                  group_addr = Message.print_address(message_and_retries.message.src).delete(
                                    "."
                                  )
                                  @groups[group_addr]&.motor_objects&.map(&:addr) || []
                                else
                                  []
                                end

        if message_and_retries.message.dest == BROADCAST_ADDRESS || (
          Message.group_address?(message_and_retries.message.src) &&
          message_and_retries.message.is_a?(Message::GetNodeAddr))
          @broadcast_pending = Process.clock_gettime(Process::CLOCK_MONOTONIC) + BROADCAST_WAIT
        end
      end

      # wait until there is a message
      if @response_pending
        message_and_retries.remaining_retries -= 1
        @prior_message = message_and_retries
      elsif message_and_retries
        @prior_message = nil
      elsif @auto_discover && @motors_found
        message_and_retries = MessageAndRetries.new(Message::GetNodeAddr.new, 1, 50)
        @motors_found = false
      # nothing pending to write, and motors found on the last iteration;
      # look for more motors
      else
        @cond.wait(@mutex)
      end
    end
    next unless message_and_retries

    message = message_and_retries.message
    # minimum time between messages
    now = Process.clock_gettime(Process::CLOCK_MONOTONIC)
    sleep_time = 0.1 - (now - last_write_at)
    sleep(sleep_time) if sleep_time.positive?
    @sdn.send(message)
    last_write_at = now
  end
rescue => e
  SDN.logger.fatal "Failure writing: #{e}: #{e.backtrace}"
  exit 1
end