Elixir Pubsub In Less Than 50 Lines

:pg2 is a mostly unknown, but powerful Erlang module. It provides an API for creating process groups.

Process Group

So, what’s a process group? Well… it’s a group of Erlang/Elixir processes.

Perhaps, the correct question would be, why do we care about process groups? Well, process groups are the foundation for publisher-subscribers (pubsubs for short).

PG2

Understanding :pg2 API and how it relates to a pubsub API will make it easier to understand:

• Every process group is a channel e.g. a group called :my_channel is created:

   iex> :pg2.create(:my_channel)
   :ok
  

• Every process in a group is a subscriber e.g. self() is part of :my_channel group:

   iex> :pg2.join(:my_channel, self())
   :ok
  

• A publisher can send/2 messages to a channel e.g. the publisher gets all the members of the group :my_channel and sends "Some message":

   iex> members = :pg2.get_members(:my_channel)
   :ok
   iex> for member <- members, do: send(member, "Some message")
  

• A subscriber will receive the messages in its mailbox:

   iex> flush()
   "Some message"
   :ok
  

• A subscriber can unsubscribe from a channel e.g. self() leaves the group :my_channel:

   iex> :pg2.leave(:my_channel, self())
   :ok
  

• A channel can be deleted:

   iex> :pg2.delete(:my_channel)
   :ok
  

And that’s it! That’s the API. And you know what’s the best thing about it? It can work between connected nodes. Keep reading and you’ll see :)

Implementing a PubSub

A PubSub has three main functions:

• subscribe/1 for subscribing to a channel:

   def subscribe(channel) do
     pid = self()
  
     case :pg2.get_members(channel) do
       members when is_list(members) ->
         if pid in members do
           :ok                     # It's already subscribed.
         else
           :pg2.join(channel, pid) # Subscribes to channel
         end
  
       {:error, {:no_such_group, ^channel}} ->
         :pg2.create(channel)      # Creates channel
         :pg2.join(channel, pid)   # Subscribe to channel
     end
   end
  

• unsubscribe/1 for unsubscribing from a channel.

    def unsubscribe(channel) do
      pid = self()
  
      case :pg2.get_members(channel) do
        [^pid] ->
          :pg2.leave(channel, pid)   # Unsubscribes from channel
          :pg2.delete(channel)       # Deletes the channel
  
        members when is_list(members) ->
          if pid in members do
            :pg2.leave(channel, pid) # Unsubscribes from channel
          else
            :ok                      # It's already unsubscribed
          end
  
        _ ->
          :ok
      end
    end
  

• publish/2 for sending a message to a channel.

   def publish(channel, message) do
     case :pg2.get_members(channel) do
       [_ | _] = members ->
         for member <- members, do: send(member, message)
         :ok
  
       _ ->
         :ok
     end
   end
  

For a full implementation of PubSub you can check this gist.

I usually create a .iex.exs file in my $HOME folder and then run iex. You could do the same with the previous gist by doing the following:

~ $ PUBSUB="https://gist.githubusercontent.com/alexdesousa/4d592fe206cca17393affaefa4c8fd33/raw/4d84894f016bd9eef84bba647c77c62b9c9a6094/pub_sub.ex"
~ $ curl "$PUBSUB" -o .iex.exs
~ $ iex

Distributed PubSub

For our distributed experiment we’ll need two nodes. My machine is called matrix and both nodes will be neo and trinity respectively:

• :neo@matrix:

   alex@matrix ~ $ iex --sname neo
   iex(neo@matrix)1>
  

• :trinity@matrix:

   alex@matrix ~ $ iex --sname trinity
   iex(trinity@matrix)1> Node.connect(:neo@matrix) # Connects both nodes
  

Now :neo@matrix can subscribe to :mainframe channel:

iex(neo@matrix)1> PubSub.subscribe(:mainframe)
:ok

And :trinity@matrix can send a message:

iex(trinity@matrix)2> PubSub.publish(:mainframe, "Wake up, Neo...")
:ok 

Note: Sometimes it takes a bit of time for nodes to synchronize their process groups, so you might need to publish/2 your message twice.

Finally, :neo@matrix should receive the message:

iex(neo@matrix)2> flush()
"Wake up, Neo..."
:ok

And that’s it. A powerful pubsub in a few lines of code thanks to :pg2.

Conclusion

Erlang has several built-in hidden gems like :pg2 that make our lives easier.

Happy coding!