feature/add_influx_timestamp

lib/fluxter.ex

@@ -320,19 +320,21 @@ defmodule Fluxter do
         end
       end
 
-      def write(measurement, tags \\ [], fields)
+      def write(measurement, tags \\ [], fields) do
+        write(measurement, tags, fields, nil)
+      end
 
-      def write(measurement, tags, fields) when is_list(fields) do
+      def write(measurement, tags, fields, timestamp_milli_secs) when is_list(fields) do
         System.unique_integer([:positive])
         |> rem(@pool_size)
         |> worker_name()
-        |> Fluxter.Conn.write(measurement, tags, fields)
+        |> Fluxter.Conn.write(measurement, tags, fields, timestamp_milli_secs)
       end
 
-      def write(measurement, tags, value)
+      def write(measurement, tags, value, timestamp_milli_secs)
           when is_float(value) or is_integer(value)
           when is_boolean(value) or is_binary(value) do
-        write(measurement, tags, [value: value])
+        write(measurement, tags, [value: value], timestamp_milli_secs)
       end
 
       def measure(measurement, tags \\ [], fields \\ [], fun)

lib/fluxter/conn.ex

@@ -23,11 +23,11 @@ defmodule Fluxter.Conn do
     GenServer.start_link(__MODULE__, conn, [name: worker])
   end
 
-  def write(worker, name, tags, fields)
+  def write(worker, name, tags, fields, timestamp_milli_secs)
       when (is_binary(name) or is_list(name)) and is_list(tags) and is_list(fields) do
     # TODO: Remove `try` wrapping when we depend on Elixir ~> 1.3
     try do
-      GenServer.cast(worker, {:write, name, tags, fields})
+      GenServer.cast(worker, {:write, name, tags, fields, timestamp_milli_secs})
     catch
       _, _ -> :ok
     end
@@ -38,8 +38,8 @@ defmodule Fluxter.Conn do
     {:ok, %{conn | sock: sock}}
   end
 
-  def handle_cast({:write, name, tags, fields}, conn) do
-    packet = Packet.build(conn.header, name, tags, fields)
+  def handle_cast({:write, name, tags, fields, timestamp_milli_secs}, conn) do
+    packet = Packet.build(conn.header, name, tags, fields, timestamp_milli_secs)
     send(conn.sock, {self(), {:command, packet}})
     {:noreply, conn}
   end

lib/fluxter/packet.ex

@@ -17,10 +17,21 @@ defmodule Fluxter.Packet do
     ]
   end
 
-  def build(header, name, tags, fields) do
+  def build(header, name, tags, fields, unix_timestamp_ms) do
     tags   = encode_tags(tags)
     fields = encode_fields(fields)
-    [header, encode_key(name), tags, ?\s, fields]
+
+    case is_nil(unix_timestamp_ms) do
+      true  -> [header, encode_key(name), tags, ?\s, fields]
+      false ->
+        # Convert time to nanoseconds, which is the precision influxdb uses
+        unix_timestamp_nano_secs = unix_timestamp_ms
+          |> Kernel.*(1_000_000)
+          |> Integer.to_string()
+
+        [header, encode_key(name), tags, ?\s, fields, ?\s,
+          unix_timestamp_nano_secs]
+    end
   end
 
   defp encode_tags([]), do: ""

test/fluxter_test.exs

@@ -89,6 +89,17 @@ defmodule FluxterTest do
     TestFluxter.write("foo", [bar: "baz", qux: "baz"], 0)
     assert_receive {:echo, "foo,bar=baz,qux=baz value=0i"}
 
+
+    timestamp_milli_secs = 1415521167028459
+    TestFluxter.write("foo", [bar: "baz", qux: "baz"], 0, timestamp_milli_secs)
+
+    timestamp_nanoseconds = (timestamp_milli_secs * 1_000_000)
+      |> Integer.to_string()
+
+    expected_line_msg = "foo,bar=baz,qux=baz value=0i #{timestamp_nanoseconds}"
+
+    assert_receive {:echo, ^expected_line_msg}
+
     refute_receive _any
   end