fix: don't block GenServer.init/1

Changelog.md

@@ -6,6 +6,7 @@ _Released unreleased_
 
 - Emitting `off_broadway_emqtt.receive_message.ack` reads message topic from message receipt instead of from the message body.
  This ensures that topic is included in telemetry events even if the message has been altered during dispatch.
+- Move `emqtt.start_link/1` and `emqtt.connect/1` to a `handle_continue/2` callback to prevent blocking `GenServer.init/1`.
 
 ## v0.1.0 - Initial release
 

lib/off_broadway/emqqt/broker.ex

@@ -20,9 +20,7 @@ defmodule OffBroadway.EMQTT.Broker do
          {:ok, client_id} <- Keyword.fetch(config, :clientid),
          {:ok, buffer_size} <- Keyword.fetch(args, :buffer_size),
          {:ok, buffer_overflow} <- Keyword.fetch(args, :buffer_overflow_strategy),
-         {:ok, _message_handler} <- Keyword.fetch(args, :message_handler),
-         {:ok, emqtt} <- :emqtt.start_link(config),
-         {:ok, _props} <- :emqtt.connect(emqtt) do
+         {:ok, _message_handler} <- Keyword.fetch(args, :message_handler) do
       Process.flag(:trap_exit, true)
 
       {:ok,
@@ -33,8 +31,8 @@ defmodule OffBroadway.EMQTT.Broker do
          buffer_threshold: {20.0, 80.0},
          buffer_threshold_ref: nil,
          ets_table: String.to_existing_atom(client_id),
-         emqtt: emqtt,
-         emqtt_ref: Process.monitor(emqtt),
+         emqtt: nil,
+         emqtt_ref: nil,
          emqtt_config: config,
          topics: topics,
          topic_subscriptions: []
@@ -56,7 +54,18 @@ defmodule OffBroadway.EMQTT.Broker do
       {:read_concurrency, true}
     ])
 
-    {:noreply, state, {:continue, :subscribe_to_topics}}
+    {:noreply, state, {:continue, :connect_to_broker}}
+  end
+
+  def handle_continue(:connect_to_broker, state) do
+    with {:ok, pid} <- :emqtt.start_link(state.emqtt_config),
+         {:ok, _props} <- :emqtt.connect(pid) do
+      {:noreply, %{state | emqtt: pid, emqtt_ref: Process.monitor(pid)}, {:continue, :subscribe_to_topics}}
+    else
+      {:error, reason} ->
+        Logger.error("Failed to connect to MQTT broker: #{inspect(reason)}")
+        {:stop, :error}
+    end
   end
 
   def handle_continue(:subscribe_to_topics, state) do