Class: Tina4::MqttMessage

Inherits:
Object
  • Object
show all
Defined in:
lib/tina4/mqtt_message.rb

Overview

One MQTT 3.1.1 application message as delivered by the broker.

Shaped like Tina4::Job (the Queue's unit of work): it carries the payload plus the delivery metadata a consumer needs, and it knows how to acknowledge itself back to the client it came from.

The two flags matter for correctness, not decoration:

retained?  — the broker replayed the topic's last known value to us
           because we subscribed AFTER it was published. It is current
           state, not a fresh event.
duplicate? — the DUP flag. The broker is REDELIVERING a QoS 1 message it
           never saw acknowledged. QoS 1 is at-least-once, so a
           duplicate is guaranteed eventually; a consumer that treats a
           DUP delivery as a new sample double-counts energy or mileage
           (test case A5). Key the ingest on (device_id,
           device_timestamp) and this is harmless.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(topic:, payload:, qos: 0, retained: false, duplicate: false, packet_id: nil, client: nil) ⇒ MqttMessage

Returns a new instance of MqttMessage.



24
25
26
27
28
29
30
31
32
33
34
 
# File 'lib/tina4/mqtt_message.rb', line 24

def initialize(topic:, payload:, qos: 0, retained: false, duplicate: false,
               packet_id: nil, client: nil)
  @topic = topic
  @payload = payload
  @qos = qos
  @retained = retained
  @duplicate = duplicate
  @packet_id = packet_id
  @client = client
  @acknowledged = false
end

Instance Attribute Details

#packet_id ⇒ Object (readonly)

Returns the value of attribute packet_id.



22
23
24
 
# File 'lib/tina4/mqtt_message.rb', line 22

def packet_id
  @packet_id
end

#payload ⇒ Object (readonly)

Returns the value of attribute payload.



22
23
24
 
# File 'lib/tina4/mqtt_message.rb', line 22

def payload
  @payload
end

#qos ⇒ Object (readonly)

Returns the value of attribute qos.



22
23
24
 
# File 'lib/tina4/mqtt_message.rb', line 22

def qos
  @qos
end

#topic ⇒ Object (readonly)

Returns the value of attribute topic.



22
23
24
 
# File 'lib/tina4/mqtt_message.rb', line 22

def topic
  @topic
end

Instance Method Details

#acknowledge ⇒ Object

Acknowledge a QoS 1 delivery (PUBACK) so the broker stops redelivering it. A QoS 0 message needs no acknowledgement, and a second call is a no-op, so this is always safe to call once processing succeeded.



50
51
52
53
54
55
 
# File 'lib/tina4/mqtt_message.rb', line 50

def acknowledge
  return false if @acknowledged || @qos.zero? || @packet_id.nil? || @client.nil?

  @client.acknowledge(@packet_id)
  @acknowledged = true
end

#acknowledged? ⇒ Boolean

True once this message has been acknowledged back to the broker.

Returns:

  • (Boolean) 


58
59
60
 
# File 'lib/tina4/mqtt_message.rb', line 58

def acknowledged?
  @acknowledged
end

#duplicate? ⇒ Boolean

True when the broker set the DUP flag — this is a REDELIVERY of a QoS 1 message we never acknowledged, not a new sample.

Returns:

  • (Boolean) 


43
44
45
 
# File 'lib/tina4/mqtt_message.rb', line 43

def duplicate?
  @duplicate
end

#retained? ⇒ Boolean

True when the broker replayed this as the topic's retained (last known) value.

Returns:

  • (Boolean) 


37
38
39
 
# File 'lib/tina4/mqtt_message.rb', line 37

def retained?
  @retained
end

#to_h ⇒ Object 



67
68
69
70
71
72
73
74
75
76
 
# File 'lib/tina4/mqtt_message.rb', line 67

def to_h
  {
    topic: @topic,
    payload: @payload,
    qos: @qos,
    retained: @retained,
    duplicate: @duplicate,
    packet_id: @packet_id
  }
end

#to_s ⇒ Object

The payload as a String — the common case, so "#{message}" just works.



63
64
65
 
# File 'lib/tina4/mqtt_message.rb', line 63

def to_s
  @payload.to_s
end