Pipeline support

lib/srh/http/base_router.ex

@@ -12,33 +12,55 @@ defmodule Srh.Http.BaseRouter do
   end
 
   post "/" do
+    conn
+    |> handle_extract_auth(&(CommandHandler.handle_command(conn, &1)))
+    |> handle_response(conn)
+  end
+
+  post "/pipeline" do
+    conn
+    |> handle_extract_auth(
+         &(CommandHandler.handle_command_array(conn, &1))
+       )
+    |> handle_response(conn)
+  end
+
+  match _ do
+    send_resp(conn, 404, "Endpoint not found")
+  end
+
+  defp handle_extract_auth(conn, success_lambda) do
     case conn
          |> get_req_header("authorization")
          |> RequestValidator.validate_bearer_header()
       do
       {:ok, token} ->
-        CommandHandler.handle_command(conn, token)
+        success_lambda.(token)
       {:error, _} ->
         {:malformed_data, "Missing/Invalid authorization header"}
     end
-    |> handle_response(conn)
-  end
-
-  match _ do
-    send_resp(conn, 404, "Endpoint not found")
   end
 
   defp handle_response(response, conn) do
-    %{code: code, message: message} =
+    %{code: code, message: message, json: json} =
       case response do
-        {:ok, data} -> %{code: 200, message: Jason.encode!(data)}
-        {:not_found, message} -> %{code: 404, message: message}
-        {:malformed_data, message} -> %{code: 400, message: message}
-        {:not_authorized, message} -> %{code: 401, message: message}
-        {:server_error, _} -> %{code: 500, message: "An error occurred internally"}
+        {:ok, data} -> %{code: 200, message: Jason.encode!(data), json: true}
+        {:not_found, message} -> %{code: 404, message: message, json: false}
+        {:malformed_data, message} -> %{code: 400, message: message, json: false}
+        {:not_authorized, message} -> %{code: 401, message: message, json: false}
+        {:server_error, _} -> %{code: 500, message: "An error occurred internally", json: false}
+        other ->
+          IO.inspect(other)
+          %{code: 500, message: "An error occurred internally", json: false}
       end
 
-    conn
+    case json do
+      true ->
+        conn
+        |> put_resp_header("content-type", "application/json")
+      false ->
+        conn
+    end
     |> send_resp(code, message)
   end
 end

lib/srh/http/command_handler.ex

@@ -14,6 +14,16 @@ defmodule Srh.Http.CommandHandler do
     end
   end
 
+  def handle_command_array(conn, token) do
+    case RequestValidator.validate_pipeline_redis_body(conn.body_params) do
+      {:ok, array_of_command_arrays} ->
+        IO.inspect(array_of_command_arrays)
+        do_handle_command_array(array_of_command_arrays, token)
+      {:error, error_message} ->
+        {:malformed_data, error_message}
+    end
+  end
+
   defp do_handle_command(command_array, token) do
     case TokenResolver.resolve(token) do
       {:ok, connection_info} ->
@@ -22,6 +32,31 @@ defmodule Srh.Http.CommandHandler do
     end
   end
 
+  defp do_handle_command_array(array_of_command_arrays, token) do
+    case TokenResolver.resolve(token) do
+      {:ok, connection_info} ->
+        dispatch_command_array(array_of_command_arrays, connection_info)
+      {:error, msg} -> {:not_authorized, msg}
+    end
+  end
+
+  defp dispatch_command_array([current | rest], connection_info, responses \\ []) do
+    updated_responses = case dispatch_command(current, connection_info) do
+      {:ok, result_map} ->
+        [result_map | responses]
+      {:malformed_data, result_json} ->
+        # TODO: change up the chain to json this at the last moment, so this isn't here
+        [Jason.decode!(result_json) | responses]
+    end
+
+    dispatch_command_array(rest, connection_info, updated_responses)
+  end
+
+  defp dispatch_command_array([], connection_info, responses) do
+    # The responses will be in reverse order, as we're adding them to the list with the faster method of putting them at head.
+    {:ok, Enum.reverse(responses)}
+  end
+
   defp dispatch_command(command_array, %{"srh_id" => srh_id, "max_connections" => max_connections} = connection_info)
        when is_number(max_connections) do
     case GenRegistry.lookup_or_start(Client, srh_id, [max_connections, connection_info]) do

lib/srh/http/request_validator.ex

@@ -4,6 +4,24 @@ defmodule Srh.Http.RequestValidator do
   def validate_redis_body(payload),
       do: {:error, "Invalid command array. Expected a string array at root of the command and its arguments."}
 
+  def validate_pipeline_redis_body(%{"_json" => array_of_command_arrays}) when is_list(array_of_command_arrays) do
+    do_validate_pipeline_redis_body(array_of_command_arrays, array_of_command_arrays)
+  end
+
+  # any amount of items left
+  defp do_validate_pipeline_redis_body([first_item | rest], original) do
+    case do_validate_pipeline_item(first_item) do
+      :ok -> do_validate_pipeline_redis_body(rest, original)
+      :error -> {:error, "Invalid command array. Expected an array of string arrays at root."}
+    end
+  end
+
+  defp do_validate_pipeline_redis_body([], original), do: {:ok, original}
+
+  defp do_validate_pipeline_item(item) when is_list(item), do: :ok
+
+  defp do_validate_pipeline_item(item), do: :error
+
   def validate_bearer_header(header_value_array) when is_list(header_value_array) do
     do_validate_bearer_header(header_value_array)
   end
@@ -18,7 +36,7 @@ defmodule Srh.Http.RequestValidator do
         do_validate_bearer_header(rest)
     end
   end
-  
+
   # no items left
   defp do_validate_bearer_header([]), do: {:error, :not_found}
 end