Merge pull request #27 from poanetwork/ferigis.17.sending_history

[#17] Send history to the Dashboard

lib/poa_agent/entity/ethereum/history.ex

@@ -1,5 +1,11 @@
 defmodule POAAgent.Entity.Ethereum.History do
   @moduledoc false
 
-  @type t :: [POAAgent.Entity.Ethereum.Block.t()]
+  @type t :: %__MODULE__{
+    history: [POAAgent.Entity.Ethereum.Block.t()]
+  }
+
+  defstruct [
+    :history
+  ]
 end

lib/poa_agent/entity/ethereum/pending.ex

@@ -9,4 +9,9 @@ defmodule POAAgent.Entity.Ethereum.Pending do
     pending: nil
   ]
 
+  defimpl POAAgent.Entity.NameConvention do
+    def from_elixir_to_node(x) do
+      Map.from_struct(x)
+    end
+  end
 end

lib/poa_agent/plugins/collectors/eth/latest_block.ex

@@ -5,6 +5,8 @@ defmodule POAAgent.Plugins.Collectors.Eth.LatestBlock do
   This is a Collector's Plugin which makes requests to a Ethereum node in order to know if
   a new block has been added.
 
+  It also sends the history when the plugin starts.
+
   This Collector needs the url of the node to iteract. That url must be placed in the args field 
   in the config file. For example:
 
@@ -23,7 +25,17 @@ defmodule POAAgent.Plugins.Collectors.Eth.LatestBlock do
   def init_collector(args) do
     :ok = config(args)
 
-    {:ok, %{last_block: get_latest_block()}}
+    with block_number <- get_latest_block(),
+         {:ok, block} <- Ethereumex.HttpClient.eth_get_block_by_number(block_number, :false)
+    do
+      block = format_block(block)
+      range = history_range(block, 0)
+      history = history(range)
+
+      {:transfer, [block, history], %{last_block: get_latest_block()}}
+    else
+      _error -> {:ok, %{last_block: get_latest_block()}}
+    end
   end
 
   @doc false
@@ -34,9 +46,9 @@ defmodule POAAgent.Plugins.Collectors.Eth.LatestBlock do
         {:notransfer, state}
       ^latest_block ->
         {:notransfer, state}
-      latest_block ->
+      block_number ->
-        {:ok, block} = Ethereumex.HttpClient.eth_get_block_by_number(latest_block, :false)
+        {:ok, block} = Ethereumex.HttpClient.eth_get_block_by_number(block_number, :false)
-        {:transfer, format_block(block), %{state | last_block: latest_block}}
+        {:transfer, format_block(block), %{state | last_block: block_number}}
     end
   end
 
@@ -46,6 +58,19 @@ defmodule POAAgent.Plugins.Collectors.Eth.LatestBlock do
     :ok
   end
 
+  @doc false
+  def history(range) do
+    history = for i <- range do
+      block_number = "0x" <> Integer.to_string(i, 16)
+      {:ok, block} = Ethereumex.HttpClient.eth_get_block_by_number(block_number, :false)
+      format_block(block)
+    end
+
+    %POAAgent.Entity.Ethereum.History{
+      history: Enum.reverse(history)
+    }
+  end
+
   @doc false
   defp config([url: url]) do
     Application.put_env(:ethereumex, :url, url)
@@ -61,7 +86,7 @@ defmodule POAAgent.Plugins.Collectors.Eth.LatestBlock do
   end
 
   @doc false
-  defp format_block(block) do
+  defp format_block(block) when is_map(block) do
     difficulty = POAAgent.Format.Literal.Hex.decimalize(block["difficulty"])
     gas_limit = String.to_integer(POAAgent.Format.Literal.Hex.decimalize(block["gasLimit"]))
     gas_used = String.to_integer(POAAgent.Format.Literal.Hex.decimalize(block["gasUsed"]))
@@ -94,5 +119,17 @@ defmodule POAAgent.Plugins.Collectors.Eth.LatestBlock do
       uncles: block["uncles"]
     }
   end
+  defp format_block(_) do
+    nil
+  end
+
+  defp history_range(block, last_block) do
+    max_blocks_history = 40
+
+    from = Enum.max([block.number - max_blocks_history, last_block + 1])
+    to = Enum.max([block.number, 0])
+
+    from..to
+  end
 
 end

lib/poa_agent/plugins/transfers/web_socket/primus.ex

@@ -40,17 +40,23 @@ defmodule POAAgent.Plugins.Transfers.WebSocket.Primus do
     {:ok, %{client: client, context: context}}
   end
 
-  def data_received(label, data, %{client: client, context: context} = state) do
+  def data_received(label, data, %{client: client, context: context} = state) when is_list(data) do
     require Logger
     Logger.info("Received data from the collector referenced by label: #{label}.")
 
-    event = data
+    :ok = Enum.each(data, fn(message) ->
+      event =
+      message
       |> Primus.encode(context)
       |> Jason.encode!()
       :ok = Primus.Client.send(client, event)
+    end)
 
     {:ok, state}
   end
+  def data_received(label, data, state) do
+    data_received(label, [data], state)
+  end
 
   def handle_message(:ping, %{client: client, context: context} = state) do
 
@@ -99,13 +105,26 @@ defmodule POAAgent.Plugins.Transfers.WebSocket.Primus do
     |> POAAgent.Format.PrimusEmitter.wrap(event: "stats")
   end
 
-  def encode(x, %Primus.State{identifier: i}) do
+  def encode(%POAAgent.Entity.Ethereum.History{} = x, %Primus.State{identifier: i}) do
+    history = for i <- x.history do
+      Entity.NameConvention.from_elixir_to_node(i)
+    end
+
     %{}
     |> Map.put(:id, i)
-    |> Map.put(:history, x)
+    |> Map.put(:history, history)
     |> POAAgent.Format.PrimusEmitter.wrap(event: "history")
   end
 
+  def encode(%POAAgent.Entity.Ethereum.Pending{} = x, %Primus.State{identifier: i}) do
+    x = Entity.NameConvention.from_elixir_to_node(x)
+
+    %{}
+    |> Map.put(:id, i)
+    |> Map.put(:stats, x)
+    |> POAAgent.Format.PrimusEmitter.wrap(event: "pending")
+  end
+
   def information() do
     config = Application.get_env(:poa_agent, __MODULE__)
 
@@ -171,6 +190,23 @@ defmodule POAAgent.Plugins.Transfers.WebSocket.Primus do
 
       {:reply, {:text, event}, state}
     end
+    defp handle_primus_event(["history", %{"max" => max, "min" => min}], state) do
+      context = struct!(Primus.State, Application.get_env(:poa_agent, Primus))
+
+      h = POAAgent.Plugins.Collectors.Eth.LatestBlock.history(min..max)
+
+      history = for i <- h.history do
+        Entity.NameConvention.from_elixir_to_node(i)
+      end
+
+      event = %{}
+      |> Map.put(:id, context.identifier)
+      |> Map.put(:history, history)
+      |> POAAgent.Format.PrimusEmitter.wrap(event: "history")
+      |> Jason.encode!()
+
+      {:reply, {:text, event}, state}
+    end
     defp handle_primus_event(data, state) do
       require Logger