class MqttRails::Sender
Attributes
last_packet_sent_at[R]
last_pingreq_sent_at[R]
Public Class Methods
new(ack_timeout)
click to toggle source
# File lib/mqtt_rails/sender.rb, line 21 def initialize(ack_timeout) @socket = nil @writing_queue = [] @publish_queue = [] @publish_mutex = Mutex.new @writing_mutex = Mutex.new @ack_timeout = ack_timeout end
Public Instance Methods
append_to_writing(packet)
click to toggle source
# File lib/mqtt_rails/sender.rb, line 66 def append_to_writing(packet) begin if packet.is_a?(MqttRails::Packet::Publish) prepare_sending(@publish_queue, @publish_mutex, MAX_PUBLISH, packet) else prepare_sending(@writing_queue, @writing_mutex, MAX_QUEUE, packet) end rescue FullWritingException sleep SELECT_TIMEOUT retry end MQTT_ERR_SUCCESS end
check_ack_alive(queue, mutex)
click to toggle source
# File lib/mqtt_rails/sender.rb, line 115 def check_ack_alive(queue, mutex) mutex.synchronize do now = Time.now queue.each do |pck| if now >= pck[:timestamp] + @ack_timeout pck[:packet].dup ||= true unless pck[:packet].class == MqttRails::Packet::Subscribe || pck[:packet].class == MqttRails::Packet::Unsubscribe Rails.logger.info("[MQTT RAILS][INFO] Acknowledgement timeout is over, resending #{pck[:packet].inspect}") send_packet(pck[:packet]) pck[:timestamp] = now end end end end
flush_waiting_packet(sending=true)
click to toggle source
# File lib/mqtt_rails/sender.rb, line 98 def flush_waiting_packet(sending=true) if sending @writing_mutex.synchronize do @writing_queue.each do |packet| send_packet(packet) end end @publish_mutex.synchronize do @publish_queue.each do |packet| send_packet(packet) end end end @writing_queue = [] @publish_queue = [] end
prepare_sending(queue, mutex, max_packet, packet)
click to toggle source
# File lib/mqtt_rails/sender.rb, line 55 def prepare_sending(queue, mutex, max_packet, packet) if queue.length < max_packet mutex.synchronize do queue.push(packet) end else Rails.logger.error('[MQTT RAILS][ERROR] Writing queue is full, slowing down') raise FullWritingException end end
send_packet(packet)
click to toggle source
# File lib/mqtt_rails/sender.rb, line 34 def send_packet(packet) begin unless @socket.nil? || @socket.closed? @socket.write(packet.to_s) @last_packet_sent_at = Time.now MQTT_ERR_SUCCESS else MQTT_ERR_FAIL end end rescue StandardError raise WritingException rescue IO::WaitWritable IO.select(nil, [@socket], nil, SELECT_TIMEOUT) retry end
send_pingreq()
click to toggle source
# File lib/mqtt_rails/sender.rb, line 51 def send_pingreq @last_pingreq_sent_at = Time.now if send_packet(MqttRails::Packet::Pingreq.new) == MQTT_ERR_SUCCESS end
socket=(socket)
click to toggle source
# File lib/mqtt_rails/sender.rb, line 30 def socket=(socket) @socket = socket end
writing_loop()
click to toggle source
# File lib/mqtt_rails/sender.rb, line 80 def writing_loop @writing_mutex.synchronize do MAX_QUEUE.times do break if @writing_queue.empty? packet = @writing_queue.shift send_packet(packet) end end @publish_mutex.synchronize do MAX_PUBLISH.times do break if @publish_queue.empty? packet = @publish_queue.shift send_packet(packet) end end MQTT_ERR_SUCCESS end