feat: Make a test work

lib/client/lifecycle.ex

@@ -19,7 +19,7 @@ defmodule RabbitMQStream.Client.Lifecycle do
   How each command is handled by the Client
   - `:connect`: NOOP
   - `:close`: NOOP
-  - `:create_stream`:
+  - `:create_stream`: (Issue: Creates the stream on node)
   - `:delete_stream`:
   - `:query_offset`: Forward to 'control' connection
   - `:delete_producer`:
@@ -36,7 +36,7 @@ defmodule RabbitMQStream.Client.Lifecycle do
   - `:query_producer_sequence`: Forward to 'control' connection
   - `:create_super_stream`:
   - `:declare_producer`: Spawns a new connection
-  - `:publish`:
+  - `:publish`: Forwards it to the correct broker
 
   """
 
@@ -142,6 +142,13 @@ defmodule RabbitMQStream.Client.Lifecycle do
     {:reply, :ok, conn}
   end
 
+  @impl true
+  def handle_cast({:publish, opts}, %Client{} = conn) do
+    {_client_pid, broker_pid, _args} = Map.get(conn.clients, opts[:producer_id])
+    GenServer.cast(broker_pid, {:publish, opts})
+    {:noreply, conn}
+  end
+
   @impl true
   def handle_info({:DOWN, ref, :process, _pid, _reason}, %Client{} = conn) do
     case Map.get(conn.clients, ref) do

lib/connection/connection.ex

@@ -303,7 +303,7 @@ defmodule RabbitMQStream.Connection do
              is_binary(stream_name) do
     GenServer.call(
       server,
-      {:declare_producer, stream_name: stream_name, producer_reference: producer_reference}
+      {:declare_producer, stream_name: stream_name, producer_reference: producer_reference, pid: self()}
     )
   end
 

test/client_test.exs

@@ -20,11 +20,8 @@ defmodule RabbitMQStreamTest.ClientTest do
     assert :ok = RabbitMQStream.Connection.connect(conn)
 
     {:ok, client} = RabbitMQStream.Client.start_link(host: "localhost")
-    dbg(client)
 
     RabbitMQStream.Connection.create_stream(conn, "stream1")
-    RabbitMQStream.Connection.create_stream(conn, "stream2")
-    RabbitMQStream.Connection.create_stream(conn, "stream3")
 
     {:ok, _subscription_id} = RabbitMQStream.Connection.subscribe(client, "stream1", self(), :next, 999)
 
@@ -34,5 +31,11 @@ defmodule RabbitMQStreamTest.ClientTest do
         reference_name: "client-producer",
         stream_name: "stream1"
       )
+
+    message = Jason.encode!(%{message: "Hello, world2!"})
+
+    ClientProducer.publish(message)
+
+    assert_receive {:deliver, %{osiris_chunk: %{data_entries: [^message]}}}, 500
   end
 end