Merge pull request #127 from fishcakez/simple_notify

Use simple query to listen/unlisten

Author: ericmj

lib/postgrex/notifications.ex

@@ -90,7 +90,7 @@ defmodule Postgrex.Notifications do
   end
 
   def connect(_, opts) do
-    case Protocol.connect(opts) do
+    case Protocol.connect([types: false] ++ opts) do
       {:ok, protocol, parameters, _} ->
         {:ok, %__MODULE__{protocol: protocol, parameters: parameters}}
       {:error, reason} ->
@@ -156,7 +156,7 @@ defmodule Postgrex.Notifications do
   defp listener_query(statement, result, from, buffer, s) do
     %{protocol: protocol, parameters: parameters} = s
 
-    case Protocol.query(protocol, statement, [], buffer) do
+    case Protocol.simple_query(protocol, statement, buffer) do
       {:ok, %Postgrex.Result{}, new_parameters, notifications, buffer} ->
         if from, do: Connection.reply(from, result)
         notify_listeners(notifications, s)

lib/postgrex/protocol.ex

@@ -25,10 +25,11 @@ defmodule Postgrex.Protocol do
     custom     = opts[:extensions] || []
     extensions = custom ++ @default_extensions
     ssl?       = opts[:ssl] || false
+    types?     = opts[:types] || true
 
     s = %{sock: nil, backend_key: nil, types: nil, timeout: timeout}
 
-    types_key = {host, port, Keyword.fetch!(opts, :database), custom}
+    types_key = if types?, do: {host, port, Keyword.fetch!(opts, :database), custom}
     status = %{opts: opts, parameters: %{}, notifications: [],
                types_key: types_key, types_ref: nil, extensions: extensions,
                extension_info: nil}
@@ -54,6 +55,15 @@ defmodule Postgrex.Protocol do
     activate(sock, buffer)
   end
 
+  @spec simple_query(state, String.t, binary | :active_once) ::
+    {:ok, Postgrex.Result.t | Postgrex.Error.t, parameters, notifications,
+      binary} |
+    {:error, Postgrex.Error.t}
+  def simple_query(s, statement, buffer) do
+    status = %{parameters: %{}, notifications: [], ok: :result, sync: :sync}
+    simple_send(s, status, statement, buffer)
+  end
+
   @spec query(state, String.t, [any], binary | :active_once) ::
     {:ok, Postgrex.Result.t | Postgrex.Error.t |
       {:error | :throw | :exit, any, list}, parameters, notifications, binary} |
@@ -243,6 +253,9 @@ defmodule Postgrex.Protocol do
 
   ## bootstrap
 
+  defp bootstrap(s, %{types_key: nil} = status, buffer) do
+    bootstrap_ready(s, status, buffer)
+  end
   defp bootstrap(s, %{types_key: types_key} = status, buffer) do
     case Postgrex.TypeServer.fetch(types_key) do
       {:ok, table} ->
@@ -331,6 +344,30 @@ defmodule Postgrex.Protocol do
     end
   end
 
+  ## simple
+
+  defp simple_send(%{sock: sock} = s, status, statement, buffer) do
+    msg = msg_query(statement: statement)
+    case msg_send(msg, sock) do
+      :ok               -> simple_recv(s, status, buffer)
+      {:error, _} = err -> err
+    end
+  end
+
+  defp simple_recv(s, status, buffer) do
+    %{sock: sock, timeout: timeout} = s
+    case msg_recv(sock, buffer, timeout) do
+      {:ok, msg_command_complete(tag: tag), buffer} ->
+        complete(s, status, %Query{}, [], tag, buffer)
+      {:ok, msg_error(fields: fields), buffer} ->
+        sync_recv(s, status, Postgrex.Error.exception(postgres: fields), buffer)
+      {:ok, msg, buffer} ->
+        simple_recv(s, handle_msg(status, msg), buffer)
+      {:error, _} = err ->
+        err
+    end
+  end
+
   ## query
 
   defp query_encode(s, status, query, params, buffer) do