Current section
34 Versions
Jump to
Current section
34 Versions
Compare versions
14
files changed
+453
additions
-475
deletions
| @@ -15,7 +15,7 @@ Ravix is a in-development project to implement a client for the amazing [RavenDB | |
| 15 15 | Add Ravix to your mix.exs dependencies |
| 16 16 | |
| 17 17 | ```elixir |
| 18 | - {:ravix, "~> 0.2.3"} |
| 18 | + {:ravix, "~> 0.3.0"} |
| 19 19 | ``` |
| 20 20 | |
| 21 21 | ## Setting up your Repository |
| @@ -17,6 +17,7 @@ | |
| 17 17 | <<"lib/connection/models/server_node.ex">>, |
| 18 18 | <<"lib/connection/models/connection_state.ex">>, |
| 19 19 | <<"lib/connection/connection.ex">>,<<"lib/connection/executor">>, |
| 20 | + <<"lib/connection/executor/request_executor_worker.ex">>, |
| 20 21 | <<"lib/connection/executor/supervisor.ex">>, |
| 21 22 | <<"lib/connection/executor/request_executor_option.ex">>, |
| 22 23 | <<"lib/connection/executor/request_executor.ex">>, |
| @@ -67,12 +68,12 @@ | |
| 67 68 | {<<"name">>,<<"vex">>}, |
| 68 69 | {<<"optional">>,false}, |
| 69 70 | {<<"repository">>,<<"hexpm">>}, |
| 70 | - {<<"requirement">>,<<"~> 0.9.0">>}], |
| 71 | + {<<"requirement">>,<<"~> 0.9">>}], |
| 71 72 | [{<<"app">>,<<"mappable">>}, |
| 72 73 | {<<"name">>,<<"mappable">>}, |
| 73 74 | {<<"optional">>,false}, |
| 74 75 | {<<"repository">>,<<"hexpm">>}, |
| 75 | - {<<"requirement">>,<<"~> 0.2.4">>}], |
| 76 | + {<<"requirement">>,<<"~> 0.2">>}], |
| 76 77 | [{<<"app">>,<<"elixir_uuid">>}, |
| 77 78 | {<<"name">>,<<"elixir_uuid">>}, |
| 78 79 | {<<"optional">>,false}, |
| @@ -87,7 +88,7 @@ | |
| 87 88 | {<<"name">>,<<"morphix">>}, |
| 88 89 | {<<"optional">>,false}, |
| 89 90 | {<<"repository">>,<<"hexpm">>}, |
| 90 | - {<<"requirement">>,<<"~> 0.8.1">>}], |
| 91 | + {<<"requirement">>,<<"~> 0.8">>}], |
| 91 92 | [{<<"app">>,<<"timex">>}, |
| 92 93 | {<<"name">>,<<"timex">>}, |
| 93 94 | {<<"optional">>,false}, |
| @@ -102,7 +103,7 @@ | |
| 102 103 | {<<"name">>,<<"retry">>}, |
| 103 104 | {<<"optional">>,false}, |
| 104 105 | {<<"repository">>,<<"hexpm">>}, |
| 105 | - {<<"requirement">>,<<"~> 0.16.0">>}], |
| 106 | + {<<"requirement">>,<<"~> 0.16">>}], |
| 106 107 | [{<<"app">>,<<"inflex">>}, |
| 107 108 | {<<"name">>,<<"inflex">>}, |
| 108 109 | {<<"optional">>,false}, |
| @@ -112,5 +113,10 @@ | |
| 112 113 | {<<"name">>,<<"castore">>}, |
| 113 114 | {<<"optional">>,false}, |
| 114 115 | {<<"repository">>,<<"hexpm">>}, |
| 115 | - {<<"requirement">>,<<"~> 0.1.17">>}]]}. |
| 116 | - {<<"version">>,<<"0.2.3">>}. |
| 116 | + {<<"requirement">>,<<"~> 0.1">>}], |
| 117 | + [{<<"app">>,<<"poolboy">>}, |
| 118 | + {<<"name">>,<<"poolboy">>}, |
| 119 | + {<<"optional">>,false}, |
| 120 | + {<<"repository">>,<<"hexpm">>}, |
| 121 | + {<<"requirement">>,<<"~> 1.5">>}]]}. |
| 122 | + {<<"version">>,<<"0.3.0">>}. |
| @@ -24,21 +24,20 @@ defmodule Ravix.Connection.State.Manager do | |
| 24 24 | Logger.info("[RAVIX] Initializing connection for the repository '#{inspect(state.store)}'") |
| 25 25 | |
| 26 26 | OK.try do |
| 27 | - node_pids <- register_nodes(state) |
| 27 | + pools <- register_node_pools(state) |
| 28 28 | |
| 29 | - Logger.debug( |
| 30 | - "[RAVIX] Nodes with PIDs '#{inspect(node_pids)}' were registered successfully for the repository '#{inspect(state.store)}'" |
| 29 | + Logger.info( |
| 30 | + "[RAVIX] '#{length(pools)}' Node pools were registered successfully for the repository '#{inspect(state.store)}'" |
| 31 31 | ) |
| 32 32 | |
| 33 33 | _ = |
| 34 34 | case state.force_create_database do |
| 35 35 | true -> |
| 36 | - Logger.debug( |
| 36 | + Logger.info( |
| 37 37 | "[RAVIX] Forcing the database creation is enabled, it will create the #{inspect(state.database)} if it does not exists" |
| 38 38 | ) |
| 39 39 | |
| 40 | - node_pids |
| 41 | - |> Enum.at(0) |
| 40 | + NodeSelector.random_executor_for(state.store) |
| 42 41 | |> DatabaseMaintenance.create_database(state.database) |
| 43 42 | |
| 44 43 | false -> |
| @@ -67,8 +66,8 @@ defmodule Ravix.Connection.State.Manager do | |
| 67 66 | def update_topology(%ConnectionState{} = state) do |
| 68 67 | OK.for do |
| 69 68 | Logger.debug("[RAVIX] Updating the topology for the store '#{inspect(state.store)}'") |
| 70 | - current_node = NodeSelector.current_node(state) |
| 71 | - topology <- __MODULE__.request_topology([current_node], state.database) |
| 69 | + random_node = NodeSelector.random_executor_for(state.store) |
| 70 | + topology <- __MODULE__.request_topology(random_node, state.database) |
| 72 71 | _ = ExecutorSupervisor.update_topology(state.store, topology) |
| 73 72 | |
| 74 73 | state = put_in(state.topology_etag, topology.etag) |
| @@ -90,25 +89,14 @@ defmodule Ravix.Connection.State.Manager do | |
| 90 89 | - `{:ok, Ravix.Connection.Topology}` if the topology request was successful |
| 91 90 | - `{:error, :invalid_cluster_topology}` if it fails to pool the topology |
| 92 91 | """ |
| 93 | - @spec request_topology(list(pid()), String.t()) :: |
| 92 | + @spec request_topology(pid(), String.t()) :: |
| 94 93 | {:error, :invalid_cluster_topology} | {:ok, Ravix.Connection.Topology.t()} |
| 95 | - def request_topology(node_pids, database) do |
| 94 | + def request_topology(node_pid, database) do |
| 96 95 | Logger.debug( |
| 97 96 | "[RAVIX] Requesting the cluster topology for the database '#{inspect(database)}'" |
| 98 97 | ) |
| 99 98 | |
| 100 | - topology = |
| 101 | - node_pids |
| 102 | - |> Enum.map(fn node -> |
| 103 | - RequestExecutor.execute_for_node( |
| 104 | - %GetTopology{database_name: database}, |
| 105 | - node, |
| 106 | - database |
| 107 | - ) |
| 108 | - end) |
| 109 | - |> Enum.find(fn topology_response -> elem(topology_response, 0) == :ok end) |
| 110 | - |
| 111 | - case topology do |
| 99 | + case RequestExecutor.execute_with_node(%GetTopology{database_name: database}, node_pid) do |
| 112 100 | {:ok, response} -> |
| 113 101 | topology = %Topology{ |
| 114 102 | etag: response.data["Etag"], |
| @@ -136,19 +124,17 @@ defmodule Ravix.Connection.State.Manager do | |
| 136 124 | @spec connection_id(atom()) :: {:via, Registry, {:connections, atom()}} |
| 137 125 | def connection_id(state), do: {:via, Registry, {:connections, state}} |
| 138 126 | |
| 139 | - defp register_nodes(%ConnectionState{} = state) do |
| 140 | - registered_nodes = |
| 127 | + defp register_node_pools(%ConnectionState{} = state) do |
| 128 | + registered_node_pools = |
| 141 129 | state.urls |
| 142 | - |> Enum.map(fn url -> ServerNode.from_url(url, state) end) |
| 143 | - |> Enum.map(fn node -> |
| 144 | - RequestExecutor.Supervisor.register_node_executor(state.store, node) |
| 145 | - end) |
| 146 | - |> Enum.filter(fn pids -> elem(pids, 0) == :ok end) |
| 130 | + |> Enum.map(&ServerNode.from_url(&1, state)) |
| 131 | + |> Enum.map(&RequestExecutor.Supervisor.register_node_pool(state.store, &1)) |
| 132 | + |> Enum.filter(fn pid -> elem(pid, 0) == :ok end) |
| 147 133 | |> Enum.map(fn pid -> elem(pid, 1) end) |
| 148 134 | |
| 149 | - case registered_nodes do |
| 150 | - pids when is_nil(pids) or pids == [] -> {:error, :no_node_registered} |
| 151 | - pids -> {:ok, pids} |
| 135 | + case registered_node_pools do |
| 136 | + nodes when is_nil(nodes) or nodes == [] -> {:error, :no_node_registered} |
| 137 | + nodes -> {:ok, nodes} |
| 152 138 | end |
| 153 139 | end |
| 154 140 | end |
| @@ -1,55 +1,13 @@ | |
| 1 1 | defmodule Ravix.Connection.RequestExecutor do |
| 2 2 | @moduledoc false |
| 3 | - use GenServer |
| 4 3 | use Retry |
| 5 4 | |
| 6 5 | require OK |
| 7 6 | require Logger |
| 8 7 | |
| 9 | - @default_headers [{"content-type", "application/json"}, {"accept", "application/json"}] |
| 10 | - |
| 11 | - alias Ravix.Connection |
| 12 8 | alias Ravix.Connection.State, as: ConnectionState |
| 13 9 | alias Ravix.Connection.{ServerNode, NodeSelector, Response} |
| 14 10 | alias Ravix.Connection.RequestExecutor |
| 15 | - alias Ravix.Documents.Protocols.CreateRequest |
| 16 | - |
| 17 | - @doc """ |
| 18 | - Initializes the connection for the informed ServerNode |
| 19 | - |
| 20 | - The process will take care of the connection state, if the connection closes, the process |
| 21 | - will die automatically |
| 22 | - """ |
| 23 | - @spec init(ServerNode.t()) :: |
| 24 | - {:ok, ServerNode.t()} |
| 25 | - | {:stop, |
| 26 | - %{ |
| 27 | - :__exception__ => any, |
| 28 | - :__struct__ => Mint.HTTPError | Mint.TransportError, |
| 29 | - :reason => any, |
| 30 | - optional(:module) => any |
| 31 | - }} |
| 32 | - def init(%ServerNode{} = node) do |
| 33 | - Logger.info( |
| 34 | - "[RAVIX] Connecting to node '#{inspect(node.url)}' for store '#{inspect(node.store)}'" |
| 35 | - ) |
| 36 | - |
| 37 | - Process.send_after(self(), :healthcheck, :timer.seconds(node.healthcheck_every)) |
| 38 | - |
| 39 | - case connect(node) do |
| 40 | - {:ok, node} -> {:ok, node, {:continue, :healthcheck}} |
| 41 | - {:error, node} -> {:stop, node} |
| 42 | - end |
| 43 | - end |
| 44 | - |
| 45 | - @spec start_link(any, ServerNode.t()) :: :ignore | {:error, any} | {:ok, pid} |
| 46 | - def start_link(_attrs, %ServerNode{} = node) do |
| 47 | - GenServer.start_link( |
| 48 | - __MODULE__, |
| 49 | - node, |
| 50 | - name: executor_id(node.url, node.database) |
| 51 | - ) |
| 52 | - end |
| 53 11 | |
| 54 12 | @doc """ |
| 55 13 | Executes a RavenDB command for the informed connection |
| @@ -65,14 +23,15 @@ defmodule Ravix.Connection.RequestExecutor do | |
| 65 23 | - `{:ok, Ravix.Connection.Response}` for a successful call |
| 66 24 | - `{:error, cause}` if the request fails |
| 67 25 | """ |
| 68 | - @spec execute(map, ConnectionState.t(), any, keyword) :: {:error, any} | {:ok, Response.t()} |
| 26 | + @spec execute(struct(), ConnectionState.t(), tuple(), keyword()) :: |
| 27 | + {:ok, Response.t()} | {:error, any} |
| 69 28 | def execute( |
| 70 29 | command, |
| 71 30 | %ConnectionState{} = conn_state, |
| 72 31 | headers \\ {}, |
| 73 32 | opts \\ [] |
| 74 33 | ) do |
| 75 | - node_pid = NodeSelector.current_node(conn_state) |
| 34 | + pool_name = NodeSelector.current_node(conn_state) |
| 76 35 | opts = opts ++ RequestExecutor.Options.from_connection_state(conn_state) |
| 77 36 | |
| 78 37 | headers = |
| @@ -81,7 +40,7 @@ defmodule Ravix.Connection.RequestExecutor do | |
| 81 40 | false -> [{"Topology-Etag", conn_state.topology_etag}] |
| 82 41 | end |
| 83 42 | |
| 84 | - execute_for_node(command, node_pid, nil, headers, opts) |
| 43 | + execute_with_node_pool(command, pool_name, headers, opts) |
| 85 44 | end |
| 86 45 | |
| 87 46 | @doc """ |
| @@ -90,7 +49,7 @@ defmodule Ravix.Connection.RequestExecutor do | |
| 90 49 | ## Parameters |
| 91 50 | |
| 92 51 | - command: The command that will be executed, must be a RavenCommand |
| 93 | - - pid_or_url: The PID or the Url of the node where the command will be executed |
| 52 | + - url: The Url of the node where the command will be executed |
| 94 53 | - database: The database name |
| 95 54 | - headers: HTTP headers to send to RavenDB |
| 96 55 | - opts: Request options |
| @@ -99,20 +58,25 @@ defmodule Ravix.Connection.RequestExecutor do | |
| 99 58 | - `{:ok, Ravix.Connection.Response}` for a successful call |
| 100 59 | - `{:error, cause}` if the request fails |
| 101 60 | """ |
| 102 | - @spec execute_for_node(map(), binary | pid, String.t() | nil, any, keyword) :: |
| 103 | - {:error, any} | {:ok, Response.t()} |
| 104 | - def execute_for_node(command, pid_or_url, database, headers \\ {}, opts \\ []) |
| 105 | - |
| 106 | - def execute_for_node(command, pid_or_url, _database, headers, opts) when is_pid(pid_or_url) do |
| 107 | - call_raven(pid_or_url, command, headers, opts) |
| 61 | + @spec execute_with_node(struct(), pid(), tuple(), keyword()) :: |
| 62 | + {:ok, Response.t()} | {:error, any()} |
| 63 | + def execute_with_node(command, pid, headers \\ {}, opts \\ []) do |
| 64 | + call_raven(pid, command, headers, opts) |
| 108 65 | end |
| 109 66 | |
| 110 | - def execute_for_node(command, pid_or_url, database, headers, opts) |
| 111 | - when is_bitstring(pid_or_url) do |
| 112 | - call_raven(executor_id(database, pid_or_url), command, headers, opts) |
| 67 | + @spec execute_with_node_pool(struct(), binary(), tuple(), keyword()) :: |
| 68 | + {:ok, Response.t()} | {:error, any()} |
| 69 | + def execute_with_node_pool(command, pool_name, headers, opts) do |
| 70 | + :poolboy.transaction( |
| 71 | + pool_id(pool_name), |
| 72 | + fn pid -> |
| 73 | + call_raven(pid, command, headers, opts) |
| 74 | + end |
| 75 | + ) |
| 113 76 | end |
| 114 77 | |
| 115 | - defp call_raven(executor, command, headers, opts) do |
| 78 | + @spec call_raven(pid(), struct(), tuple(), keyword) :: {:ok, Response.t()} | {:error, any()} |
| 79 | + defp call_raven(pid, command, headers, opts) do |
| 116 80 | should_retry = Keyword.get(opts, :retry_on_failure, false) |
| 117 81 | retry_backoff = Keyword.get(opts, :retry_backoff, 100) |
| 118 82 | timeout = Keyword.get(opts, :timeout, 15000) |
| @@ -124,7 +88,7 @@ defmodule Ravix.Connection.RequestExecutor do | |
| 124 88 | end |
| 125 89 | |
| 126 90 | retry with: constant_backoff(retry_backoff) |> Stream.take(retry_count) do |
| 127 | - GenServer.call(executor, {:request, command, headers, opts}, timeout) |
| 91 | + GenServer.call(pid, {:request, command, headers, opts}, timeout) |
| 128 92 | after |
| 129 93 | {:ok, result} -> {:ok, result} |
| 130 94 | {:non_retryable_error, response} -> {:error, response} |
| @@ -134,22 +98,6 @@ defmodule Ravix.Connection.RequestExecutor do | |
| 134 98 | end |
| 135 99 | end |
| 136 100 | |
| 137 | - @doc """ |
| 138 | - Asynchronously updates the cluster tag for the current node |
| 139 | - |
| 140 | - ## Parameters |
| 141 | - - url: Node url |
| 142 | - - database: Database name |
| 143 | - - cluster_tag: new cluster tag |
| 144 | - |
| 145 | - ## Returns |
| 146 | - - :ok |
| 147 | - """ |
| 148 | - @spec update_cluster_tag(String.t(), String.t(), String.t()) :: :ok |
| 149 | - def update_cluster_tag(url, database, cluster_tag) do |
| 150 | - GenServer.cast(executor_id(url, database), {:update_cluster_tag, cluster_tag}) |
| 151 | - end |
| 152 | - |
| 153 101 | @doc """ |
| 154 102 | Fetches the current node executor state |
| 155 103 | |
| @@ -169,276 +117,37 @@ defmodule Ravix.Connection.RequestExecutor do | |
| 169 117 | end |
| 170 118 | end |
| 171 119 | |
| 172 | - @doc """ |
| 173 | - Fetches the current node executor state |
| 120 | + def start_link(_attrs, %ServerNode{} = node) do |
| 121 | + poolboy_config = [ |
| 122 | + {:name, pool_registry(node)}, |
| 123 | + {:worker_module, Ravix.Connection.RequestExecutor.Worker}, |
| 124 | + {:size, node.min_pool_size}, |
| 125 | + {:max_overflow, node.max_pool_size} |
| 126 | + ] |
| 174 127 | |
| 175 | - ## Parameters |
| 176 | - url = The node url |
| 177 | - database = the node database name |
| 128 | + executor_config = |
| 129 | + Keyword.new(Map.from_struct(node), fn {k, v} -> |
| 130 | + {k, v} |
| 131 | + end) |
| 178 132 | |
| 179 | - ## Returns |
| 180 | - - `{:ok, Ravix.Connection.ServerNode}` if there's a node |
| 181 | - - `{:error, :node_not_found}` if there's not a node with the informed pid |
| 182 | - """ |
| 183 | - @spec fetch_node_state(binary, binary) :: {:ok, ServerNode.t()} | {:error, :node_not_found} |
| 184 | - def fetch_node_state(url, database) when is_bitstring(url) do |
| 185 | - try do |
| 186 | - {:ok, executor_id(url, database) |> :sys.get_state()} |
| 187 | - catch |
| 188 | - :exit, _ -> {:error, :node_not_found} |
| 189 | - end |
| 190 | - end |
| 191 | - |
| 192 | - defp executor_id(url, database), |
| 193 | - do: {:via, Registry, {:request_executors, url <> "/" <> database}} |
| 194 | - |
| 195 | - #################### |
| 196 | - # Handlers # |
| 197 | - #################### |
| 198 | - def handle_call({:request, command, headers, opts}, from, %ServerNode{} = node) do |
| 199 | - request = CreateRequest.create_request(command, node) |
| 200 | - |
| 201 | - case maximum_url_length_reached?(opts, request.url) do |
| 202 | - true -> {:reply, {:error, :maximum_url_length_reached}, node} |
| 203 | - false -> exec_request(node, from, request, command, headers, opts) |
| 204 | - end |
| 205 | - end |
| 206 | - |
| 207 | - def handle_continue(:healthcheck, node) do |
| 208 | - {:noreply, healthcheck(node)} |
| 209 | - end |
| 210 | - |
| 211 | - def handle_info(:healthcheck, node) do |
| 212 | - {:noreply, healthcheck(node)} |
| 213 | - end |
| 214 | - |
| 215 | - def handle_info(message, %ServerNode{} = node) do |
| 216 | - case Mint.HTTP.stream(node.conn, message) do |
| 217 | - :unknown -> |
| 218 | - _ = Logger.error(fn -> "Received unknown message: " <> inspect(message) end) |
| 219 | - {:noreply, node} |
| 220 | - |
| 221 | - {:ok, conn, responses} -> |
| 222 | - node = put_in(node.conn, conn) |
| 223 | - state = Enum.reduce(responses, node, &process_response/2) |
| 224 | - {:noreply, state} |
| 225 | - |
| 226 | - {:error, conn, error, _headers} when is_struct(error, Mint.HTTPError) -> |
| 227 | - node = put_in(node.conn, conn) |
| 228 | - |
| 229 | - _ = |
| 230 | - Logger.error(fn -> |
| 231 | - "Received error message: " <> inspect(message) <> " " <> inspect(error) |
| 232 | - end) |
| 233 | - |
| 234 | - {:noreply, node} |
| 235 | - |
| 236 | - {:error, _conn, error, _headers} when is_struct(error, Mint.TransportError) -> |
| 237 | - Logger.debug( |
| 238 | - "The connection with the node #{inspect(node.url)} for the store #{inspect(node.store)} timed out gracefully" |
| 239 | - ) |
| 240 | - |
| 241 | - {:stop, :normal, node} |
| 242 | - end |
| 243 | - end |
| 244 | - |
| 245 | - def handle_cast({:update_cluster_tag, cluster_tag}, %ServerNode{} = node) do |
| 246 | - {:noreply, %ServerNode{node | cluster_tag: cluster_tag}} |
| 247 | - end |
| 248 | - |
| 249 | - defp exec_request( |
| 250 | - %ServerNode{conn: %{state: :closed}} = node, |
| 251 | - from, |
| 252 | - request, |
| 253 | - command, |
| 254 | - headers, |
| 255 | - opts |
| 256 | - ) do |
| 257 | - case connect(node) do |
| 258 | - {:ok, node} -> exec_request(node, from, request, command, headers, opts) |
| 259 | - _ -> {:error, :node_unreachable} |
| 260 | - end |
| 261 | - end |
| 262 | - |
| 263 | - defp exec_request(%ServerNode{} = node, from, request, command, headers, opts) do |
| 264 | - case Mint.HTTP.request( |
| 265 | - node.conn, |
| 266 | - request.method, |
| 267 | - request.url, |
| 268 | - @default_headers ++ [headers], |
| 269 | - request.data |
| 270 | - ) do |
| 271 | - {:ok, conn, request_ref} -> |
| 272 | - Logger.debug( |
| 273 | - "[RAVIX] Executing command #{inspect(command)} under the request '#{inspect(request_ref)}' for the store #{inspect(node.store)}" |
| 274 | - ) |
| 275 | - |
| 276 | - node = put_in(node.conn, conn) |
| 277 | - node = put_in(node.requests[request_ref], %{from: from, response: %{}}) |
| 278 | - node = put_in(node.opts, opts) |
| 279 | - {:noreply, node} |
| 280 | - |
| 281 | - {:error, conn, reason} -> |
| 282 | - state = put_in(node.conn, conn) |
| 283 | - {:reply, {:error, reason}, state} |
| 284 | - end |
| 285 | - end |
| 286 | - |
| 287 | - defp process_response({:status, request_ref, status}, state) do |
| 288 | - put_in(state.requests[request_ref].response[:status], status) |
| 289 | - end |
| 290 | - |
| 291 | - defp process_response({:headers, request_ref, headers}, state) do |
| 292 | - put_in(state.requests[request_ref].response[:headers], headers) |
| 293 | - end |
| 294 | - |
| 295 | - defp process_response({:data, request_ref, new_data}, state) do |
| 296 | - update_in(state.requests[request_ref].response[:data], fn data -> (data || "") <> new_data end) |
| 297 | - end |
| 298 | - |
| 299 | - defp process_response({:done, request_ref}, state) do |
| 300 | - {%{response: response, from: from}, state} = pop_in(state.requests[request_ref]) |
| 301 | - response = put_in(response[:data], decode_body(response[:data])) |
| 302 | - |
| 303 | - parsed_response = |
| 304 | - case response do |
| 305 | - %{status: 404} -> |
| 306 | - {:non_retryable_error, :document_not_found} |
| 307 | - |
| 308 | - %{status: 403} -> |
| 309 | - {:non_retryable_error, :unauthorized} |
| 310 | - |
| 311 | - %{status: 409} -> |
| 312 | - {:error, :conflict} |
| 313 | - |
| 314 | - %{status: 410} -> |
| 315 | - {:error, :node_gone} |
| 316 | - |
| 317 | - %{data: :invalid_response_payload} -> |
| 318 | - {:non_retryable_error, "Unable to parse the response payload"} |
| 319 | - |
| 320 | - %{data: data} when is_map_key(data, "Error") -> |
| 321 | - {:non_retryable_error, data["Message"]} |
| 322 | - |
| 323 | - %{data: %{"IsStale" => true}} -> |
| 324 | - Logger.warn("[RAVIX] The request '#{inspect(request_ref)}' is Stale!") |
| 325 | - |
| 326 | - case ServerNode.retry_on_stale?(state) do |
| 327 | - true -> {:error, :stale} |
| 328 | - false -> {:non_retryable_error, :stale} |
| 329 | - end |
| 330 | - |
| 331 | - error_response when error_response.status in [408, 502, 503, 504] -> |
| 332 | - parse_error(error_response) |
| 333 | - |
| 334 | - parsed_response -> |
| 335 | - {:ok, parsed_response} |
| 336 | - end |
| 337 | - |> check_if_needs_topology_update(state) |
| 338 | - |
| 339 | - Logger.debug( |
| 340 | - "[RAVIX] Request #{inspect(request_ref)} finished with response #{inspect(parsed_response)}" |
| 133 | + :poolboy.start_link( |
| 134 | + poolboy_config, |
| 135 | + executor_config |
| 341 136 | ) |
| 342 | - |
| 343 | - GenServer.reply(from, parsed_response) |
| 344 | - |
| 345 | - state |
| 346 137 | end |
| 347 138 | |
| 348 | - defp check_if_needs_topology_update({:ok, response}, %ServerNode{} = node) do |
| 349 | - case Enum.find(response.headers, fn header -> elem(header, 0) == "Refresh-Topology" end) do |
| 350 | - nil -> |
| 351 | - {:ok, response} |
| 352 | - |
| 353 | - _ -> |
| 354 | - Logger.debug( |
| 355 | - "[RAVIX] The database requested a topology refresh for the store '#{inspect(node.store)}'" |
| 356 | - ) |
| 357 | - |
| 358 | - Connection.update_topology(node.store) |
| 359 | - {:ok, response} |
| 360 | - end |
| 139 | + def child_spec(%ServerNode{} = node) do |
| 140 | + %{ |
| 141 | + id: String.to_atom(node.url <> "_" <> node.database), |
| 142 | + start: {__MODULE__, :start_link, [node]} |
| 143 | + } |
| 361 144 | end |
| 362 145 | |
| 363 | - defp check_if_needs_topology_update({error_kind, err}, _), do: {error_kind, err} |
| 364 | - |
| 365 | - defp parse_error(error_response) do |
| 366 | - case Enum.find(error_response.headers, fn header -> elem(header, 0) == "Database-Missing" end) do |
| 367 | - nil -> |
| 368 | - {:retryable_error, error_response.data["Message"]} |
| 369 | - |
| 370 | - _ -> |
| 371 | - {:error, error_response.data["Message"]} |
| 372 | - end |
| 146 | + defp pool_registry(%ServerNode{} = node) do |
| 147 | + {:via, Registry, {:request_executor_pools, NodeSelector.node_id(node), node}} |
| 373 148 | end |
| 374 149 | |
| 375 | - defp build_params(_, :http), do: {:ok, []} |
| 376 | - |
| 377 | - defp build_params(ssl_config, :https) do |
| 378 | - case ssl_config do |
| 379 | - [] -> |
| 380 | - {:error, :no_ssl_configurations_informed} |
| 381 | - |
| 382 | - transport_ops -> |
| 383 | - {:ok, transport_opts: transport_ops} |
| 384 | - end |
| 385 | - end |
| 386 | - |
| 387 | - defp decode_body(raw_response) when is_binary(raw_response) do |
| 388 | - case Jason.decode(raw_response) do |
| 389 | - {:ok, parsed_response} -> parsed_response |
| 390 | - {:error, %Jason.DecodeError{}} -> :invalid_response_payload |
| 391 | - end |
| 392 | - end |
| 393 | - |
| 394 | - defp decode_body(_), do: nil |
| 395 | - |
| 396 | - @spec maximum_url_length_reached?(keyword(), String.t()) :: boolean() |
| 397 | - defp maximum_url_length_reached?(opts, url) do |
| 398 | - max_url_length = Keyword.get(opts, :max_length_of_query_using_get_url, 1024 + 512) |
| 399 | - |
| 400 | - String.length(url) > max_url_length |
| 401 | - end |
| 402 | - |
| 403 | - defp connect(node) do |
| 404 | - {:ok, conn_params} = build_params(node.ssl_config, node.protocol) |
| 405 | - |
| 406 | - case Mint.HTTP.connect(node.protocol, node.url, node.port, conn_params) do |
| 407 | - {:ok, conn} -> |
| 408 | - Logger.debug( |
| 409 | - "[RAVIX] Connected to node '#{inspect(node.url)}' for store '#{inspect(node.store)}'" |
| 410 | - ) |
| 411 | - |
| 412 | - {:ok, %ServerNode{node | conn: conn}} |
| 413 | - |
| 414 | - {:error, reason} -> |
| 415 | - Logger.error( |
| 416 | - "[RAVIX] Unable to connect to the node '#{inspect(node.url)} for store '#{inspect(node.store)}', cause: #{inspect(reason)}'" |
| 417 | - ) |
| 418 | - |
| 419 | - {:error, reason} |
| 420 | - end |
| 421 | - end |
| 422 | - |
| 423 | - defp healthcheck(%ServerNode{conn: %{state: :closed}} = node) do |
| 424 | - node = |
| 425 | - case connect(node) do |
| 426 | - {:ok, node} -> |
| 427 | - %ServerNode{node | state: :healthy} |
| 428 | - |
| 429 | - _ -> |
| 430 | - Logger.warn("The node '${node.url}' is now in an unhealthy state!") |
| 431 | - %ServerNode{node | state: :unhealthy} |
| 432 | - end |
| 433 | - |
| 434 | - Process.send_after(self(), :healthcheck, :timer.seconds(node.healthcheck_every)) |
| 435 | - |
| 436 | - node |
| 437 | - end |
| 438 | - |
| 439 | - defp healthcheck(node) do |
| 440 | - Process.send_after(self(), :healthcheck, :timer.seconds(node.healthcheck_every)) |
| 441 | - |
| 442 | - %ServerNode{node | state: :healthy} |
| 150 | + defp pool_id(pool_id) do |
| 151 | + {:via, Registry, {:request_executor_pools, pool_id}} |
| 443 152 | end |
| 444 153 | end |
| @@ -0,0 +1,267 @@ | |
| 1 | + defmodule Ravix.Connection.RequestExecutor.Worker do |
| 2 | + @moduledoc false |
| 3 | + use GenServer |
| 4 | + use Retry |
| 5 | + |
| 6 | + require OK |
| 7 | + require Logger |
| 8 | + |
| 9 | + @default_headers [{"content-type", "application/json"}, {"accept", "application/json"}] |
| 10 | + |
| 11 | + alias Ravix.Connection |
| 12 | + alias Ravix.Connection.{ServerNode} |
| 13 | + alias Ravix.Documents.Protocols.CreateRequest |
| 14 | + |
| 15 | + @doc """ |
| 16 | + Initializes the connection for the informed ServerNode |
| 17 | + |
| 18 | + The process will take care of the connection state, if the connection closes, the process |
| 19 | + will die automatically |
| 20 | + """ |
| 21 | + @spec init(Ravix.Connection.ServerNode.t()) :: |
| 22 | + {:ok, Ravix.Connection.ServerNode.t()} |
| 23 | + | {:stop, |
| 24 | + %{ |
| 25 | + :__exception__ => any, |
| 26 | + :__struct__ => Mint.HTTPError | Mint.TransportError, |
| 27 | + :reason => any, |
| 28 | + optional(:module) => any |
| 29 | + }} |
| 30 | + def init(%ServerNode{} = node) do |
| 31 | + Logger.info( |
| 32 | + "[RAVIX] Connecting to node '#{inspect(node.url)}:#{inspect(node.port)}' for store '#{inspect(node.store)}' PID: #{inspect(self())}" |
| 33 | + ) |
| 34 | + |
| 35 | + Registry.register(:request_executors, node.store, []) |
| 36 | + |
| 37 | + case connect(node) do |
| 38 | + {:ok, node} -> {:ok, node} |
| 39 | + {:error, node} -> {:stop, node} |
| 40 | + end |
| 41 | + end |
| 42 | + |
| 43 | + @spec start_link(keyword()) :: :ignore | {:error, any} | {:ok, pid} |
| 44 | + def start_link(opts) do |
| 45 | + GenServer.start_link(__MODULE__, struct!(ServerNode, opts)) |
| 46 | + end |
| 47 | + |
| 48 | + #################### |
| 49 | + # Handlers # |
| 50 | + #################### |
| 51 | + def handle_call({:request, command, headers, opts}, from, %ServerNode{} = node) do |
| 52 | + request = CreateRequest.create_request(command, node) |
| 53 | + |
| 54 | + case maximum_url_length_reached?(opts, request.url) do |
| 55 | + true -> {:reply, {:error, :maximum_url_length_reached}, node} |
| 56 | + false -> exec_request(node, from, request, command, headers, opts) |
| 57 | + end |
| 58 | + end |
| 59 | + |
| 60 | + def handle_info(message, %ServerNode{} = node) do |
| 61 | + case Mint.HTTP.stream(node.conn, message) do |
| 62 | + :unknown -> |
| 63 | + _ = Logger.error(fn -> "[RAVIX] Received unknown message: " <> inspect(message) end) |
| 64 | + {:noreply, node} |
| 65 | + |
| 66 | + {:ok, conn, responses} -> |
| 67 | + node = put_in(node.conn, conn) |
| 68 | + state = Enum.reduce(responses, node, &process_response/2) |
| 69 | + {:noreply, state} |
| 70 | + |
| 71 | + {:error, conn, error, _headers} when is_struct(error, Mint.HTTPError) -> |
| 72 | + node = put_in(node.conn, conn) |
| 73 | + |
| 74 | + _ = |
| 75 | + Logger.error(fn -> |
| 76 | + "[RAVIX] Received error message: " <> inspect(message) <> " " <> inspect(error) |
| 77 | + end) |
| 78 | + |
| 79 | + {:noreply, node} |
| 80 | + |
| 81 | + {:error, _conn, error, _headers} when is_struct(error, Mint.TransportError) -> |
| 82 | + Logger.debug( |
| 83 | + "[RAVIX] The connection with the node '#{inspect(node.url)}:#{inspect(node.port)}' for the store '#{inspect(node.store)}' timed out gracefully" |
| 84 | + ) |
| 85 | + |
| 86 | + {:stop, :normal, node} |
| 87 | + end |
| 88 | + end |
| 89 | + |
| 90 | + def handle_cast({:update_cluster_tag, cluster_tag}, %ServerNode{} = node) do |
| 91 | + {:noreply, %ServerNode{node | cluster_tag: cluster_tag}} |
| 92 | + end |
| 93 | + |
| 94 | + defp exec_request( |
| 95 | + %ServerNode{conn: %{state: :closed}} = node, |
| 96 | + from, |
| 97 | + request, |
| 98 | + command, |
| 99 | + headers, |
| 100 | + opts |
| 101 | + ) do |
| 102 | + case connect(node) do |
| 103 | + {:ok, node} -> exec_request(node, from, request, command, headers, opts) |
| 104 | + _ -> {:error, :node_unreachable} |
| 105 | + end |
| 106 | + end |
| 107 | + |
| 108 | + defp exec_request(%ServerNode{} = node, from, request, command, headers, opts) do |
| 109 | + case Mint.HTTP.request( |
| 110 | + node.conn, |
| 111 | + request.method, |
| 112 | + request.url, |
| 113 | + @default_headers ++ [headers], |
| 114 | + request.data |
| 115 | + ) do |
| 116 | + {:ok, conn, request_ref} -> |
| 117 | + Logger.debug( |
| 118 | + "[RAVIX] Executing command #{inspect(command)} under the request '#{inspect(request_ref)}' for the store #{inspect(node.store)}" |
| 119 | + ) |
| 120 | + |
| 121 | + node = put_in(node.conn, conn) |
| 122 | + node = put_in(node.requests[request_ref], %{from: from, response: %{}}) |
| 123 | + node = put_in(node.opts, opts) |
| 124 | + {:noreply, node} |
| 125 | + |
| 126 | + {:error, conn, reason} -> |
| 127 | + state = put_in(node.conn, conn) |
| 128 | + {:reply, {:error, reason}, state} |
| 129 | + end |
| 130 | + end |
| 131 | + |
| 132 | + defp process_response({:status, request_ref, status}, state) do |
| 133 | + put_in(state.requests[request_ref].response[:status], status) |
| 134 | + end |
| 135 | + |
| 136 | + defp process_response({:headers, request_ref, headers}, state) do |
| 137 | + put_in(state.requests[request_ref].response[:headers], headers) |
| 138 | + end |
| 139 | + |
| 140 | + defp process_response({:data, request_ref, new_data}, state) do |
| 141 | + update_in(state.requests[request_ref].response[:data], fn data -> (data || "") <> new_data end) |
| 142 | + end |
| 143 | + |
| 144 | + defp process_response({:done, request_ref}, state) do |
| 145 | + {%{response: response, from: from}, state} = pop_in(state.requests[request_ref]) |
| 146 | + response = put_in(response[:data], decode_body(response[:data])) |
| 147 | + |
| 148 | + parsed_response = |
| 149 | + case response do |
| 150 | + %{status: 404} -> |
| 151 | + {:non_retryable_error, :document_not_found} |
| 152 | + |
| 153 | + %{status: 403} -> |
| 154 | + {:non_retryable_error, :unauthorized} |
| 155 | + |
| 156 | + %{status: 409} -> |
| 157 | + {:error, :conflict} |
| 158 | + |
| 159 | + %{status: 410} -> |
| 160 | + {:error, :node_gone} |
| 161 | + |
| 162 | + %{data: :invalid_response_payload} -> |
| 163 | + {:non_retryable_error, "Unable to parse the response payload"} |
| 164 | + |
| 165 | + %{data: data} when is_map_key(data, "Error") -> |
| 166 | + {:non_retryable_error, data["Message"]} |
| 167 | + |
| 168 | + %{data: %{"IsStale" => true}} -> |
| 169 | + Logger.warn("[RAVIX] The request '#{inspect(request_ref)}' is Stale!") |
| 170 | + |
| 171 | + case ServerNode.retry_on_stale?(state) do |
| 172 | + true -> {:error, :stale} |
| 173 | + false -> {:non_retryable_error, :stale} |
| 174 | + end |
| 175 | + |
| 176 | + error_response when error_response.status in [408, 502, 503, 504] -> |
| 177 | + parse_error(error_response) |
| 178 | + |
| 179 | + parsed_response -> |
| 180 | + {:ok, parsed_response} |
| 181 | + end |
| 182 | + |> check_if_needs_topology_update(state) |
| 183 | + |
| 184 | + Logger.debug( |
| 185 | + "[RAVIX] Request #{inspect(request_ref)} finished with response #{inspect(parsed_response)}" |
| 186 | + ) |
| 187 | + |
| 188 | + GenServer.reply(from, parsed_response) |
| 189 | + |
| 190 | + state |
| 191 | + end |
| 192 | + |
| 193 | + defp check_if_needs_topology_update({:ok, response}, %ServerNode{} = node) do |
| 194 | + case Enum.find(response.headers, fn header -> elem(header, 0) == "Refresh-Topology" end) do |
| 195 | + nil -> |
| 196 | + {:ok, response} |
| 197 | + |
| 198 | + _ -> |
| 199 | + Logger.debug( |
| 200 | + "[RAVIX] The database requested a topology refresh for the store '#{inspect(node.store)}'" |
| 201 | + ) |
| 202 | + |
| 203 | + Connection.update_topology(node.store) |
| 204 | + {:ok, response} |
| 205 | + end |
| 206 | + end |
| 207 | + |
| 208 | + defp check_if_needs_topology_update({error_kind, err}, _), do: {error_kind, err} |
| 209 | + |
| 210 | + defp parse_error(error_response) do |
| 211 | + case Enum.find(error_response.headers, fn header -> elem(header, 0) == "Database-Missing" end) do |
| 212 | + nil -> |
| 213 | + {:retryable_error, error_response.data["Message"]} |
| 214 | + |
| 215 | + _ -> |
| 216 | + {:error, error_response.data["Message"]} |
| 217 | + end |
| 218 | + end |
| 219 | + |
| 220 | + defp build_params(_, :http), do: {:ok, []} |
| 221 | + |
| 222 | + defp build_params(ssl_config, :https) do |
| 223 | + case ssl_config do |
| 224 | + [] -> |
| 225 | + {:error, :no_ssl_configurations_informed} |
| 226 | + |
| 227 | + transport_ops -> |
| 228 | + {:ok, transport_opts: transport_ops} |
| 229 | + end |
| 230 | + end |
| 231 | + |
| 232 | + defp decode_body(raw_response) when is_binary(raw_response) do |
| 233 | + case Jason.decode(raw_response) do |
| 234 | + {:ok, parsed_response} -> parsed_response |
| 235 | + {:error, %Jason.DecodeError{}} -> :invalid_response_payload |
| 236 | + end |
| 237 | + end |
| 238 | + |
| 239 | + defp decode_body(_), do: nil |
| 240 | + |
| 241 | + @spec maximum_url_length_reached?(keyword(), String.t()) :: boolean() |
| 242 | + defp maximum_url_length_reached?(opts, url) do |
| 243 | + max_url_length = Keyword.get(opts, :max_length_of_query_using_get_url, 1024 + 512) |
| 244 | + |
| 245 | + String.length(url) > max_url_length |
| 246 | + end |
| 247 | + |
| 248 | + defp connect(node) do |
| 249 | + {:ok, conn_params} = build_params(node.ssl_config, node.protocol) |
| 250 | + |
| 251 | + case Mint.HTTP.connect(node.protocol, node.url, node.port, conn_params) do |
| 252 | + {:ok, conn} -> |
| 253 | + Logger.debug( |
| 254 | + "[RAVIX] Connected to node '#{inspect(node.url)}:#{inspect(node.port)}' for store '#{inspect(node.store)}'" |
| 255 | + ) |
| 256 | + |
| 257 | + {:ok, %ServerNode{node | conn: conn}} |
| 258 | + |
| 259 | + {:error, reason} -> |
| 260 | + Logger.error( |
| 261 | + "[RAVIX] Unable to connect to the node '#{inspect(node.url)}:#{inspect(node.port)}' for store '#{inspect(node.store)}', cause: #{inspect(reason)}" |
| 262 | + ) |
| 263 | + |
| 264 | + {:error, reason} |
| 265 | + end |
| 266 | + end |
| 267 | + end |
Loading more files…