Class: Mqttopia::Client

Inherits:
Object
  • Object
show all
Defined in:
lib/mqttopia/client.rb

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(username: Mqttopia.username, password: Mqttopia.password, port: Mqttopia.port, ssl: Mqttopia.ssl) ⇒ Client

Returns a new instance of Client.



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
 
# File 'lib/mqttopia/client.rb', line 22

def initialize(
  username: Mqttopia.username,
  password: Mqttopia.password,
  port: Mqttopia.port,
  ssl: Mqttopia.ssl
)
  @client_id = SecureRandom.hex(3) + "-#{Mqttopia.client_id}"
  @mqtt_client = MQTT::Client.new(
    version: Mqttopia.version,
    port: port,
    username: username,
    password: password,
    ssl: ssl,
    client_id: @client_id,
    keep_alive: Mqttopia.keep_alive,
    will_topic: Mqttopia.will_topic,
    will_retain: Mqttopia.will_retain,
    will_payload: Mqttopia.will_payload,
    will_qos: Mqttopia.will_qos,
    ack_timeout: Mqttopia.ack_timeout,
    connect_timeout: Mqttopia.connect_timeout,
    max_retries: Mqttopia.max_retries,
    retry_interval: Mqttopia.retry_interval,
    logger: Mqttopia.logger
  )
  @subscriptions = {}
  @subscription_thread_mutex ||= Mutex.new
  @debugging = Mqttopia.debugging
  @debugging_topic = Mqttopia.debugging_topic
rescue StandardError => e
  log_error("initialize", e)
end

Instance Attribute Details

#client_id ⇒ Object (readonly)

Returns the value of attribute client_id.



12
13
14
 
# File 'lib/mqttopia/client.rb', line 12

def client_id
  @client_id
end

#debugging ⇒ Object (readonly)

Returns the value of attribute debugging.



12
13
14
 
# File 'lib/mqttopia/client.rb', line 12

def debugging
  @debugging
end

#debugging_topic ⇒ Object (readonly)

Returns the value of attribute debugging_topic.



12
13
14
 
# File 'lib/mqttopia/client.rb', line 12

def debugging_topic
  @debugging_topic
end

#mqtt_client ⇒ Object (readonly)

Returns the value of attribute mqtt_client.



12
13
14
 
# File 'lib/mqttopia/client.rb', line 12

def mqtt_client
  @mqtt_client
end

Class Method Details

.instance ⇒ Object 



14
15
16
17
18
 
# File 'lib/mqttopia/client.rb', line 14

def self.instance
  @instance_mutex.synchronize do
    @instance ||= new
  end
end

.publish(topic, message = nil, qos = 0) ⇒ Object 



85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
 
# File 'lib/mqttopia/client.rb', line 85

def self.publish(topic, message = nil, qos = 0)
  # Publishes a message to a topic.
  #
  client = new
  # @param message [String, NilClass] Message to publish, defaults to nil
  # @param qos [Integer] Quality of service, defaults to 0
  #
  # @return [Boolean] True if publish succeeded, false otherwise
  client.connect

  client.mqtt_client.publish(topic, message, false, qos)
  client.disconnect

  Mqttopia::Logger.debug("\n#{message}\n") if client.debugging
  true
rescue StandardError => e
  Mqttopia::Logger.error("\nMqttopia::Client -> publish: #{e}\n")
  false
end

Instance Method Details

#connect ⇒ Object 



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
 
# File 'lib/mqttopia/client.rb', line 55

def connect
  retry_count = 0
  max_retries = 15

  loop do
    host = active_host(Mqttopia.hosts)
    raise ArgumentError, "No hosts available" if host.nil?

    mqtt_client.host = host
    mqtt_client.connect(client_id)

    retry_count = 0 # Reset retry count on successful connection
    break # Exit the loop once connected
  rescue StandardError => e
    retry_count += 1
    disconnect_and_log("connect", e)

    if retry_count >= max_retries
      log_error("connect", "Terminating connection attempts after #{max_retries} failures")
      break # Stop retrying after max retries
    end

    sleep(2 * retry_count) # Delay before retrying to prevent rapid retries
  end
end

#disconnect ⇒ Object 



115
116
117
 
# File 'lib/mqttopia/client.rb', line 115

def disconnect
  mqtt_client.disconnect if connected?
end

#publish(topic, message = nil, qos = 0) ⇒ Object 



81
82
83
 
# File 'lib/mqttopia/client.rb', line 81

def publish(topic, message = nil, qos = 0)
  self.class.publish(topic, message, qos)
end

#ready_to_connect? ⇒ Boolean

Returns:

  • (Boolean) 


119
120
121
 
# File 'lib/mqttopia/client.rb', line 119

def ready_to_connect?
  !active_host.nil?
end

#subscribe(topic, listener_event = nil, qos = 0, &block) ⇒ Object 



105
106
107
108
109
110
111
112
113
 
# File 'lib/mqttopia/client.rb', line 105

def subscribe(topic, listener_event = nil, qos = 0, &block)
  return unless ensure_connection
  raise StandardError, "Blocked Topic" if [debugging_topic].include?(topic)

  mqtt_client.subscribe(topic => qos)
  register_subscription(listener_event, &block)
rescue StandardError => e
  disconnect_and_log("subscribe", e)
end