Merge pull request #140 from norpan/binary-datetime-format

time_format and binary_format options

Author: hamiltop

lib/rethinkdb/connection.ex

@@ -96,6 +96,8 @@ defmodule RethinkDB.Connection do
   * `durability` - possible values are 'hard' and 'soft'. In soft durability mode RethinkDB will acknowledge the write immediately after receiving it, but before the write has been committed to disk.
   * `noreply` - set to true to not receive the result object or cursor and return immediately.
   * `profile` - whether or not to return a profile of the query’s execution (default: false).
+  * `time_format` - what format to return times in (default: :native). Set this to :raw if you want times returned as JSON objects for exporting.
+  * `binary_format` - what format to return binary data in (default: :native). Set this to :raw if you want the raw pseudotype.
   """
   def run(query, conn, opts \\ []) do
     timeout = Dict.get(opts, :timeout, 5000)
@@ -110,7 +112,7 @@ defmodule RethinkDB.Connection do
       false -> {:query, query}
     end
     case Connection.call(conn, msg, timeout) do
-      {response, token} -> RethinkDB.Response.parse(response, token, conn)
+      {response, token} -> RethinkDB.Response.parse(response, token, conn, opts)
       :noreply -> :ok
       result -> result
     end
@@ -122,9 +124,9 @@ defmodule RethinkDB.Connection do
   Since a feed is tied to a particular connection, no connection is needed when calling
   `next`.
   """
-  def next(%{token: token, pid: pid}) do
+  def next(%{token: token, pid: pid, opts: opts}) do
     case Connection.call(pid, {:continue, token}, :infinity) do
-      {response, token} -> RethinkDB.Response.parse(response, token, pid)
+      {response, token} -> RethinkDB.Response.parse(response, token, pid, opts)
       x -> x
     end
   end
@@ -137,15 +139,15 @@ defmodule RethinkDB.Connection do
   """
   def close(%{token: token, pid: pid}) do
     {response, token} = Connection.call(pid, {:stop, token}, :infinity)
-    RethinkDB.Response.parse(response, token, pid)
+    RethinkDB.Response.parse(response, token, pid, [])
   end
 
   @doc """
   `noreply_wait` ensures that previous queries with the noreply flag have been processed by the server. Note that this guarantee only applies to queries run on the given connection.
   """
   def noreply_wait(conn, timeout \\ 5000) do
     {response, token} = Connection.call(conn, :noreply_wait, timeout)
-    case RethinkDB.Response.parse(response, token, conn) do
+    case RethinkDB.Response.parse(response, token, conn, []) do
       %RethinkDB.Response{data: %{"t" => 4}} -> :ok
       r -> r
     end

lib/rethinkdb/prepare.ex

@@ -39,6 +39,10 @@ defmodule RethinkDB.Prepare do
     {[k,v], state}
   end
   defp prepare(el, state) do
-    {el, state}
+    if is_binary(el) and not String.valid?(el) do
+      {RethinkDB.Query.binary(el), state}
+    else
+      {el, state}
+    end
   end
 end

lib/rethinkdb/pseudotypes.ex

@@ -4,8 +4,13 @@ defmodule RethinkDB.Pseudotypes do
     @moduledoc false
     defstruct data: nil
 
-    def parse(%{"$reql_type$" => "BINARY", "data" => data}) do
-      %__MODULE__{data: :base64.decode(data)}
+    def parse(%{"$reql_type$" => "BINARY", "data" => data}, opts) do
+      case Dict.get(opts, :binary_format) do
+        :raw ->
+          %__MODULE__{data: data}
+        _ ->
+          :base64.decode(data)
+      end
     end
   end
 
@@ -41,33 +46,56 @@ defmodule RethinkDB.Pseudotypes do
     @moduledoc false
     defstruct epoch_time: nil, timezone: nil
 
-    def parse(%{"$reql_type$" => "TIME", "epoch_time" => epoch_time, "timezone" => timezone}) do
-      %__MODULE__{epoch_time: epoch_time, timezone: timezone}
+    def parse(%{"$reql_type$" => "TIME", "epoch_time" => epoch_time, "timezone" => timezone}, opts) do
+      case Dict.get(opts, :time_format) do
+        :raw ->
+          %__MODULE__{epoch_time: epoch_time, timezone: timezone}
+        _ ->
+          {seconds, ""} = Calendar.ISO.parse_offset(timezone)
+          zone_abbr = case seconds do
+              0 -> "UTC"
+              _ -> timezone
+          end
+          negative = seconds < 0
+          seconds = abs(seconds)
+          time_zone = case {div(seconds,3600),rem(seconds,3600)} do
+              {0,0} -> "Etc/UTC"
+              {hours,0} ->
+                "Etc/GMT" <> if negative do "+" else "-" end <> Integer.to_string(hours)
+              {hours,seconds} ->
+                "Etc/GMT" <> if negative do "+" else "-" end <> Integer.to_string(hours) <> ":"  <>
+                String.pad_leading(Integer.to_string(seconds), 2, "0")
+          end
+          epoch_time * 1000
+            |> trunc()
+            |> DateTime.from_unix!(:milliseconds)
+            |> struct(utc_offset: seconds, zone_abbr: zone_abbr, time_zone: time_zone)
+      end
     end
   end
 
-  def convert_reql_pseudotypes(nil), do: nil
-  def convert_reql_pseudotypes(%{"$reql_type$" => "BINARY"} = data) do
-    Binary.parse(data)
+  def convert_reql_pseudotypes(nil, opts), do: nil
+  def convert_reql_pseudotypes(%{"$reql_type$" => "BINARY"} = data, opts) do
+    Binary.parse(data, opts)
   end
-  def convert_reql_pseudotypes(%{"$reql_type$" => "GEOMETRY"} = data) do
+  def convert_reql_pseudotypes(%{"$reql_type$" => "GEOMETRY"} = data, opts) do
     Geometry.parse(data)
   end
-  def convert_reql_pseudotypes(%{"$reql_type$" => "GROUPED_DATA"} = data) do
+  def convert_reql_pseudotypes(%{"$reql_type$" => "GROUPED_DATA"} = data, opts) do
     parse_grouped_data(data)
   end
-  def convert_reql_pseudotypes(%{"$reql_type$" => "TIME"} = data) do
-    Time.parse(data)
+  def convert_reql_pseudotypes(%{"$reql_type$" => "TIME"} = data, opts) do
+    Time.parse(data, opts)
   end
-  def convert_reql_pseudotypes(list) when is_list(list) do
-    Enum.map(list, &convert_reql_pseudotypes/1)
+  def convert_reql_pseudotypes(list, opts) when is_list(list) do
+    Enum.map(list, fn data -> convert_reql_pseudotypes(data, opts) end)
   end
-  def convert_reql_pseudotypes(map) when is_map(map) do
+  def convert_reql_pseudotypes(map, opts) when is_map(map) do
     Enum.map(map, fn {k, v} ->
-      {k, convert_reql_pseudotypes(v)}
+      {k, convert_reql_pseudotypes(v, opts)}
     end) |> Enum.into(%{})
   end
-  def convert_reql_pseudotypes(string), do: string
+  def convert_reql_pseudotypes(string, opts), do: string
 
   def parse_grouped_data(%{"$reql_type$" => "GROUPED_DATA", "data" => data}) do
     Enum.map(data, fn ([k, data]) ->

lib/rethinkdb/query.ex

@@ -188,6 +188,14 @@ defmodule RethinkDB.Query do
   @doc """
   Encapsulate binary data within a query.
 
+  The type of data binary accepts depends on the client language. In
+  Elixir, it expects a Binary. Using a Binary object within a query implies
+  the use of binary and the ReQL driver will automatically perform the coercion.
+
+  Binary objects returned to the client in Elixir will also be
+  Binary objects. This can be changed with the binary_format option :raw
+  to run to return “raw” objects.
+
   Only a limited subset of ReQL commands may be chained after binary:
 
   * coerce_to can coerce binary objects to string types
@@ -198,9 +206,9 @@ defmodule RethinkDB.Query do
   * info will return information on a binary object.
   """
   @spec binary(Q.reql_binary) :: Q.t
-  def binary(%RethinkDB.Pseudotypes.Binary{data: data}), do: binary(data)
-  def binary(data), do: do_binary(%{"$reql_type$" => "BINARY", "data" => :base64.encode(data)})
-  def do_binary(data), do: %Q{query: [155, [data]]}
+  def binary(%RethinkDB.Pseudotypes.Binary{data: data}), do: do_binary(data)
+  def binary(data), do: do_binary(:base64.encode(data))
+  def do_binary(data), do: %Q{query: [155, [%{"$reql_type$" => "BINARY", "data" => data}]]}
 
   @doc """
   Call an anonymous function using return values from other ReQL commands or 

lib/rethinkdb/query/macros.ex

@@ -106,6 +106,22 @@ defmodule RethinkDB.Query.Macros do
     m = Map.from_struct(t) |> Map.put_new("$reql_type$", "TIME")
     wrap(m)
   end
+  def wrap(t = %DateTime{utc_offset: utc_offset, std_offset: std_offset}) do
+    offset = utc_offset + std_offset
+    offset_negative = offset < 0
+    offset_hour = div(abs(offset), 3600)
+    offset_minute = rem(abs(offset), 3600)
+    time_zone =
+        if offset_negative do "-" else "+" end <>
+        String.pad_leading(Integer.to_string(offset_hour), 2, "0") <>
+        ":" <>
+        String.pad_leading(Integer.to_string(offset_minute), 2, "0")
+    wrap(%{
+      "$reql_type$" => "TIME",
+       "epoch_time" => DateTime.to_unix(t, :milliseconds) / 1000,
+       "timezone" => time_zone
+    })
+  end
   def wrap(map) when is_map(map) do
     Enum.map(map, fn {k,v} ->
       {k, wrap(v)}

lib/rethinkdb/response.ex

@@ -19,7 +19,7 @@ end
 
 defmodule RethinkDB.Feed do
   @moduledoc false
-  defstruct token: nil, data: nil, pid: nil, note: nil, profile: nil
+  defstruct token: nil, data: nil, pid: nil, note: nil, profile: nil, opts: nil
 
   defimpl Enumerable, for: __MODULE__ do
     def reduce(changes, acc, fun) do
@@ -45,13 +45,13 @@ defmodule RethinkDB.Response do
   @moduledoc false
   defstruct token: nil, data: "", profile: nil
 
-  def parse(raw_data, token, pid) do
+  def parse(raw_data, token, pid, opts) do
     d = Poison.decode!(raw_data)
-    data = RethinkDB.Pseudotypes.convert_reql_pseudotypes(d["r"])
+    data = RethinkDB.Pseudotypes.convert_reql_pseudotypes(d["r"], opts)
     {code, resp} = case d["t"] do
       1  -> {:ok, %RethinkDB.Record{data: hd(data)}}
       2  -> {:ok, %RethinkDB.Collection{data: data}}
-      3  -> {:ok, %RethinkDB.Feed{token: token, data: data, pid: pid, note: d["n"]}}
+      3  -> {:ok, %RethinkDB.Feed{token: token, data: data, pid: pid, note: d["n"], opts: opts}}
       4  ->  {:ok, %RethinkDB.Response{token: token, data: d}}
       16  -> {:error, %RethinkDB.Response{token: token, data: d}}
       17  -> {:error, %RethinkDB.Response{token: token, data: d}}

test/changes_test.exs

@@ -71,4 +71,32 @@ defmodule ChangesTest do
     [h|[]] = Task.await(t)
     assert %{"new_val" => %{"id" => "0"}} = h
   end
+
+  test "changes opts binary native" do
+    q = table(@table_name) |> get("0") |> changes
+    {:ok, changes} = {:ok, %Feed{}} = run(q)
+    t = Task.async fn ->
+      changes |> Enum.take(1)
+    end
+    data = %{"id" => "0", "binary" => binary(<<1>>)}
+    q = table(@table_name) |> insert(data)
+    {:ok, res} = run(q)
+    expected = res.data["id"]
+    [h|[]] = Task.await(t)
+    assert %{"new_val" => %{"id" => "0", "binary" => <<1>>}} = h
+  end
+
+  test "changes opts binary raw" do
+    q = table(@table_name) |> get("0") |> changes
+    {:ok, changes} = {:ok, %Feed{}} = run(q, [binary_format: :raw])
+    t = Task.async fn ->
+      changes |> Enum.take(1)
+    end
+    data = %{"id" => "0", "binary" => binary(<<1>>)}
+    q = table(@table_name) |> insert(data)
+    {:ok, res} = run(q)
+    expected = res.data["id"]
+    [h|[]] = Task.await(t)
+    assert %{"new_val" => %{"id" => "0", "binary" => %RethinkDB.Pseudotypes.Binary{data: "AQ=="}}} = h
+  end
 end

test/query/control_structures_test.exs

@@ -17,13 +17,33 @@ defmodule ControlStructuresTest do
     assert data == [%{"a" => 5}, %{"a" => 4, "c" => 7}]
   end
 
-  test "binary" do
+  test "binary raw" do
+    d = << 220, 2, 3, 4, 5, 192 >>
+    q = binary d
+    {:ok, %Record{data: data}} = run q, [binary_format: :raw]
+    assert data == %RethinkDB.Pseudotypes.Binary{data: :base64.encode(d)}
+    q = binary data
+    {:ok, %Record{data: result}} = run q, [binary_format: :raw]
+    assert data == result
+  end
+
+  test "binary native" do
     d = << 220, 2, 3, 4, 5, 192 >>
     q = binary d
     {:ok, %Record{data: data}} = run q
-    assert data == %RethinkDB.Pseudotypes.Binary{data: d}
+    assert data == d
     q = binary data
-    {:ok, %Record{data: result}} = run q
+    {:ok, %Record{data: result}} = run q, [binary_format: :native]
+    assert data == result
+  end
+
+  test "binary native no wrapper" do
+    d = << 220, 2, 3, 4, 5, 192 >>
+    q = d
+    {:ok, %Record{data: data}} = run q
+    assert data == d
+    q = data
+    {:ok, %Record{data: result}} = run q, [binary_format: :native]
     assert data == result
   end
 
@@ -39,10 +59,10 @@ defmodule ControlStructuresTest do
   test "branch" do
     q = branch(true, 1, 2)
     {:ok, %Record{data: data}} = run q
-    assert data == 1 
+    assert data == 1
     q = branch(false, 1, 2)
     {:ok, %Record{data: data}} = run q
-    assert data == 2 
+    assert data == 2
   end
 
   test "error" do
@@ -105,7 +125,7 @@ defmodule ControlStructuresTest do
   end
 
   test "uuid" do
-    q = uuid  
+    q = uuid
     {:ok, %Record{data: data}} = run q
     assert String.length(String.replace(data, "-", ""))  == 32
   end