Bulk requests via post (#12)

* Add intial support for post

* Add value validations for structured event (#10)

* Add emitter for bulk and lone requests

* Add processor module to send events

* Fix scheduler for bulk processor

* Add cache module

* Add placeholder files for test cases

* Update travis CI version

* Refactor application.ex and rename bulk.ex

* Add tests for cache module

* Add tests

* Remove unused variables

* Add try catch

* Remove tab file when deleting ETS table

* Move delete to cache module

* Comment cache test cases

* Add cache test

* Add test cases for set_lock

* Add tests for release_lock

* Add check_lock test case

* Cleanup tests

* Add more tests

* Fix cache issue

* Add more tests

* Fix failing test cases

* Fix error

* Add alias

* Remove unused alias

Author: goodhamgupta

.gitignore

@@ -21,3 +21,6 @@ erl_crash.dump
 
 # Ignore package tarball (built via "mix hex.build").
 snowplow_elixir_tracker-*.tar
+
+# Ignore elixir tables created using PersistentEts
+*.tab

.travis.yml

@@ -6,7 +6,7 @@ otp_release: '17.4'
 
 matrix:
   include:
-    - elixir: '1.6'
+    - elixir: '1.8'
       otp_release: '18.0'
 script:
       - MIX_ENV=test mix do compile, coveralls.json

config/config.exs

@@ -33,4 +33,5 @@ config :snowplow_tracker,
   default_options: [
     timeout: 5000,
     recv_timeout: 2000
-  ]
+  ],
+  table: :snowplow_events

config/test.exs

@@ -0,0 +1,8 @@
+use Mix.Config
+
+config :snowplow_tracker,
+  default_options: [
+    timeout: 5000,
+    recv_timeout: 2000
+  ],
+  table: :snowplow_events_test

lib/application.ex

@@ -0,0 +1,16 @@
+defmodule SnowplowTracker.Application do
+  use Application
+
+  def start(_type, _args) do
+    children =
+      if Mix.env() != :test do
+        [
+          {SnowplowTracker.Emitters.Server, []}
+        ]
+      else
+        []
+      end
+
+    Supervisor.start_link(children, strategy: :one_for_all)
+  end
+end

lib/snowplow_tracker/constants.ex

@@ -12,6 +12,8 @@ defmodule SnowplowTracker.Constants do
   @post_protocol_vendor "com.snowplowanalytics.snowplow"
   @post_protocol_version "tp2"
   @post_content_type "application/json; charset=utf-8"
+  @post_wrapper_bytes 88
+  @post_stm_bytes 22
   @get_protocol_path "i"
   @schema_payload_data "iglu:com.snowplowanalytics.snowplow/payload_data/jsonschema/1-0-4"
   @schema_contexts "iglu:com.snowplowanalytics.snowplow/contexts/jsonschema/1-0-1"
@@ -85,6 +87,8 @@ defmodule SnowplowTracker.Constants do
   def post_protocol_vendor, do: @post_protocol_vendor
   def post_protocol_version, do: @post_protocol_version
   def post_content_type, do: @post_content_type
+  def post_wrapper_bytes, do: @post_wrapper_bytes
+  def post_stm_bytes, do: @post_stm_bytes
 
   # GET Requests
   def get_protocol_path, do: @get_protocol_path

lib/snowplow_tracker/emitter.ex

@@ -4,33 +4,35 @@ defmodule SnowplowTracker.Emitter do
   snowplow collector.
   """
   alias __MODULE__
-
-  alias SnowplowTracker.{Payload, Request, Response, Errors}
+  alias SnowplowTracker.Payload
   alias SnowplowTracker.Emitters.Helper
+  alias SnowplowTracker.Emitters.Unary, as: UnaryEmitter
+  alias SnowplowTracker.Emitters.Bulk , as: BulkEmitter
 
   @keys [
     collector_uri: "localhost",
     request_type: "GET",
     collector_port: nil,
     protocol: "http"
   ]
-
-  defstruct @keys
-
   @type t :: %__MODULE__{
           collector_uri: String.t(),
           request_type: String.t(),
           collector_port: number(),
           protocol: String.t()
         }
 
+  defstruct @keys
+
   # Public API
 
   @spec new(Emitter.t()) :: Emitter.t()
   def new(uri), do: struct(%Emitter{}, collector_uri: uri)
 
   @spec input(Payload.t(), Emitter.t(), struct()) :: {:ok, String.t()} | no_return()
-  def input(%Payload{} = payload, %Emitter{} = emitter, module \\ Helper) do
+  def input(payload, emitter, module \\ Helper)
+
+  def input(%Payload{} = payload, %Emitter{request_type: "GET"} = emitter, module) do
     url =
       module.generate_endpoint(
         emitter.protocol,
@@ -40,16 +42,19 @@ defmodule SnowplowTracker.Emitter do
         emitter.request_type
       )
 
-    with {:ok, response} <- Request.get(url, [], default_options()),
-         {:ok, body} <- Response.parse(response) do
-      {:ok, body}
-    else
-      {:error, error} ->
-        raise Errors.ApiError, Kernel.inspect(error)
-    end
+    UnaryEmitter.create(payload, url)
   end
 
-  defp default_options do
-    Application.get_env(:snowplow_tracker, :default_options) || []
+  def input(%Payload{} = payload, %Emitter{request_type: "POST"} = emitter, module) do
+    url =
+      module.generate_endpoint(
+        emitter.protocol,
+        emitter.collector_uri,
+        emitter.collector_port,
+        payload,
+        emitter.request_type
+      )
+
+    BulkEmitter.create(payload, url)
   end
 end

lib/snowplow_tracker/emitters/bulk.ex

@@ -0,0 +1,21 @@
+defmodule SnowplowTracker.Emitters.Bulk do
+  @moduledoc """
+  This module defines the emitter that will buffer the events to be sent via
+  POST requests. 
+  """
+
+  alias SnowplowTracker.Payload
+  alias SnowplowTracker.Emitters.Cache
+
+  @table Application.get_env(:snowplow_tracker, :table)
+
+  def create(payload, url, table \\ @table) do
+    eid =
+      payload
+      |> Payload.get()
+      |> Map.fetch!("eid")
+
+    Cache.insert({eid, payload, url}, table)
+    {:ok, :success}
+  end
+end

lib/snowplow_tracker/emitters/cache.ex

@@ -0,0 +1,77 @@
+defmodule SnowplowTracker.Emitters.Cache do
+  @moduledoc """
+  Module which implements functions to interact with ETS
+  """
+
+  require Logger
+
+  alias :ets, as: Ets
+
+  @table Application.get_env(:snowplow_tracker, :table)
+  @path Application.get_env(:snowplow_tracker, :path, ".")
+  @lock "lock"
+
+  def init(table \\ @table) do
+    PersistentEts.new(
+      table,
+      "#{@path}/#{Atom.to_string(table)}.tab",
+      [:public, :set, :named_table, {:read_concurrency, true}]
+    )
+
+    Logger.log(:debug, "Table #{table} created successfully!")
+    {:ok, :success}
+  end
+
+  def insert(payload, table \\ @table) do
+    {
+      :ok,
+      Ets.insert(table, payload)
+    }
+  end
+
+  def match(table \\ @table) do
+    {
+      :ok,
+      Ets.match(table, {:"$1", :"$2", :"$3"})
+    }
+  end
+
+  def check_lock(table \\ @table) do
+    {
+      :ok,
+      Ets.lookup(table, @lock)
+    }
+  end
+
+  # coveralls-ignore-start
+  def release_lock(payload, table \\ @table)
+  # coveralls-ignore-stop
+
+  def release_lock({:ok, _msg}, table), do: delete_key(@lock, table)
+
+  def release_lock({:error, _}, _table), do: :ok
+
+  def set_lock(table \\ @table) do
+    try do
+      Ets.insert(table, {@lock, true})
+      Logger.log(:debug, "#{__MODULE__}: Lock set successfully!")
+      {:ok, :success}
+    rescue
+      _e in ArgumentError ->
+        Logger.log(:debug, "#{__MODULE__}: Falied to set lock!")
+        {:error, :failed}
+    end
+  end
+
+  def delete_key(key, table \\ @table) do
+    Ets.delete(table, key)
+    Logger.log(:debug, "Key deleted: #{key}")
+    {:ok, :success}
+  end
+
+  def delete_table(table \\ @table) do
+    Ets.delete(table)
+    Logger.log(:debug, "Deleted table #{table}")
+    File.rm("#{Atom.to_string(table)}.tab")
+  end
+end

lib/snowplow_tracker/emitters/helper.ex

@@ -4,39 +4,51 @@ defmodule SnowplowTracker.Emitters.Helper do
   necessary for the emitter module
   """
 
-  alias SnowplowTracker.{Errors, Constants}
+  alias SnowplowTracker.{Constants, Payload}
 
   @get_method "GET"
+  @post_method "POST"
+  @get_path Constants.get_protocol_path()
+  @post_vendor Constants.post_protocol_vendor()
+  @post_version Constants.post_protocol_version()
 
   # Public API
 
   @doc """
   This function is used to generate the endpoint with the query parameters
   which are used to send events to the collector.
   """
-  def generate_endpoint(protocol, uri, nil = _port, payload, @get_method) do
-    do_generate_endpoint(protocol, uri, "", payload)
+  @spec generate_endpoint(String.t(), String.t(), String.t(), Payload.t(), String.t()) ::
+          String.t()
+  def generate_endpoint(protocol, uri, nil = _port, payload, request_method) do
+    do_generate_endpoint(
+      protocol,
+      uri,
+      "",
+      payload,
+      request_method
+    )
   end
 
-  def generate_endpoint(protocol, uri, port, payload, @get_method) do
-    do_generate_endpoint(protocol, uri, ":#{port}", payload)
-  end
-
-  def generate_endpoint(_protocol, _uri, _port, _payload, invalid_method) do
-    message = "#{invalid_method} method not implemented"
-    raise Errors.NotImplemented, message
+  def generate_endpoint(protocol, uri, port, payload, request_method) do
+    do_generate_endpoint(
+      protocol,
+      uri,
+      ":#{port}",
+      payload,
+      request_method
+    )
   end
 
   # Private API
 
   @doc false
-  defp do_generate_endpoint(protocol, uri, port, payload) do
+  defp do_generate_endpoint(protocol, uri, port, payload, @get_method) do
     params = URI.encode_query(payload)
-    "#{protocol}://#{uri}#{port}/#{protocol_path()}?#{params}"
+    "#{protocol}://#{uri}#{port}/#{@get_path}?#{params}"
   end
 
-  @doc false
-  defp protocol_path do
-    Constants.get_protocol_path()
+  defp do_generate_endpoint(protocol, uri, port, _payload, @post_method) do
+    "#{protocol}://#{uri}#{port}/#{@post_vendor}?#{@post_version}"
   end
 end