Merge pull request #20 from bzzt/develop

Connection Pooling and Refactor

guides/introduction/installation.md

@@ -6,7 +6,7 @@
 #mix.exs
 def deps do
   [
-    {:bigtable, "~> 0.1.0"},
+    {:bigtable, "~> 0.7.0"},
   ]
 end
 ```
@@ -18,11 +18,14 @@ end
 ```elixir
 #dev.exs
 config :bigtable,
-  project: "project_id",
-  instance: "instance_id",
-  table: "table_name",
-  endpoint: "localhost:8086"
+  project: "project",
+  instance: "instance",
+  table: "table_name", # Default table name to use in requests
+  endpoint: "localhost:9035",
   ssl: false
+
+config :goth,
+  disabled: true
 ```
 
 #### Production Configuration
@@ -32,5 +35,10 @@ config :bigtable,
 config :bigtable,
   project: "project_id",
   instance: "instance_id",
-  table: "table_name"
+  # Default table name to use in requests
+  table: "table_name",
+  # Optional connection pool size. Defaults to 128
+  pool_size: 128,
+  # Optional connection pool overflow when size is exceeded
+  pool_overflow: 128
 ```

lib/admin/table_admin/gc_rule.ex

@@ -1,4 +1,7 @@
 defmodule Bigtable.Admin.GcRule do
+  @moduledoc """
+  Provides functions for creating garbage collection rules
+  """
   alias Google.Bigtable.Admin.V2
   alias Google.Protobuf.Duration
   alias V2.GcRule.{Intersection, Union}

lib/admin/table_admin/modification.ex

@@ -1,4 +1,5 @@
 defmodule Bigtable.Admin.Modification do
-  def create(id) do
+  @moduledoc false
+  def create(_id) do
   end
 end

lib/admin/table_admin/table.ex

@@ -1,4 +1,8 @@
 defmodule Bigtable.Admin.Table do
+  @moduledoc """
+  Provides functionality for building `Google.Bigtable.Admin.V2.Table`.
+  """
+
   alias Google.Bigtable.Admin.V2
 
   def build(column_families) when is_map(column_families) do

lib/admin/table_admin/table_admin.ex

@@ -2,14 +2,15 @@ defmodule Bigtable.Admin.TableAdmin do
   @moduledoc """
   Provides functions to build `Google.Bigtable.Admin.V2.ListTablesRequest` and submit them to Bigtable.
   """
-  alias Bigtable.Utils
+  alias Bigtable.{Request, Utils}
   alias Google.Bigtable.Admin.V2
   alias V2.BigtableTableAdmin.Stub
 
   def list_tables(opts \\ []) do
-    Keyword.put_new(opts, :parent, Utils.configured_instance_name())
+    opts
+    |> Keyword.put_new(:parent, Utils.configured_instance_name())
     |> V2.ListTablesRequest.new()
-    |> Utils.process_request(&Stub.list_tables/3)
+    |> Request.process_request(&Stub.list_tables/3)
   end
 
   def create_table(table, table_id, opts \\ []) do
@@ -19,16 +20,16 @@ defmodule Bigtable.Admin.TableAdmin do
       table: table,
       initial_splits: Keyword.get(opts, :initial_splits, [])
     )
-    |> Utils.process_request(&Stub.create_table/3)
+    |> Request.process_request(&Stub.create_table/3)
   end
 
   def delete_table(name) do
     V2.DeleteTableRequest.new(name: name)
-    |> Utils.process_request(&Stub.delete_table/3)
+    |> Request.process_request(&Stub.delete_table/3)
   end
 
   def get_table(name) do
     V2.GetTableRequest.new(name: name)
-    |> Utils.process_request(&Stub.get_table/3)
+    |> Request.process_request(&Stub.get_table/3)
   end
 end

lib/connection/auth.ex → lib/auth.ex

@@ -1,4 +1,4 @@
-defmodule Bigtable.Connection.Auth do
+defmodule Bigtable.Auth do
   @moduledoc false
 
   @scopes [

lib/bigtable.ex

@@ -6,8 +6,16 @@ defmodule Bigtable do
 
   @doc false
   def start(_type, _args) do
+    poolboy_config = [
+      {:name, {:local, :connection_pool}},
+      {:worker_module, Bigtable.Connection.Worker},
+      {:size, Application.get_env(:bigtable, :pool_size, 128)},
+      {:max_overflow, Application.get_env(:bigtable, :pool_overflow, 0)}
+    ]
+
     children = [
-      Bigtable.Supervisor
+      Bigtable.Supervisor,
+      :poolboy.child_spec(:connection_pool, poolboy_config, [])
     ]
 
     opts = [strategy: :one_for_one, name: Bigtable]

lib/connection.ex

@@ -0,0 +1,77 @@
+defmodule Bigtable.Connection do
+  @moduledoc false
+
+  use GenServer
+  @default_endpoint "bigtable.googleapis.com:443"
+
+  ## Client API
+  def start_link(_opts) do
+    GenServer.start_link(__MODULE__, :ok, name: __MODULE__)
+  end
+
+  @doc """
+  Connects to Bigtable and returns a `GRPC.Channel`.
+  """
+  @spec connect() :: GRPC.Channel.t()
+  def connect do
+    GenServer.call(__MODULE__, :connect)
+  end
+
+  @doc """
+  Disconnects from the provided `GRPC.Channel`.
+  """
+  @spec disconnect(GRPC.Channel.t()) :: :ok
+  def disconnect(channel) do
+    GenServer.cast(__MODULE__, {:disconnect, channel})
+  end
+
+  # Server Callbacks
+  @spec init(:ok) :: {:ok, map()}
+  def init(:ok) do
+    {:ok, %{endpoint: get_endpoint(), opts: build_opts()}}
+  end
+
+  def handle_call(:connect, _from, %{endpoint: endpoint, opts: opts} = state) do
+    {:ok, channel} =
+      GRPC.Stub.connect(
+        endpoint,
+        opts
+      )
+
+    {:reply, channel, state}
+  end
+
+  def handle_cast({:disconnect, channel}, state) do
+    GRPC.Stub.disconnect(channel)
+    {:noreply, state}
+  end
+
+  def handle_info(_msg, state) do
+    {:noreply, state}
+  end
+
+  @spec build_opts() :: list()
+  defp build_opts do
+    if Application.get_env(:bigtable, :ssl, true) do
+      [
+        cred: %GRPC.Credential{
+          ssl: []
+        }
+      ]
+    else
+      []
+    end
+  end
+
+  @spec get_endpoint() :: binary()
+  def get_endpoint do
+    emulator = System.get_env("BIGTABLE_EMULATOR_HOST")
+    endpoint = Application.get_env(:bigtable, :endpoint, @default_endpoint)
+
+    if emulator != nil do
+      emulator
+    else
+      endpoint
+    end
+  end
+end

lib/connection/worker.ex

@@ -0,0 +1,40 @@
+defmodule Bigtable.Connection.Worker do
+  @moduledoc false
+  alias Bigtable.Connection
+  use GenServer
+
+  def start_link(_) do
+    GenServer.start_link(__MODULE__, nil, [])
+  end
+
+  def get_connection(pid) do
+    GenServer.call(pid, :get_connection)
+  end
+
+  def init(_) do
+    Process.flag(:trap_exit, true)
+    {:ok, Connection.connect()}
+  end
+
+  def handle_call(:get_connection, _from, state) do
+    {:reply, state, state}
+  end
+
+  def handle_info({:EXIT, _from, reason}, state) do
+    disconnect(state)
+    {:stop, reason, state}
+  end
+
+  def handle_info(_msg, state) do
+    {:noreply, state}
+  end
+
+  def terminate(_reason, state) do
+    disconnect(state)
+    state
+  end
+
+  defp disconnect(connection) do
+    Connection.disconnect(connection)
+  end
+end