Tortoise311 ve emqtt ile Elixir'de MQTT

Elixir'de iki sağlam MQTT istemcisi var: handler behaviour'ı ve otomatik yeniden bağlanma sunan, MQTT 3.1.1 için saf Elixir istemcisi Tortoise311 ve EMQX ekibinin MQTT 5 konuşan Erlang istemcisi emqtt. İkisi de bir OTP supervision tree'ye doğal biçimde oturur; böylece broker kesintisi çöken bir uygulamaya değil, bir yeniden başlatmaya dönüşür.

Bu rehberde sensör topic'lerine abone olan, bunları fonksiyon başlıklarında pattern matching ile ayıran ve QoS 1 ile publish eden küçük bir telemetry uygulaması kuracağız — önce Tortoise311 ile, ardından MQTT 5 için emqtt ile. Tüm örnekler broker.connectmq.com adresini kullanıyor; böylece kod yazarken trafiği online MQTT istemcisinde izleyebilirsiniz.

Hangi kütüphane?

Tortoise311, orijinal Tortoise kütüphanesinin bakımı süren bir fork'udur; API aynıdır, yalnızca modül adı değişmiştir. MQTT 5'e ihtiyacınız olup olmadığından emin değilseniz MQTT 5 ile 3.1.1 karşılaştırmasını okuyun — basit telemetri ve komutlar için 3.1.1 yeterlidir.

Projeyi hazırlama

mix new telemetry --sup
cd telemetry

defp deps do
  [
    {:tortoise311, "~> 0.12"},
    # Only needed for the MQTT 5 section. Pin a release tag in real projects.
    {:emqtt, github: "emqx/emqtt", system_env: [{"BUILD_WITHOUT_QUIC", "1"}]}
  ]
end

BUILD_WITHOUT_QUIC=1, emqtt'nin isteğe bağlı QUIC bağımlılığını atlar; aksi halde bir C toolchain'i ve CMake gerekir. mix deps.get çalıştırın ve IEx'i başlatmadan önce broker'ınızın kimlik bilgilerini export edin:

export MQTT_USER="username"
export MQTT_PASS="password"

Supervisor altında bir Tortoise311 bağlantısı

Bir Tortoise311 bağlantısı, supervision tree'nizdeki sıradan bir çocuktur. TCP soketinin sahibidir, keep alive ping'lerini gönderir, kendi kendine yeniden bağlanır ve yeniden abone olur.

defmodule Telemetry.Application do
  use Application

  @impl true
  def start(_type, _args) do
    children = [
      {Tortoise311.Connection,
       client_id: Telemetry.Mqtt.client_id(),
       server: {Tortoise311.Transport.Tcp, host: ~c"broker.connectmq.com", port: 1883},
       user_name: System.fetch_env!("MQTT_USER"),
       password: System.fetch_env!("MQTT_PASS"),
       keep_alive: 30,
       handler: {Telemetry.MqttHandler, []},
       subscriptions: [{"testmqtt/elixir/+/temp", 1}, {"testmqtt/elixir/+/cmd", 1}]}
    ]

    Supervisor.start_link(children, strategy: :one_for_one, name: Telemetry.Supervisor)
  end
end

defmodule Telemetry.Mqtt do
  # One client ID per node: two nodes sharing an ID kick each other off the broker.
  def client_id do
    {:ok, host} = :inet.gethostname()
    "testmqtt-elixir-#{host}"
  end

  def publish_temp(device, celsius) do
    payload = Jason.encode!(%{device: device, celsius: celsius})
    Tortoise311.publish_sync(client_id(), "testmqtt/elixir/#{device}/temp", payload,
      qos: 1, timeout: 5_000)
  end
end

Örnek JSON için Jason kullanıyor; bağımlılıklarınıza {:jason, "~> 1.4"} ekleyin ya da düz bir string kullanın. publish_sync/4, broker QoS 1 mesajını onaylayana (PUBACK) kadar bekler ve :ok döner; publish/4 ise hemen {:ok, ref} döner ve onayı daha sonra çağırana mesaj olarak gönderir. Client ID'ler göründüğünden daha önemlidir; client ID üreteci benzersiz bir şema seçmenize yardımcı olur.

Mesajları pattern matching ile işleme

Tortoise311 gelen her topic'i seviyelerden oluşan bir listeye böler; böylece yönlendirme sıradan bir Elixir pattern matching'ine dönüşür — string ayrıştırma yok, regex yok.

defmodule Telemetry.MqttHandler do
  use Tortoise311.Handler
  require Logger

  @impl true
  def init(args), do: {:ok, args}

  @impl true
  def connection(status, state) do
    # status is :up, :down or :terminating
    Logger.info("MQTT connection #{status}")
    {:ok, state}
  end

  @impl true
  def subscription(status, topic_filter, state) do
    Logger.info("subscription #{topic_filter}: #{inspect(status)}")
    {:ok, state}
  end

  @impl true
  def handle_message(["testmqtt", "elixir", device, "temp"], payload, state) do
    Logger.info("#{device} reports #{payload}")
    {:ok, state}
  end

  def handle_message(["testmqtt", "elixir", device, "cmd"], "reboot", state) do
    Logger.warning("reboot requested for #{device}")
    {:ok, state}
  end

  def handle_message(topic, _payload, state) do
    Logger.debug("ignored #{Enum.join(topic, "/")}")
    {:ok, state}
  end

  @impl true
  def terminate(_reason, _state), do: :ok
end

Abonelikteki + wildcard'ı fonksiyon başlığında device değişkenine dönüşür. Sona her zaman genel bir yakalama clause'u koyun: wildcard aboneliği beklemediğiniz topic'leri de getirebilir ve eksik bir clause bağlantı sürecinin içinde FunctionClauseError fırlatır. Filtreler sizin için yeniyse MQTT topic'leri ve wildcard'lar yazısı + ve # işaretlerinin nasıl eşleştiğini anlatır.

Uçtan uca test

Uygulamayı iex -S mix ile başlatın. MQTT connection up ve iki adet subscription ... :ok satırı görmelisiniz. Ardından başka bir terminalden publish edin:

mosquitto_pub -h broker.connectmq.com -u "$MQTT_USER" -P "$MQTT_PASS" \
  -t testmqtt/elixir/sensor-1/temp -m '{"celsius":21.5}' -q 1
mosquitto_pub -h broker.connectmq.com -u "$MQTT_USER" -P "$MQTT_PASS" \
  -t testmqtt/elixir/sensor-1/cmd -m reboot -q 1

Diğer yönde de IEx'ten:

iex> Telemetry.Mqtt.publish_temp("sensor-2", 22.1)
:ok

Her iki mesajın geldiğini görmek için online istemcide testmqtt/elixir/# topic'ine abone olun. Supervision'ı kanıtlamak için bağlantı sürecini Process.exit(pid, :kill) ile öldürün (Supervisor.which_children(Telemetry.Supervisor) ile bulabilirsiniz) ve supervisor'ın kendiliğinden yeniden abone olan yeni bir süreç başlattığını izleyin.

Elixir topic'lerini tarayıcıda izleyin

8883 portunda TLS

Transport'u Tortoise311.Transport.SSL olarak değiştirin ve sistemin CA deposunu verin (OTP 25 veya üzeri):

server:
  {Tortoise311.Transport.SSL,
   host: ~c"broker.connectmq.com",
   port: 8883,
   verify: :verify_peer,
   cacerts: :public_key.cacerts_get()}

verify: :verify_peer ile Erlang'ın :ssl modülü hem sertifika zincirini hem de hostname'i doğrular. Production'da bir handshake hatasını asla :verify_none kullanarak "düzeltmeyin"; bunun yerine MQTT TLS bağlantısını test etme yazısında anlatıldığı gibi hostname'i ve CA'yı kontrol edin.

emqtt ile MQTT 5

emqtt gelen mesajları sahip sürece teslim eder; bu yüzden idiomatik sarmalayıcı, istemciyi start_link/1 ile başlatan bir GenServer'dır. İki süreç birbirine link'li olduğundan, kopan bir bağlantı sarmalayıcıyı da düşürür ve supervisor ikisini birden yeniden başlatır — yeniden bağlanma stratejisi olarak "let it crash".

defmodule Telemetry.Mqtt5 do
  use GenServer
  require Logger

  def start_link(opts), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  def publish(topic, payload), do: GenServer.call(__MODULE__, {:publish, topic, payload})

  @impl true
  def init(_opts) do
    {:ok, conn} =
      :emqtt.start_link(
        host: ~c"broker.connectmq.com",
        port: 1883,
        clientid: Telemetry.Mqtt.client_id() <> "-v5",
        username: System.fetch_env!("MQTT_USER"),
        password: System.fetch_env!("MQTT_PASS"),
        proto_ver: :v5,
        clean_start: false,
        properties: %{"Session-Expiry-Interval": 3600}
      )

    {:ok, _props} = :emqtt.connect(conn)
    {:ok, _props, _reason_codes} = :emqtt.subscribe(conn, "testmqtt/elixir/v5/#", qos: 1)
    {:ok, %{conn: conn}}
  end

  @impl true
  def handle_call({:publish, topic, payload}, _from, %{conn: conn} = state) do
    {:reply, :emqtt.publish(conn, topic, payload, 1), state}
  end

  @impl true
  def handle_info({:publish, %{topic: topic, payload: payload}}, state) do
    Logger.info("v5 #{topic}: #{payload}")
    {:noreply, state}
  end

  def handle_info({:disconnected, reason_code, _props}, state) do
    {:stop, {:disconnected, reason_code}, state}
  end
end

Telemetry.Mqtt5'i çocuk listesine ekleyin. clean_start: false ve bir saatlik session expiry, node yeniden başlarken broker'ın aboneliklerinizi ve kuyruktaki QoS 1 mesajlarını saklaması anlamına gelir — bkz. kalıcı oturumlar. Paylaşılan test endpoint'i çevrimdışı oturum saklamaz; bu bölümü kendi özel broker instance'ınızda deneyin. connect/1 başarısız olursa pattern match hata fırlatır; reason code nedenini söyler (:bad_username_or_password, :not_authorized vb.) ve MQTT reason code aracı her birini açıklar.

MQTT'yi daha büyük bir uygulamaya yerleştirme

Bir Phoenix veya Nerves projesinde MQTT bağlantısını iş mantığının yaşadığı yer olarak değil, sistemin bir kenarı olarak ele alın:

• Node başına, broker başına tek bağlantı. Publish etmek isteyen her süreç Telemetry.Mqtt.publish_temp/2'yi (veya benzer bir modülü) çağırır; hiçbiri kendi soketini açmaz.
• Mesajları BEAM içinde dağıtın. handle_message/3 içinde çözümlenmiş payload'u Phoenix.PubSub veya bir Registry üzerinden yayınlayın; böylece LiveView'lar, GenServer'lar ve pipeline'lar MQTT'nin varlığından habersiz yerel olarak abone olur.
• Kenarda çözümleyin. JSON'u handler'da bir kez parse edin ve içeriye struct'lar iletin; bozuk payload'ları çökmek yerine bir log satırıyla reddedin.
• Topic'leri yapılandırma yapın. Topic önekini config/runtime.exs içinde tutun; böylece staging ve production kodu paylaşır ama topic'leri asla paylaşmaz.

Bu test etmeyi de kolaylaştırır: birim testleri modüllerinizi doğrudan çağırır, tek bir entegrasyon testi ise gerçek broker'a publish edilen bir mesajın yerel PubSub'a ulaştığını kanıtlar.

Sık karşılaşılan sorunlar

• Bağlantı sürekli up ve down arasında gidip geliyor — başka bir node aynı client ID'yi kullanıyor. Her node'da :inet.gethostname() ile kontrol edin ve yinelenen client ID yazısına bakın.
• {:error, :econnrefused} veya hiç up olayı yok — yanlış port ya da bir güvenlik duvarı. 1883 düz TCP, 8883 TLS portudur; ikisini karıştırmak timeout gibi görünür.
• QoS 1 ile publish etmenize rağmen mesajlar QoS 0 ile geliyor — abonelik QoS 0 ile verilmiş. subscription/3 içindeki durumu veya :emqtt.subscribe/3'ün döndürdüğü reason code'ları inceleyin.
• handle_message/3 içinde FunctionClauseError — genel bir yakalama clause'u ekleyin; wildcard filtreler beklediğinizden fazlasını getirir.

Elixir'in süreç modeli MQTT istemcilerini olağanüstü dayanıklı kılar: broker başına supervisor altında tek bağlantı, pattern matching ile çözülen topic'ler ve handler dışına taşınan yavaş işler. Tortoise311 ile başlayın, MQTT 5 gerektiğinde emqtt'ye geçin ve test ederken topic'leri bir tarayıcı sekmesinde açık tutun.

Sıkça sorulan sorular