FabienHenon

iex(node2@192.168.1.21)4> Swarm.members(:notification_sender)
[#PID<0.794.0>, #PID<0.824.0>, #PID<0.792.0>, #PID<24706.878.0>]
iex(node2@192.168.1.21)5> *DBG* 'Elixir.Swarm.Tracker' receive info {'DOWN',#Ref<0.3005574842.2392326147.106534>,process,<24706.878.0>,
             noconnection} in state tracking
[debug] [swarm on node2@192.168.1.21] [tracker:handle_monitor] lost connection to "7a20ed8c-98b8-4db5-bb59-dfa9d038aae1" (#PID<24706.878.0>) on node1@192.168.1.21, node is down
[debug] [swarm on node2@192.168.1.21] [tracker:handle_topology_change] topology change (nodedown for node1@192.168.1.21)
[debug] [swarm on node2@192.168.1.21] [tracker:handle_topology_change] #PID<0.792.0> belongs on node1@192.168.1.21
[error] Scheduler BEGIN_HANDOFF %{}
[debug] [swarm on node2@192.168.1.21] [tracker:handle_topology_change] ModNotifications.Workers.Scheduler has requested to be resumed
[warn] [swarm on node2@192.168.1.21] [tracker:broadcast_event] broadcast of event ({:untrack, #PID<0.792.0>}) was not recevied by [:"

FabienHenon

[info] [swarm on node1@192.168.1.21] [tracker:init] started
[info] Running API.Endpoint with Cowboy using http://localhost:4000
Interactive Elixir (1.6.0) - press Ctrl+C to exit (type h() ENTER for help)
iex(node1@192.168.1.21)1> *DBG* 'Elixir.Swarm.Tracker' receive info cluster_join in state cluster_wait
[info] [swarm on node1@192.168.1.21] [tracker:cluster_wait] joining cluster..
[info] [swarm on node1@192.168.1.21] [tracker:cluster_wait] no connected nodes, proceeding without sync
*DBG* 'Elixir.Swarm.Tracker' consume info cluster_join in state cluster_wait
:observer.start
:ok
iex(node1@192.168.1.21)2> *DBG* 'Elixir.Swarm.Tracker' receive info {nodeup,'node2@192.168.1.21',[{node_type,visible}]} in state tracking

FabienHenon

[info] [swarm on node1@192.168.1.21] [tracker:init] started
[info] Running API.Endpoint with Cowboy using http://localhost:4000
Interactive Elixir (1.6.0) - press Ctrl+C to exit (type h() ENTER for help)
iex(node1@192.168.1.21)1> *DBG* 'Elixir.Swarm.Tracker' receive info cluster_join in state cluster_wait
[info] [swarm on node1@192.168.1.21] [tracker:cluster_wait] joining cluster..
[info] [swarm on node1@192.168.1.21] [tracker:cluster_wait] no connected nodes, proceeding without sync
*DBG* 'Elixir.Swarm.Tracker' consume info cluster_join in state cluster_wait
iex(node1@192.168.1.21)2> :observer.start
:ok
*DBG* 'Elixir.Swarm.Tracker' receive call {track,<<"1fc15cb6-8dcb-4377-91e8-02932acc9075">>,

FabienHenon

defmodule ModContact.Import.CsvAdapter do
  use GenServer


  def start_link(name, path) do
    GenServer.start_link(__MODULE__, [name: name, path: path])
  end

  def init(opts) do
    {:ok,

FabienHenon

defmodule ModContact.Import.CsvAdapter do
  use GenServer


  def start_link(name, path) do
    GenServer.start_link(__MODULE__, [name: name, path: path])
  end

  def init(opts) do
    {:ok,

FabienHenon

defmodule RedisApp.Application do
  use Application

  def start(_type, _args) do
    import Supervisor.Spec, warn: false

    children = [
      worker(RedisApp, [])
    ]

FabienHenon

defmodule Redis do
  @name {:global, :redis}

  def start_link do
    Agent.start_link(&Map.new/0, name: @name)
  end

  def set(key, value) do
    Agent.update(@name, &Map.put(&1, key, value))
  end

FabienHenon

defmodule ModContact.Types.FieldTypes do
  import Utils.Gettext

  @external_resource field_types_path = Path.join([__DIR__, "res", "field_types.json"])

  # Loads field types in module attributes
  Module.register_attribute __MODULE__, :field_types, accumulate: true

  {:ok, json} =
    with {:ok, body} <- File.read(field_types_path),

FabienHenon

retrieve_subscriptions()                                                                      # Returns a stream with active subscriptions and users
    |> Flow.from_enumerable(max_demand: 2)                                                        # Start flow
    |> Flow.partition(max_demand: 5, stages: 10, hash: fn {u, s} -> {{u, s}, rem(u.id, 10)} end)  # Partition dispatching by user id to be able to keep unique users
    |> Flow.uniq_by(fn {user, _} -> user.id end)                                                  # Keep unique users
    |> Flow.partition(max_demand: 5, stages: 10)                                                  # Partition
    |> Flow.map(&retrieve_best_place/1)                                                           # Retrieve best place for user
    |> Flow.partition(window: Flow.Window.count(100), stages: 1)                                  # Partition and creating window bound for reduce
    |> Flow.reduce(fn -> [] end, fn item, list -> [item | list] end)