Added an async subscribe to topic to avoid issues in test

Author: rodrigo-o

lib/lambda_ethereum_consensus/p2p/gossip/attestation.ex

@@ -66,7 +66,7 @@ defmodule LambdaEthereumConsensus.P2P.Gossip.Attestation do
   def collect(subnet_id, attestation) do
     join(subnet_id)
     SubnetInfo.new_subnet_with_attestation(subnet_id, attestation)
-    Libp2pPort.subscribe_to_topic(topic(subnet_id), __MODULE__)
+    Libp2pPort.async_subscribe_to_topic(topic(subnet_id), __MODULE__)
   end
 
   @spec stop_collecting(non_neg_integer()) ::

lib/lambda_ethereum_consensus/p2p/gossip/beacon_block.ex

@@ -38,20 +38,19 @@ defmodule LambdaEthereumConsensus.P2P.Gossip.BeaconBlock do
     :ok
   end
 
-  # # TODO: Is anyone using this function?
-  # @spec subscribe_to_topic() :: :ok | :error
-  # def subscribe_to_topic() do
-  #   topic()
-  #   |> Libp2pPort.subscribe_to_topic(__MODULE__)
-  #   |> case do
-  #     :ok ->
-  #       :ok
+  @spec subscribe_to_topic() :: :ok | :error
+  def subscribe_to_topic() do
+    topic()
+    |> Libp2pPort.subscribe_to_topic(__MODULE__)
+    |> case do
+      :ok ->
+        :ok
 
-  #     {:error, reason} ->
-  #       Logger.error("[Gossip] Subscription failed: '#{reason}'")
-  #       :error
-  #   end
-  # end
+      {:error, reason} ->
+        Logger.error("[Gossip] Subscription failed: '#{reason}'")
+        :error
+    end
+  end
 
   def topic() do
     fork_context = ForkChoice.get_fork_digest() |> Base.encode16(case: :lower)

lib/lambda_ethereum_consensus/p2p/gossip/blob_sidecar.ex

@@ -26,20 +26,19 @@ defmodule LambdaEthereumConsensus.P2P.Gossip.BlobSideCar do
     end
   end
 
-  # TODO: Is anyone using this function?
-  # @spec subscribe_to_topics() :: :ok | {:error, String.t()}
-  # def subscribe_to_topics() do
-  #   Enum.each(topics(), fn topic ->
-  #     case Libp2pPort.subscribe_to_topic(topic, __MODULE__) do
-  #       :ok ->
-  #         :ok
+  @spec subscribe_to_topics() :: :ok | {:error, String.t()}
+  def subscribe_to_topics() do
+    Enum.each(topics(), fn topic ->
+      case Libp2pPort.subscribe_to_topic(topic, __MODULE__) do
+        :ok ->
+          :ok
 
-  #       {:error, reason} ->
-  #         Logger.error("[Gossip] Subscription failed: '#{reason}'")
-  #         {:error, reason}
-  #     end
-  #   end)
-  # end
+        {:error, reason} ->
+          Logger.error("[Gossip] Subscription failed: '#{reason}'")
+          {:error, reason}
+      end
+    end)
+  end
 
   def topics() do
     # TODO: this doesn't take into account fork digest changes

lib/lambda_ethereum_consensus/p2p/gossip/operations_collector.ex

@@ -40,18 +40,17 @@ defmodule LambdaEthereumConsensus.P2P.Gossip.OperationsCollector do
     "bls_to_execution_change"
   ]
 
-  # TODO: Is anyone using this function?
-  # def subscribe_to_topics() do
-  #   Enum.reduce_while(topics(), :ok, fn topic, _acc ->
-  #     case Libp2pPort.subscribe_to_topic(topic, __MODULE__) do
-  #       :ok ->
-  #         {:cont, :ok}
-
-  #       {:error, reason} ->
-  #         {:halt, {:error, "[OperationsCollector] Subscription failed: '#{reason}'"}}
-  #     end
-  #   end)
-  # end
+  def subscribe_to_topics() do
+    Enum.reduce_while(topics(), :ok, fn topic, _acc ->
+      case Libp2pPort.subscribe_to_topic(topic, __MODULE__) do
+        :ok ->
+          {:cont, :ok}
+
+        {:error, reason} ->
+          {:halt, {:error, "[OperationsCollector] Subscription failed: '#{reason}'"}}
+      end
+    end)
+  end
 
   @spec get_bls_to_execution_changes(non_neg_integer()) :: list(SignedBLSToExecutionChange.t())
   def get_bls_to_execution_changes(count) do

lib/libp2p_port.ex

@@ -256,6 +256,23 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
 
     GenServer.cast(pid, {:new_subscriber, topic_name, module})
 
+    call_command(pid, {:subscribe, %SubscribeToTopic{name: topic_name}})
+  end
+
+  @doc """
+  Subscribes to the given topic async, not waiting for a response at the subscribe.
+  After this, messages published to the topicwill be received by `self()`.
+  """
+  @spec async_subscribe_to_topic(GenServer.server(), String.t(), module()) ::
+          :ok | {:error, String.t()}
+  def async_subscribe_to_topic(pid \\ __MODULE__, topic_name, module) do
+    :telemetry.execute([:port, :message], %{}, %{
+      function: "async_subscribe_to_topic",
+      direction: "elixir->"
+    })
+
+    GenServer.cast(pid, {:new_subscriber, topic_name, module})
+
     cast_command(pid, {:subscribe, %SubscribeToTopic{name: topic_name}})
   end