Avoid connection pool issue between cloudflare and neurow (#41)

Author: bpaquet

neurow/integration_test/message_history_test.exs

@@ -73,23 +73,23 @@ defmodule Neurow.IntegrationTest.MessageHistoryTest do
             {"cache-control", "no-cache"},
             {"connection", "close"},
             {"content-type", "text/event-stream"},
-            {"transfer-encoding", "chunked"}
           ])
 
-          assert_receive %HTTPoison.AsyncChunk{chunk: error_sse_event}
+          assert_receive %HTTPoison.AsyncChunk{chunk: body}
 
-          json_event = parse_sse_json_event(error_sse_event)
+          json_event = parse_sse_json_event(body)
 
-          assert json_event.event == "neurow_error_bad_request"
+          assert json_event.event == "neurow_error_400"
 
-          assert json_event.data == %{
-                   "errors" => [
-                     %{
-                       "error_code" => "invalid_last_event_id",
-                       "error_message" => "Wrong value for last-event-id"
-                     }
-                   ]
-                 }
+          assert json_event.data ==
+                   %{
+                     "errors" => [
+                       %{
+                         "error_code" => "invalid_last_event_id",
+                         "error_message" => "Wrong value for last-event-id"
+                       }
+                     ]
+                   }
 
           assert_receive %HTTPoison.AsyncEnd{}
         end,

neurow/lib/neurow/public_api/endpoint.ex

@@ -61,7 +61,7 @@ defmodule Neurow.PublicApi.Endpoint do
           :error ->
             conn
             |> send_http_error(
-              :bad_request,
+              400,
               :invalid_last_event_id,
               "Wrong value for last-event-id"
             )
@@ -83,7 +83,7 @@ defmodule Neurow.PublicApi.Endpoint do
   end
 
   def send_forbidden(conn, error_code, error_message) do
-    send_http_error(conn, :forbidden, error_code, error_message)
+    send_http_error(conn, 403, error_code, error_message)
   end
 
   def preflight_request(conn, _opts) do
@@ -133,16 +133,13 @@ defmodule Neurow.PublicApi.Endpoint do
 
     now = :os.system_time(:seconds)
 
-    {:ok, conn} =
-      conn
-      |> put_resp_header("content-type", "text/event-stream")
-      |> put_resp_header("access-control-allow-origin", origin)
-      |> put_resp_header("cache-control", "no-cache")
-      |> put_resp_header("connection", "close")
-      |> send_chunked(http_status)
-      |> chunk("id:#{now}\nevent: neurow_error_#{http_status}\ndata: #{response}\n\n")
-
     conn
+    |> put_resp_header("content-type", "text/event-stream")
+    |> put_resp_header("access-control-allow-origin", origin)
+    |> send_resp(
+      http_status,
+      "id:#{now}\nevent: neurow_error_#{http_status}\ndata: #{response}\n\n"
+    )
   end
 
   defp extract_last_event_id(conn) do

neurow/test/neurow/public_api/endpoint_test.exs

@@ -12,21 +12,22 @@ defmodule Neurow.PublicApi.EndpointTest do
         conn(:get, "/v1/subscribe")
 
       call(Neurow.PublicApi.Endpoint, conn, fn ->
-        assert_receive {:send_chunked, 403}
-        assert_receive {:chunk, body}
-
-        sse_event = parse_sse_json_event(body)
-
-        assert sse_event.event == "neurow_error_forbidden"
-
-        assert sse_event.data == %{
-                 "errors" => [
-                   %{
-                     "error_code" => "invalid_authorization_header",
-                     "error_message" => "Invalid authorization header"
-                   }
-                 ]
-               }
+        assert_receive {:send_resp_status, 403}
+        assert_receive {:send_resp_body, body}
+
+        json_event = parse_sse_json_event(body)
+
+        assert json_event.event == "neurow_error_403"
+
+        assert json_event.data ==
+                 %{
+                   "errors" => [
+                     %{
+                       "error_code" => "invalid_authorization_header",
+                       "error_message" => "Invalid authorization header"
+                     }
+                   ]
+                 }
       end)
     end
 
@@ -36,13 +37,14 @@ defmodule Neurow.PublicApi.EndpointTest do
         |> put_req_header("authorization", "Bearer bad_token")
 
       call(Neurow.PublicApi.Endpoint, conn, fn ->
-        assert_receive {:send_chunked, 403}
-        assert_receive {:chunk, body}
-        event = parse_sse_json_event(body)
+        assert_receive {:send_resp_status, 403}
+        assert_receive {:send_resp_body, body}
+
+        json_event = parse_sse_json_event(body)
 
-        assert event.event == "neurow_error_forbidden"
+        assert json_event.event == "neurow_error_403"
 
-        assert event.data == %{
+        assert json_event.data == %{
                  "errors" => [
                    %{
                      "error_code" => "invalid_jwt_token",
@@ -122,13 +124,14 @@ defmodule Neurow.PublicApi.EndpointTest do
         )
 
       call(Neurow.PublicApi.Endpoint, conn, fn ->
-        assert_receive {:send_chunked, 400}
-        assert_receive {:chunk, body}
-        event = parse_sse_json_event(body)
+        assert_receive {:send_resp_status, 400}
+        assert_receive {:send_resp_body, body}
 
-        assert event.event == "neurow_error_bad_request"
+        json_event = parse_sse_json_event(body)
 
-        assert event.data == %{
+        assert json_event.event == "neurow_error_400"
+
+        assert json_event.data == %{
                  "errors" => [
                    %{
                      "error_code" => "invalid_last_event_id",
@@ -466,12 +469,22 @@ defmodule Neurow.PublicApi.EndpointTest do
         conn(:get, "/v1/subscribe")
 
       call(Neurow.PublicApi.Endpoint, conn, fn ->
-        assert_receive {:send_chunked, 403}
-        assert_receive {:chunk, body}
-
-        sse_event = parse_sse_event(body)
-
-        assert sse_event.event == "neurow_error_forbidden"
+        assert_receive {:send_resp_status, 403}
+        assert_receive {:send_resp_body, body}
+
+        json_event = parse_sse_json_event(body)
+
+        assert json_event.event == "neurow_error_403"
+
+        assert json_event.data ==
+                 %{
+                   "errors" => [
+                     %{
+                       "error_code" => "invalid_authorization_header",
+                       "error_message" => "Invalid authorization header"
+                     }
+                   ]
+                 }
       end)
     end
 

neurow/test/sse_helper.exs

@@ -58,12 +58,18 @@ defmodule SseHelper do
         send(state[:owner], {:chunk, body})
         Plug.Adapters.Test.Conn.chunk(state, body)
       end
+
+      def send_resp(state, status, headers, body) do
+        send(state[:owner], {:send_resp_status, status})
+        send(state[:owner], {:send_resp_body, body})
+        Plug.Adapters.Test.Conn.send_resp(state, status, headers, body)
+      end
     end
 
     @doc """
     Tests interaction with a SSE request by:
      - Calling the plug endpoint in a children task,
-     - Intercepts calls to `send_chunk` and `chunk` from the application plugs to send messages to the test process
+     - Intercepts calls to `send_chunk`, `chunk` and `send_resp` from the application plugs to send messages to the test process
     """
     def call(plug_endpoint, conn, assertion_fn, options \\ []) do
       instrumented_conn = conn |> instrument()