Fix

neurow/lib/neurow/internal_api/endpoint.ex

@@ -56,11 +56,15 @@ defmodule Neurow.InternalApi.Endpoint do
 
   get "/history/:topic" do
     history = Neurow.ReceiverShardManager.get_history(topic)
-    history = Enum.map(history, fn {_, message} -> %{
-      timestamp: message.timestamp,
-      payload: message.payload,
-      type: message.type,
-    } end)
+
+    history =
+      Enum.map(history, fn {_, message} ->
+        %{
+          timestamp: message.timestamp,
+          payload: message.payload,
+          type: message.type
+        }
+      end)
 
     conn
     |> put_resp_header("content-type", "application/json")
@@ -90,10 +94,13 @@ defmodule Neurow.InternalApi.Endpoint do
 
         conn
         |> put_resp_header("content-type", "application/json")
-        |> send_resp(200, Jason.encode!(%{
-          nb_published: nb_publish,
-          publish_timestamp: publish_timestamp
-        }))
+        |> send_resp(
+          200,
+          Jason.encode!(%{
+            nb_published: nb_publish,
+            publish_timestamp: publish_timestamp
+          })
+        )
 
       {:error, reason} ->
         conn |> send_bad_request(:invalid_payload, reason)

neurow/lib/neurow/receiver_shard.ex

@@ -34,7 +34,9 @@ defmodule Neurow.ReceiverShard do
 
     result = :ets.lookup(table_0, topic) ++ :ets.lookup(table_1, topic)
 
-    Enum.sort(result, fn {_, message0}, {_, message1} -> message0.timestamp < message1.timestamp end)
+    Enum.sort(result, fn {_, message0}, {_, message1} ->
+      message0.timestamp < message1.timestamp
+    end)
   end
 
   @impl true

neurow/test/neurow/history_integration_test.exs

@@ -28,9 +28,11 @@ defmodule Neurow.HistoryIntegrationTest do
 
     drop_one_message()
     drop_one_message()
-    [[_, id]] = Regex.scan(~r/id=(\d+)/, call.resp_body)
+
+    # Avoid to publish message in the same millisecond
     Process.sleep(2)
-    id
+    result = Jason.decode!(call.resp_body)
+    result["publish_timestamp"]
   end
 
   defp history(topic) do
@@ -111,7 +113,7 @@ defmodule Neurow.HistoryIntegrationTest do
     assert length(expected_history) == length(actual_history)
 
     Enum.each(0..(length(expected_history) - 1), fn index ->
-      assert Enum.at(expected_history, index) == Enum.at(actual_history, index)["message"]
+      assert Enum.at(expected_history, index) == Enum.at(actual_history, index)["payload"]
     end)
 
     drop_one_message()
@@ -172,7 +174,7 @@ defmodule Neurow.HistoryIntegrationTest do
 
     :ok = :httpc.cancel_request(request_id)
 
-    request_id = subscribe("bar", [{["Last-Event-ID"], first_id}])
+    request_id = subscribe("bar", [{["Last-Event-ID"], to_string(first_id)}])
     {:stream, msg} = next_message()
     assert msg == "id: #{second_id}\ndata: foo57\n\n"
 
@@ -183,7 +185,7 @@ defmodule Neurow.HistoryIntegrationTest do
     :ok = :httpc.cancel_request(request_id)
 
     # End of is history
-    request_id = subscribe("bar", [{["Last-Event-ID"], second_id}])
+    request_id = subscribe("bar", [{["Last-Event-ID"], to_string(second_id)}])
 
     assert_raise RuntimeError, ~r/^Timeout waiting for message$/, fn ->
       next_message()
@@ -207,7 +209,7 @@ defmodule Neurow.HistoryIntegrationTest do
       end)
 
     start = 11
-    request_id = subscribe("bar", [{["Last-Event-ID"], Enum.at(ids, start)}])
+    request_id = subscribe("bar", [{["Last-Event-ID"], to_string(Enum.at(ids, start))}])
     output = all_messages()
 
     expected =