Adding date_format and binary_format to run() options. Also slightly changing the binary pseudo type to better match the other drivers.

Author: norpan

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/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!(:millisecond)
+            |> 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,13 @@ 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 Buffer 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 JavaScript will also be Node.js
+  Buffer objects. This can be changed with the binaryFormat option provided
+  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 +205,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, :millisecond) / 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,23 @@ 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
 
@@ -39,10 +49,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 +115,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

test/query/date_time_test.exs

@@ -11,35 +11,64 @@ defmodule DateTimeTest do
     :ok
   end
 
-  test "now" do
+  test "now native" do
     {:ok, %Record{data: data}} = now |> run
+    assert %DateTime{} = data
+  end
+
+  test "now raw" do
+    {:ok, %Record{data: data}} = now |> run [time_format: :raw]
     assert %Time{} = data
   end
 
-  test "time" do
+  test "time native" do
     {:ok, %Record{data: data}} = time(1970,1,1,"Z") |> run
+    assert data == DateTime.from_unix!(0, :millisecond)
+    {:ok, %Record{data: data}} = time(1970,1,1,0,0,1,"Z") |> run [binary_format: :native]
+    assert data == DateTime.from_unix!(1000, :millisecond)
+  end
+
+  test "time raw" do
+    {:ok, %Record{data: data}} = time(1970,1,1,"Z") |> run [time_format: :raw]
     assert data.epoch_time == 0
-    {:ok, %Record{data: data}} = time(1970,1,1,0,0,1,"Z") |> run
+    {:ok, %Record{data: data}} = time(1970,1,1,0,0,1,"Z") |> run [time_format: :raw]
     assert data.epoch_time == 1
   end
 
-  test "epoch_time" do
+  test "epoch_time native" do
     {:ok, %Record{data: data}} = epoch_time(1) |> run
+    assert data == DateTime.from_unix!(1000, :millisecond)
+  end
+
+  test "epoch_time raw" do
+    {:ok, %Record{data: data}} = epoch_time(1) |> run [time_format: :raw]
     assert data.epoch_time == 1
     assert data.timezone == "+00:00"
   end
 
-  test "iso8601" do
+  test "iso8601 native" do
     {:ok, %Record{data: data}} = iso8601("1970-01-01T00:00:00+00:00") |> run
+    assert data == DateTime.from_unix!(0, :millisecond)
+    {:ok, %Record{data: data}} = iso8601("1970-01-01T00:00:00", default_timezone: "+01:00") |> run
+    assert data == DateTime.from_unix!(-3600000, :millisecond) |> struct(utc_offset: 3600, time_zone: "Etc/GMT-1", zone_abbr: "+01:00")
+  end
+
+  test "iso8601 raw" do
+    {:ok, %Record{data: data}} = iso8601("1970-01-01T00:00:00+00:00") |> run [time_format: :raw]
     assert data.epoch_time == 0
     assert data.timezone == "+00:00"
-    {:ok, %Record{data: data}} = iso8601("1970-01-01T00:00:00", default_timezone: "+01:00") |> run
+    {:ok, %Record{data: data}} = iso8601("1970-01-01T00:00:00", default_timezone: "+01:00") |> run [time_format: :raw]
     assert data.epoch_time == -3600
     assert data.timezone == "+01:00"
   end
 
-  test "in_timezone" do
+  test "in_timezone native" do
     {:ok, %Record{data: data}} = epoch_time(0) |> in_timezone("+01:00") |> run
+    assert data == DateTime.from_unix!(0, :millisecond) |> struct(utc_offset: 3600, time_zone: "Etc/GMT-1", zone_abbr: "+01:00")
+  end
+
+  test "in_timezone raw" do
+    {:ok, %Record{data: data}} = epoch_time(0) |> in_timezone("+01:00") |> run [time_format: :raw]
     assert data.timezone == "+01:00"
     assert data.epoch_time == 0
   end
@@ -59,8 +88,13 @@ defmodule DateTimeTest do
     assert data == false
   end
 
-  test "date" do
+  test "date native" do
     {:ok, %Record{data: data}} = epoch_time(5) |> date |> run
+    assert data == DateTime.from_unix!(0, :millisecond)
+  end
+
+  test "date raw" do
+    {:ok, %Record{data: data}} = epoch_time(5) |> date |> run [time_format: :raw]
     assert data.epoch_time == 0
   end
 
@@ -81,12 +115,12 @@ defmodule DateTimeTest do
 
   test "day" do
     {:ok, %Record{data: data}} = epoch_time(3*60*60*24) |> day |> run
-    assert data == 4 
+    assert data == 4
   end
 
   test "day_of_week" do
     {:ok, %Record{data: data}} = epoch_time(3*60*60*24) |> day_of_week |> run
-    assert data == 7 
+    assert data == 7
   end
 
   test "day_of_year" do