Merge pull request #6 from hswick/event-listener

Event listeners

.travis.yml

@@ -6,7 +6,7 @@ elixir:
 install:
   - sudo apt-get update
   - sudo apt-get install nodejs npm
-  - npm install -g ganache-cli
+  - npm install -g ganache-cli@6.1.3
   - mix local.rebar --force # for Elixir 1.3.0 and up
   - mix local.hex --force
   - mix deps.get

README.md

@@ -80,6 +80,55 @@ iex(10)> ExW3.Contract.call(SimpleStorage, :get)
 {:ok, 1}
 ```
 
+## Listening for Events
+
+Elixir doesn't have event listeners like say JS. However, we can simulate that behavior with message passing.
+The way ExW3 handles event filters is with a background process that calls eth_getFilterChanges every cycle.
+Whenever a change is detected it will send a message to whichever process is listening.
+
+```elixir
+# Start the background listener
+ExW3.EventListener.start_link
+
+# Assuming we have already setup our contract called EventTester
+# We can then add a filter for the event listener to look out for
+# by passing in the event name, and the process we want to receive the messages when an event is triggered.
+# For now we are going to use the main process, however, we could pass in a pid of a different process.
+
+filter_id = ExW3.Contract.filter(EventTester, "Simple", self())
+
+# We can then wait for the event. Using the typical receive keyword we wait for the first instance
+# of the event, and then continue with the rest of the code. This is useful for testing.
+receive do
+  {:event, {filter_id, data}} -> IO.inspect data
+end
+
+# We can then uninstall the filter after we are done using it
+ExW3.uninstall_filter(filter_id)
+
+# ExW3 also provides a helper method to continuously listen for events.
+# One use is to combine all of our filters with pattern matching
+ExW3.EventListener.listen(fn result ->
+  case result do
+    {filter_id, data} -> IO.inspect data
+    {filter_id2, data} -> IO.inspect data
+  end
+end
+
+# The listen method is a simple receive loop waiting for `{:event, _}` messages.
+# It looks like this:
+def listen(callback) do
+  receive do
+    {:event, result} -> apply callback, [result]
+  end
+  listen(callback)
+end
+
+# You could do something similar with your own process, whether it is a simple Task or a more involved GenServer.
+```
+
+
+
 # Compiling Soldity
 
 To compile the test solidity contracts after making a change run this command:

lib/exw3.ex

@@ -127,6 +127,30 @@ defmodule ExW3 do
     end
   end
 
+  @spec new_filter(%{}) :: binary()
+  @doc "Creates a new filter, returns filter id"
+  def new_filter(map) do
+    case Ethereumex.HttpClient.eth_new_filter(map) do
+      {:ok, filter_id} -> filter_id
+      err -> err
+    end
+  end
+
+  def get_filter_changes(filter_id) do
+    case Ethereumex.HttpClient.eth_get_filter_changes(filter_id) do
+      {:ok, changes} -> changes
+      err -> err
+    end
+  end
+
+  @spec uninstall_filter(binary()) :: boolean()
+  def uninstall_filter(filter_id) do
+    case Ethereumex.HttpClient.eth_uninstall_filter(filter_id) do
+      {:ok, result} -> result
+      err -> err
+    end
+  end
+
   @spec mine(integer()) :: any()
   @doc "Mines number of blocks specified. Default is 1"
   def mine(num_blocks \\ 1) do
@@ -264,6 +288,83 @@ defmodule ExW3 do
     end
   end
 
+  defmodule Poller do
+    use GenServer
+
+    def start_link do
+      GenServer.start_link(__MODULE__, [], name: EventPoller)
+    end
+
+    def filter(filter_id) do
+      GenServer.cast(EventPoller, {:filter, filter_id})
+    end
+
+    @impl true
+    def init(state) do
+      schedule_work() # Schedule work to be performed on start
+      {:ok, state}
+    end
+
+    @impl true
+    def handle_cast({:filter, filter_id}, state) do
+      {:noreply, [filter_id | state]}
+    end
+
+    @impl true
+    def handle_info(:work, state) do
+      # Do the desired work here
+      Enum.each state, fn filter_id ->
+	send Listener, {:event, filter_id, ExW3.get_filter_changes(filter_id)}
+      end
+      
+      schedule_work() # Reschedule once more
+      {:noreply, state}
+    end
+
+    defp schedule_work() do
+      Process.send_after(self(), :work, 500) # In 1/2 sec
+    end
+  end
+  
+  defmodule EventListener do
+    def start_link do
+      Poller.start_link()
+      {:ok, pid} = Task.start_link(fn -> loop(%{}) end)
+      Process.register(pid, Listener)
+      :ok
+    end
+
+    def filter(filter_id, event_signature, pid) do
+      Poller.filter(filter_id)
+      send Listener, {:filter, filter_id, event_signature, pid}
+    end
+
+    def listen(callback) do
+      receive do
+	{:event, result} -> apply callback, [result]
+      end
+      listen(callback)
+    end
+    
+    defp loop(state) do
+      receive do
+	{:filter, filter_id, event_signature, pid} ->
+	  loop(Map.put(state, filter_id, %{pid: pid, signature: event_signature}))
+	{:event, filter_id, logs} ->
+	  filter_attributes = Map.get(state, filter_id)
+	  unless logs == [] do
+	    Enum.each(logs, fn log ->
+	      data = Map.get(log, "data")
+	      new_data = ExW3.decode_event(data, filter_attributes[:signature])
+	      new_log = Map.put(log, :data, new_data)
+	      send filter_attributes[:pid], {:event, {filter_id, new_log}}
+	    end)
+	  end
+	  loop(state)
+      end
+    end
+  end
+
   defmodule Contract do
     use GenServer
 
@@ -311,11 +412,15 @@ defmodule ExW3 do
       GenServer.call(pid, {:tx_receipt, tx_hash})
     end
 
+    def filter(pid, event_name, other_pid, event_data \\ %{}) do
+      GenServer.call(pid, {:filter, {event_name, other_pid, event_data}})
+    end
+
     # Server
 
     def init(state) do
       if state[:abi] do
-        {:ok, [{:events, init_events(state[:abi])} | state]}
+	{:ok, state ++ init_events(state[:abi])}
       else
         raise "ABI not provided upon initialization"
       end
@@ -327,16 +432,28 @@ defmodule ExW3 do
           v["type"] == "event"
         end)
 
-      signature_types_map =
+      names_and_signature_types_map =
         Enum.map(events, fn {name, v} ->
           types = Enum.map(v["inputs"], &Map.get(&1, "type"))
           names = Enum.map(v["inputs"], &Map.get(&1, "name"))
           signature = Enum.join([name, "(", Enum.join(types, ","), ")"])
 
-          {"0x#{ExW3.encode_event(signature)}", %{signature: signature, names: names}}
+	  encoded_event_signature = "0x#{ExW3.encode_event(signature)}"
+
+          {{encoded_event_signature, %{signature: signature, names: names}}, {name, encoded_event_signature}}
+        end)
+
+      signature_types_map =
+	Enum.map(names_and_signature_types_map, fn {signature_types, _} ->
+	  signature_types
+	end)
+
+      names_map =
+	Enum.map(names_and_signature_types_map, fn {_, names} ->
+	  names
 	end)
 
-      Enum.into(signature_types_map, %{})
+      [events: Enum.into(signature_types_map, %{}), event_names: Enum.into(names_map, %{})]
     end
 
     # Helpers
@@ -405,6 +522,17 @@ defmodule ExW3 do
       {:noreply, [{:address, address} | state]}
     end
 
+    def handle_call({:filter, {event_name, other_pid, event_data}}, _from, state) do
+      unless Process.whereis(Listener) do
+	raise "EventListener process not alive. Call ExW3.EventListener.start_link before using ExW3.Contract.subscribe"
+      end
+      payload = Map.merge(%{address: state[:address], topics: [state[:event_names][event_name]]}, event_data)
+      filter_id = ExW3.new_filter(payload)
+      event_signature = state[:events][state[:event_names][event_name]][:signature]
+      EventListener.filter(filter_id, event_signature, other_pid)
+      {:reply, filter_id, state ++ [event_name, filter_id]}
+    end
+
     # Calls
 
     def handle_call({:deploy, args}, _from, state) do
@@ -462,4 +590,5 @@ defmodule ExW3 do
       {:reply, {:ok, {receipt, formatted_logs}}, state}
     end
   end
+  
 end

test/exw3_test.exs

@@ -133,6 +133,48 @@ defmodule EXW3Test do
     assert data == "Hello, World!"
   end
 
+  test "starts a Contract GenServer and uses the event listener", context do
+    ExW3.Contract.start_link(EventTester, abi: context[:event_tester_abi])
+
+    {:ok, address} =
+      ExW3.Contract.deploy(
+        EventTester,
+        bin: ExW3.load_bin("test/examples/build/EventTester.bin"),
+        options: %{
+          gas: 300_000,
+          from: Enum.at(context[:accounts], 0)
+        }
+      )
+
+    ExW3.Contract.at(EventTester, address)
+
+    {:ok, agent} = Agent.start_link(fn -> [] end)
+  
+    ExW3.EventListener.start_link()
+
+    filter_id = ExW3.Contract.filter(EventTester, "Simple", self())
+
+    {:ok, tx_hash} =
+      ExW3.Contract.send(
+	EventTester,
+	:simple,
+	["Hello, World!"],
+	%{from: Enum.at(context[:accounts], 0)}
+      )
+  
+    receive do
+      {:event, {filter_id, data}} ->
+	Agent.update(agent, fn list -> [data | list] end)	
+    after 3_000 ->
+	raise "Never received event"
+    end
+
+    state = Agent.get(agent, fn list -> list end)
+    assert Enum.at(state, 0) |> is_map
+
+    ExW3.uninstall_filter(filter_id)
+  end
+
   test "starts a Contract GenServer for Complex contract", context do
     ExW3.Contract.start_link(Complex, abi: context[:complex_abi])
 
@@ -217,4 +259,3 @@ defmodule EXW3Test do
     assert ExW3.is_valid_checksum_address("0x2f015c60e0be116b1f0cd534704db9c92118fb6a") == false
   end
 end
-