Merge pull request #2 from rveshovda/genstage

GenStage

README.md

@@ -33,7 +33,7 @@ If [available in Hex](https://hex.pm/docs/publish), the package can be installed
 Simply call:
 
   ```elixir
-  XGPS.Ports_supervisor.start_port("name-of-port")
+  XGPS.Ports.start_port("name-of-port")
   ```
 
 ### Config
@@ -68,26 +68,25 @@ This usage pattern is mostly for testing.
 Call:
 
   ```elixir
-  XGPS.Ports_supervisor.get_one_position
+  XGPS.Ports.get_one_position
   ```
 
 to get the latest fixed positions.
 
 Pay attention to the has_fix if it is true or false. If has_fix=false, you cannot trust the other values.
 
-### Automatically (GenEvent)
-The most common usage pattern is to subscribe to the GenEvent publisher running
+### Automatically (GenStage)
+The most common usage pattern is to subscribe to the GenStage producer running.
 Check out the code inside the example-folder for an implementation for a subscriber. You need to implement (or copy) similar code to your side to receive new positions.
 
 ## Usage: simulation
 
 ### Starting manually
 Start a simulated port by calling the following:
 ```elixir
-XGPS.Ports_supervisor.start_port(:simulate)
+XGPS.Ports.start_port(:simulate)
 ```
 
-
 ### Auto-start from config
 By adding a line to config:
 ```elixir
@@ -97,13 +96,12 @@ config :xgps, port_to_start: {:simulate,}
 ### Sending simulated position
 Send a simulated position using one of the following commands:
 ```elixir
-XGPS.Ports_supervisor.send_simulated_no_fix()
-XGPS.Ports_supervisor.send_simulated_position(1.1,2.2,3.3) # lat, lon, alt
+XGPS.Ports.send_simulated_no_fix()
+XGPS.Ports.send_simulated_position(1.1,2.2,3.3) # lat, lon, alt
 ```
 
 ## Future development
 - Simulation reading from file
-- Consider GenStage
 
 ## Note
 This application was tested on a Raspberry Pi using the AdaFruit Ultimate GPS ([1](https://www.adafruit.com/products/746), [2](https://www.adafruit.com/products/2324)), which essentially uses the chip MTK3339. Guarantees for other systems and chips cannot be given. But please provide feedback if working or not on other systems/chips.

lib/xgps.ex

@@ -5,8 +5,8 @@ defmodule XGPS do
     import Supervisor.Spec, warn: false
 
     children = [
-      supervisor(XGPS.Ports_supervisor, []),
-      XGPS.EventManager.child_spec
+      supervisor(XGPS.Ports, []),
+      worker(XGPS.Broadcaster, [])
     ]
 
     opts = [strategy: :one_for_one, name: XGPS.Supervisor]

lib/xgps/broadcaster.ex

@@ -0,0 +1,59 @@
+alias Experimental.{GenStage}
+
+defmodule XGPS.Broadcaster do
+  @moduledoc """
+  Heavily inspired (almost a copy) from the GenEvent-replacement example from GenStage-repo at:
+  https://github.com/elixir-lang/gen_stage
+  """
+   use GenStage
+
+   @doc """
+  Starts the broadcaster.
+  """
+  def start_link() do
+    GenStage.start_link(__MODULE__, :ok, name: __MODULE__)
+  end
+
+  @doc """
+  Sends an event async.
+  """
+  def async_notify(event) do
+    GenStage.cast(__MODULE__, {:notify, event})
+  end
+
+  ## Callbacks
+
+  def init(:ok) do
+    {:producer, {:queue.new, 0, 0}, dispatcher: GenStage.BroadcastDispatcher}
+  end
+
+  def handle_cancel(_, _, {queue, demand, number_of_subscribers}) do
+    {:noreply, [], {queue, demand, number_of_subscribers - 1}}
+  end
+
+  def handle_subscribe(_, _ ,_ ,{queue, demand, number_of_subscribers}) do
+    {:automatic, {queue, demand, number_of_subscribers + 1}}
+  end
+
+  def handle_cast({:notify, _event}, {_queue, _demand, 0}) do
+    {:noreply, [], {:queue.new, 0, 0}}
+  end
+
+  def handle_cast({:notify, event}, {queue, demand, number_of_subscribers}) do
+    dispatch_events(:queue.in(event, queue), demand, [], number_of_subscribers)
+  end
+
+  def handle_demand(incoming_demand, {queue, demand, number_of_subscribers}) do
+    dispatch_events(queue, incoming_demand + demand, [], number_of_subscribers)
+  end
+
+  # TODO: Make sure the queue does not grow too big
+  defp dispatch_events(queue, demand, events, number_of_subscribers) do
+    with d when d > 0 <- demand,
+         {{:value, event}, queue} <- :queue.out(queue) do
+      dispatch_events(queue, demand - 1, [event | events], number_of_subscribers)
+    else
+      _ -> {:noreply, Enum.reverse(events), {queue, demand, number_of_subscribers}}
+    end
+  end
+end

lib/xgps/examples/consumer.ex

@@ -0,0 +1,29 @@
+alias Experimental.{GenStage}
+defmodule XGPS.Example.Consumer do
+  @moduledoc """
+  The GenEvent handler implementation is a simple consumer.
+  """
+  use GenStage
+
+  def start_link() do
+    GenStage.start_link(__MODULE__, :ok)
+  end
+
+  # Callbacks
+
+  def init(:ok) do
+    # Starts a permanent subscription to the broadcaster
+    # which will automatically start requesting items.
+    {:consumer, :ok, subscribe_to: [XGPS.Broadcaster]}
+  end
+
+  @doc """
+  This function will be called once for each report from the GPS.
+  """
+  def handle_events(events, _from, state) do
+    for event <- events do
+      IO.inspect {self(), event}
+    end
+    {:noreply, [], state}
+  end
+end

lib/port/reader.ex → lib/xgps/port/reader.ex

@@ -141,7 +141,7 @@ defmodule XGPS.Port.Reader do
   defp send_update_event({:not_updated, _gps_data}), do: :ok
   defp send_update_event({:updated, gps_data}) do
     Logger.debug(fn -> "New gps_data: : " <> inspect(gps_data) end)
-    XGPS.EventManager.update(gps_data)
+    XGPS.Broadcaster.async_notify(gps_data)
     :ok
   end
 end

lib/ports_supervisor.ex → lib/xgps/ports.ex

@@ -1,16 +1,26 @@
-defmodule XGPS.Ports_supervisor do
+defmodule XGPS.Ports do
   use Supervisor
 
+  @doc """
+  Open one port to be consumed. Needs to have one GPS attached to the port to work.
+  To simulate, give port_name = :simulate
+  """
   def start_port(port_name) do
     Supervisor.start_child(__MODULE__, [{port_name}])
   end
 
+  @doc """
+  Return all the connected port names
+  """
   def get_running_port_names do
     Supervisor.which_children(__MODULE__)
     |> Enum.map(fn({_, pid, :supervisor, _}) -> pid end)
     |> Enum.map(fn(pid) -> XGPS.Port.Supervisor.get_port_name(pid) end)
   end
 
+  @doc """
+  Return the latest position if atteched to GPS.
+  """
   def get_one_position do
     children = Supervisor.which_children(__MODULE__)
     case length(children) do
@@ -22,11 +32,19 @@ defmodule XGPS.Ports_supervisor do
     end
   end
 
+  @doc """
+  Will send one GPS report as the give position.
+  Since this will effectively generate both RMC and GGA sentences, the broadcaster will produce two values
+  """
   def send_simulated_position(lat, lon, alt) when is_float(lat) and is_float(lon) and is_float(alt) do
     now = DateTime.utc_now()
     send_simulated_position(lat, lon, alt, now)
   end
 
+  @doc """
+  Will send one GPS report as the give position.
+  Since this will effectively generate both RMC and GGA sentences, the broadcaster will produce two values
+  """
   def send_simulated_position(lat, lon, alt, date_time) when is_float(lat) and is_float(lon) and is_float(alt) do
     simulators = get_running_simulators()
     case length(simulators) do
@@ -39,11 +57,16 @@ defmodule XGPS.Ports_supervisor do
     end
   end
 
+  @doc """
+  Will send one GPS report as no fix.
+  Since this will effectively generate both RMC and GGA sentences, the broadcaster will produce two values
+  """
   def send_simulated_no_fix() do
     now = DateTime.utc_now()
     send_simulated_no_fix(now)
   end
 
+
   def send_simulated_no_fix(date_time) do
     simulators = get_running_simulators()
     case length(simulators) do
@@ -70,6 +93,8 @@ defmodule XGPS.Ports_supervisor do
     end
   end
 
+  # Callbacks
+
   def init(:ok) do
     children = [
       supervisor(XGPS.Port.Supervisor, [], restart: :transient)

lib/tools.ex → lib/xgps/tools.ex

@@ -1,6 +1,12 @@
 defmodule XGPS.Tools do
+  @moduledoc """
+  Several different helper functions.
+  """
   require Bitwise
 
+  @doc """
+  Will calculate and return a checksum defined for NMEA sentence.
+  """
   def calculate_checksum text do
     Enum.reduce(String.codepoints(text), 0, &xor/2)
   end
@@ -10,16 +16,55 @@ defmodule XGPS.Tools do
     Bitwise.bxor(acc, val)
   end
 
+  @doc """
+  Converts from hex-string to int.
+  ## Examples
+
+      iex> XGPS.Tools.hex_string_to_int "C0"
+      192
+
+  """
   def hex_string_to_int(string) do
     string |> Base.decode16! |> :binary.decode_unsigned
   end
 
+  @doc """
+  Converts from int to hex-string.
+  ## Examples
+
+      iex> XGPS.Tools.int_to_hex_string 192
+      "C0"
+
+  """
   def int_to_hex_string(int) do
     int |> :binary.encode_unsigned |> Base.encode16
   end
 
+  @doc """
+  Converts latitude from degrees, minutes and bearing into decimal degrees
+  ## Examples
+
+      iex> XGPS.Tools.lat_to_decimal_degrees(54, 41.1600, "N")
+      54.686
+
+      iex> XGPS.Tools.lat_to_decimal_degrees(54, 41.1600, "S")
+      -54.686
+
+  """
   def lat_to_decimal_degrees(degrees, minutes, "N"), do: degrees + (minutes/60.0)
   def lat_to_decimal_degrees(degrees, minutes, "S"), do: (degrees + (minutes/60.0)) * (-1.0)
+
+  @doc """
+  Converts longitude from degrees, minutes and bearing into decimal degrees
+  ## Examples
+
+      iex> XGPS.Tools.lon_to_decimal_degrees(25, 15.6, "E")
+      25.26
+
+      iex> XGPS.Tools.lon_to_decimal_degrees(25, 15.6, "W")
+      -25.26
+
+  """
   def lon_to_decimal_degrees(degrees, minutes, "E"), do: degrees + (minutes/60.0)
   def lon_to_decimal_degrees(degrees, minutes, "W"), do: (degrees + (minutes/60.0)) * (-1.0)
 
@@ -30,13 +75,35 @@ defmodule XGPS.Tools do
     {degrees, minutes, bearing}
   end
 
+  @doc """
+  Convert latitude from decimal degrees into degrees, minutes and bearing
+  ## Examples
+
+      iex> XGPS.Tools.lat_from_decimal_degrees(54.686)
+      {54, 41.1600, "N"}
+
+      iex> XGPS.Tools.lat_from_decimal_degrees(-54.686)
+      {54, 41.1600, "S"}
+
+  """
   def  lat_from_decimal_degrees(decimal_degrees) when decimal_degrees < 0.0 do
     degrees = Float.ceil(decimal_degrees) * (-1.0) |> round
     minutes = (decimal_degrees + degrees) * -60.0
     bearing = "S"
     {degrees, minutes, bearing}
   end
 
+  @doc """
+  Convert longitude from decimal degrees into degrees, minutes and bearing
+  ## Examples
+
+      XGPS.Tools.lon_from_decimal_degrees(25.26)
+      {25, 15.6, "E"}
+
+      XGPS.Tools.lon_from_decimal_degrees(-25.26)
+      {25, 15.6, "W"}
+
+  """
   def  lon_from_decimal_degrees(decimal_degrees) when decimal_degrees >= 0.0 do
     degrees = Float.floor(decimal_degrees) |> round
     minutes = (decimal_degrees - degrees) * 60.0