emqtt.ex

# Connect and subscribe

def init(_args) do
  Logger.info "Starting emqtt..."

  {:ok, pid} = :emqtt.start_link(
    clientid: "emqtt",
    host: 'localhost',
    # clean_start: false,
  )

  {:ok, _props} = :emqtt.connect(pid)

  Logger.info "Connected"

  topic = "fan_mode"
  {:ok, _props, _reason} = :emqtt.subscribe(pid, %{}, [{topic, qos: 1}])

  Logger.info "Subscribed to #{topic}"

  state = %{pid: pid}

  {:ok, state}
end

# Publish

def handle_call({:publish, topic, message, qos}, _from, state) do
  payload = Jason.encode!(message)

  case :emqtt.publish(state.emqtt_pid, topic, %{}, payload, [qos: qos]) do
    :ok      -> {:reply, :ok, state}
    {:ok, _} -> {:reply, :ok, state}
    error    -> raise "Publish error: #{inspect error}"
  end
end

# If using HiveMQ's cloud broker:

{:ok, pid} = :emqtt.start_link(
  clientid: "emqtt",
  host: '<hivemq uri>',
  port: 8883,
  ssl: true,
  ssl_opts: [
    server_name_indication: '<hivemq uri>'
    cacertfile: :certifi.cacertfile(),
    verify: :verify_none,
  ],
  # clean_start: false,
  username: "",
  password: ""
)