iacobson / umbrella_stage

Subscribes GenStage consumers to producers across Elixir Umbrella apps.

UmbrellaStage

Subscribes GenStage consumers to producers across Elixir Umbrella apps.

Installation

Add umbrella_stage to your list of dependencies in mix.exs:

Elixir >= 1.4 required

def deps do
  [
    {:umbrella_stage, "~> 0.1.1"}
  ]
end

Usage

1. Configuration
   use the UmbrellaStage in the GenStages servers. It required 2 arguments:

• :type - is the GenStage type: :producer, :producer_consumer or :consumer
• :producers - required only for consumers and producer_consumers. It's the list of producers, the consumer (or producer_consumer) will subscribe to, in the following format: {ProducerName, [subscription_options]}

2. Call umbrella_sync_subscribe() in the GenStage init

This version of umbrella_stage implements just the GenStage.sync_subscribe/3 subscription.

Please check this blog post for more details.

Examples

A demo app is available on GitHub.

Consumer

defmodule App1.MyConsumer do
  use UmbrellaStage,
   type: :consumer,
    producers: [
      {MyProducerConsumer, [max_demand: 10]}
    ]

  def init() do
    umbrella_sync_subscribe()
    .....
  end
end

ProducerConsumer

defmodule App2.MyProducerConsumer do
  use UmbrellaStage,
   type: :producer_consumer,
    producers: [
      {MyProducer, [max_demand: 10]}
    ]

  def init() do
    umbrella_sync_subscribe()
    .....
  end
end

Producer

defmodule App3.MyProducer do
  use UmbrellaStage,
    type: :producer

  def init() do
    umbrella_sync_subscribe()
    .....
  end
end

umbrella_stage's Issues

Consumers from another node in the cluster are not registered.

I'm seeing that consumers that exist on another node in the Erlang cluster are not being registered as consumers. I've verified that the consumer

iex(passportqa@192.168.81.6)3> [{pid, producer}] = Registry.lookup(Registry.Subscriptions, :producers)            
[{#PID<0.1073.0>, MyApp.Producer}]                       
iex(passportqa@192.168.81.6)4> UmbrellaStage.Checker.check_producer_subscribers(pid, producer)                    
[]                          
iex(passportqa@192.168.81.6)5> Registry.lookup(Registry.Subscriptions, :consumers)                                
[]                      
iex(passportqa@192.168.81.6)6> Node.list()               
[:"[email protected]",]    # this is the consumer

In case it's useful information, I'm using peerage to cluster pods in kubernetes.