defmodule ExESDB.Repl do @moduledoc """ This module is to interact with the ExESDB.system, running a store called "reg_gh" (Regulate Greenhouse) """ alias ExESDB.Repl.EventGenerator, as: ESGen alias ExESDB.Repl.Observer, as: Observer alias ExESDB.Repl.Producer, as: Producer alias ExESDB.Repl.Subscriber, as: Subscriber alias ExESDB.GatewayAPI, as: API require Logger @store :reg_gh @greenhouse1 "greenhouse1" @greenhouse2 "greenhouse2" @greenhouse3 "greenhouse3" @greenhouse4 "greenhouse4" @greenhouse5 "greenhouse5" @greenhouses [ @greenhouse1, @greenhouse2, @greenhouse3, @greenhouse4, @greenhouse5 ] def store, do: @store def stream1, do: @greenhouse1 def stream2, do: @greenhouse2 def stream3, do: @greenhouse3 def stream4, do: @greenhouse4 def stream5, do: @greenhouse5 def get_opts, do: ExESDB.Options.app_env() def get_streams, do: API.get_streams(@store) def get_subscriptions, do: API.get_subscriptions(@store) @doc """ Append events to a stream. """ @spec append( stream :: atom(), nbr_of_events :: integer() ) :: {:ok, list(), integer()} | {:error, term()} def append(stream, nbr_of_events) do {:ok, version} = API.get_version(@store, stream) events = ESGen.generate_events(version, nbr_of_events) case @store |> API.append_events(stream, events) do {:ok, new_version} -> {:ok, result} = @store |> API.get_events(stream, 1, new_version, :forward) {:ok, result, result |> Enum.count()} {:error, reason} -> {:error, reason} end end def start_observer_for_all_streams do Logger.warning("Starting Observer for all streams in 5 seconds... Switch to :observer application to see it come online.") Process.sleep(5_000) Observer.start(store: @store, type: :by_stream, selector: "$all") end def start_observer_for_stream(stream \\ "$greenhouse-1") do Logger.warning("Starting Observer for [#{stream}] in 5 seconds... Switch to :observer application to see it come online.") Process.sleep(5_000) Observer.start(store: @store, type: :by_stream, selector: stream) end @doc """ Starts an observer for a specific event type. ## Possible event types: - "initialized:v1" (default) - "temperature_measured:v1" - "humidity_measured:v1" - "light_measured:v1" - "fan_activated:v1" - "fan_deactivated:v1" - "light_activated:v1" - "light_deactivated:v1" - "heater_activated:v1" - "heater_deactivated:v1" - "sprinkler_activated:v1" - "sprinkler_deactivated:v1" - "desired_temperature_set:v1" - "desired_humidity_set:v1" - "desired_light_set:v1" """ @spec start_observer_for_event_type(event_type :: String.t()) :: :ok def start_observer_for_event_type(event_type \\ "initialized:v1") do Logger.warning("Starting Observer for [#{event_type}] in 5 seconds... Switch to :observer application to see it come online.") Process.sleep(5_000) Observer.start(store: @store, type: :by_event_type, selector: event_type) :ok end def start_stream_subscriber(name, stream \\ "$greenhouse-1", start_from \\ 0) do Logger.warning( "Starting Subscriber for [#{name}] to [#{stream}] from [#{start_from}] in 5 seconds... Switch to :observer application to see it come online." ) Process.sleep(5_000) Subscriber.start(name, stream, start_from) end def start_producers(nbr_of_producers \\ 10) do Logger.warning("Starting #{nbr_of_producers} Producers in 5 seconds. Switch to :observer application to see them come online.") Process.sleep(5_000) 1..nbr_of_producers |> Enum.each(fn n -> Producer.start("greenhouse-#{n}") end) end end