Merge pull request #519 from podlove/draft/event_consumer

Draft/event consumer

lib/radiator/application.ex

@@ -5,6 +5,7 @@ defmodule Radiator.Application do
 
   use Application
 
+  alias Radiator.Outline.EventConsumer
   alias Radiator.Outline.EventProducer
 
   @impl true
@@ -20,7 +21,8 @@ defmodule Radiator.Application do
       # {Radiator.Worker, arg},
       # Start to serve requests, typically the last entry
       RadiatorWeb.Endpoint,
-      {EventProducer, name: EventProducer}
+      {EventProducer, name: EventProducer},
+      {EventConsumer, name: EventConsumer}
     ]
 
     # See https://hexdocs.pm/elixir/Supervisor.html

lib/radiator/outline/dispatch.ex

@@ -0,0 +1,19 @@
+defmodule Radiator.Outline.Dispatch do
+  @moduledoc false
+
+  alias Radiator.Outline.Event
+  alias Radiator.Outline.EventProducer
+
+  def insert_node(attributes, user_id, event_id) do
+    "insert_node"
+    |> Event.build(attributes, user_id, event_id)
+    |> EventProducer.enqueue()
+  end
+
+  # TODO
+  # update_node
+  # delete_node
+  # move_node
+
+  # list_node different case, sync call
+end

lib/radiator/outline/event.ex

@@ -1,10 +1,11 @@
 defmodule Radiator.Outline.Event do
   @moduledoc false
 
-  def build(event_id, event_type, user_id, payload) do
-    %{
+  alias Radiator.Outline.Event.InsertNodeEvent
+
+  def build("insert_node", payload, user_id, event_id) do
+    %InsertNodeEvent{
       event_id: event_id,
-      event_type: event_type,
       user_id: user_id,
       payload: payload
     }

lib/radiator/outline/event/insert_node_event.ex

@@ -0,0 +1,5 @@
+defmodule Radiator.Outline.Event.InsertNodeEvent do
+  @moduledoc false
+
+  defstruct [:event_id, :user_id, :payload]
+end

lib/radiator/outline/event_consumer.ex

@@ -0,0 +1,48 @@
+defmodule Radiator.Outline.EventConsumer do
+  @moduledoc false
+
+  use GenStage
+
+  alias Radiator.Outline
+  alias Radiator.Outline.Event.InsertNodeEvent
+  alias Radiator.Outline.EventProducer
+
+  def start_link(opts \\ []) do
+    GenStage.start_link(__MODULE__, opts, name: __MODULE__)
+  end
+
+  def init(opts \\ [max_demand: 1]) do
+    {:consumer, :event_producer, subscribe_to: [{EventProducer, opts}]}
+  end
+
+  def handle_events([event], _from, state) do
+    process_event(event)
+
+    {:noreply, [], state}
+  end
+
+  defp process_event(%InsertNodeEvent{payload: payload} = _event) do
+    payload
+    |> Outline.create_node()
+    |> handle_insert_result()
+
+    #      validate
+    #         true->
+    #           database action: insert node()
+    #           create && persist event (event contains all attributes, user, event_id, timestamps)
+    #           broadcast event (topic: episode_id)
+    #           broadcast node (topic: episode_id)
+    #         false->
+    #           log error and return error (audit log)
+  end
+
+  defp handle_insert_result({:ok, node}) do
+    {:ok, node}
+  end
+
+  defp handle_insert_result({:error, _error}) do
+    # log_error_please :-)
+
+    :error
+  end
+end

lib/radiator/outline/event_producer.ex

@@ -8,15 +8,31 @@ defmodule Radiator.Outline.EventProducer do
   end
 
   def init(_opts) do
-    {:producer, []}
+    {:producer, {:queue.new(), 0}}
   end
 
   def enqueue(event) do
     GenStage.cast(__MODULE__, {:enqueue, event})
     :ok
   end
 
-  def handle_cast({:enqueue, event}, state) do
-    {:noreply, [event], state}
+  def handle_cast({:enqueue, event}, {queue, 0}) do
+    queue = :queue.in(event, queue)
+    {:noreply, [], {queue, 0}}
+  end
+
+  def handle_cast({:enqueue, event}, {queue, demand}) do
+    queue = :queue.in(event, queue)
+    {{:value, event}, queue} = :queue.out(queue)
+    {:noreply, [event], {queue, demand - 1}}
+  end
+
+  def handle_demand(_incoming, {queue, demand}) do
+    with {item, queue} <- :queue.out(queue),
+         {:value, event} <- item do
+      {:noreply, [event], {queue, demand}}
+    else
+      _ -> {:noreply, [], {queue, demand + 1}}
+    end
   end
 end

test/radiator/outline/dispatch_test.exs

@@ -0,0 +1,34 @@
+defmodule Radiator.Outline.DispatchTest do
+  use Radiator.DataCase
+
+  import Radiator.AccountsFixtures
+  import Radiator.PodcastFixtures
+
+  # alias Radiator.Outline
+  alias Radiator.Outline.Dispatch
+  alias Radiator.Outline.Node
+
+  describe "outline dispatch" do
+    setup do
+      %{episode: episode_fixture()}
+    end
+
+    test "insert_node does WHAT?", %{episode: episode} do
+      user = user_fixture()
+
+      node = %Node{episode_id: episode.id, content: "something very special 1!1"}
+      attributes = Map.from_struct(node)
+
+      event_id = Ecto.UUID.generate()
+
+      Dispatch.insert_node(attributes, user.id, event_id)
+
+      # _inserted_node =
+      #  Outline.list_nodes()
+      # |> Enum.find(&(&1.content == "something very special 1!1"))
+
+      # assert inserted_node.episode_id == node.episode_id
+      # assert inserted_node.content == node.content
+    end
+  end
+end