Elixir: Live Notifications

We’ve seen at the beginning of our post series that we wanted to notify users in our application with some administrative messages in real time.

For that we chose to connect all our clients to a room where they will receive these messages through web socket and notify the end user using some front-end tricks.

On the Phoenix side this is pretty straightforward as it natively provides live messaging through web socket or long polling depending on what you need or better said depending on what the network let you have :)

Web socket is a protocol allowing you to have a bidirectional live connection between your clients and server. It will manage all the keep-alive stuff and connection refresh for you. If there is a problem during the bidirectional communication it will try to fallback to long polling if possible. We'll see how to use it with Phoenix.

So first of all, you’ll need to define an endpoint for your web socket, to do this, you should add to your endpoint.ex.

socket("/socket", MyAppWeb.UserSocket)

In your UserSocket module you’ll need to specify the transport you want to manage and the channels that will be available to the clients. We decided to call the publishing channel room:lobby but you can give it any name you want. It also provides some hook function for connecting or identifying the clients. We do not need that at the moment so I’ll just show you a basic user_socket.ex.

defmodule MyAppWeb.UserSocket do
  use Phoenix.Socket

  ## Channels

  channel("room:lobby", MyAppWeb.RoomChannel)

  ## Transports

  transport(:websocket, Phoenix.Transports.WebSocket)
  transport(:longpoll, Phoenix.Transports.LongPoll)
end

As we see we defined the room and attached the logic associated to it in RoomChannel, this will manage the message dispatching to our users. In our case we want to perform a simple join to the channel to allow channel access and a simple new_msg action so the user can receive messages.

defmodule MyAppWeb.RoomChannel do
  use Phoenix.Channel

  def join("room:lobby", _message, socket) do
    {:ok, socket}
  end

  def handle_in("new_msg", %{"body" => body}, socket) do
    broadcast!(socket, "new_msg", %{body: body})

    {:noreply, socket}
  end
end

Then you’ll need to join the channel on the client side. Phoenix offers a phoenix.js library that supports this pub sub design out of the box, as web socket is just the communication protocol but do not provides more complex patterns. You can add to your socket.js file

socket.connect()


// Now that you are connected, you can join channels with a topic:

let channel = socket.channel("room:lobby", {})


channel.on("new_msg", payload => {

    let messageItem = document.createElement("li")

    messageItem.innerText = `[${Date()}] ${payload.body}`

    messagesContainer.appendChild(messageItem)

})


channel.join()

.receive("ok", resp => {
    console.log("Joined successfully", resp)
})

.receive("error", resp => {
    console.log("Unable to join", resp)
})

You can then push messages to your clients by adding a form in your back office to allow customer service to send messages easily that will call the following code

let channel = socket.channel("room:lobby", {})

let chatInput = document.querySelector("#chat-input")

let messagesContainer = document.querySelector("#messages")


chatInput.addEventListener("keypress", event => {

    if (event.keyCode === 13) {

        channel.push("new_msg", {
            body: chatInput.value
        })

        chatInput.value = ""

    }

})


channel.join()

.receive("ok", resp => {
    console.log("Joined successfully", resp)
})

.receive("error", resp => {
    console.log("Unable to join", resp)
})

Be careful though that we do not manage any authentification system so anyone can eventually use the channel to send messages to your customers.

You can play a little with your channel by using a command line web socket client and try to follow the Phoenix.Socket.Message format.

%Phoenix.Socket.Message{
  event: term(),
  payload: term(),
  ref: term(),
  topic: term()
}

The event contains the event name like phx_join, the topic would be your channel name, ref is an identifier for the request, I generally put "1". The payload is the content expected by the channel handler.

For your information, the web socket and long poll urls are separated and you can connect using a web socket client :

wsta 'ws://localhost:4000/socket/websocket'

Or

curl 'http://localhost:4000/socket/longpoll'

Now that we've made our web socket communication we need to be sure that only our customers will connect. To do that you'll need to use a library we've already seen named Guardian. What we'll do is that we'll provide the user token on client side while connecting, this way our server will be able to check user token and attach it to the session. We'll also attach user to a channel corresponding to his UUID, this way we'll be able to send him messages like forcing deconnect for example.

So first of all you need to update your client to inject the user token. I did it by modifying app.html.eex body end this way :

<script>window.token = "<%= assigns[:token] %>";</script>
<script src="<%= static_path(@conn, "/js/app.js") %>"></script>

Then you'll need to pass the token on channel connection by editing socket.js this way :

import {Socket} from "phoenix"

let socket = new Socket("/socket", {params: {token: window.token}})

Now that our client is giving the token to our web socket handler we'll need to change the way connection is handled in user_socket.ex by loading the user resource from the token :

def connect(%{"token" => token}, socket) do
  case Account.Guardian.resource_from_token(token) do
    {:ok, %User{} = resource, _claims} ->
      {:ok, assign(socket, :current_user, resource.id)}

    {:error, reason} ->
      Logger.warn(fn ->
        "Websocket unauthorized: #{inspect(reason)}"
      end)

      :error
  end
end

I'll let you refer to the post about account management to see how to define your Account.Guardian module to load from database the user on token decoding. We see that the function returns the resource and the claims that we use to be copyed to socket under the atom :current_user.

Then we use the id function to be able to manage all active sockets for a user. This way we can for example for disconnection from all socket for a specific user :

# Would allow you to broadcast a "disconnect" event and terminate
# all active sockets and channels for a given user:
#
#     GiApiWeb.Endpoint.broadcast("user_socket:#{user.id}", "disconnect", %{})
#
# Returning `nil` makes this socket anonymous.

def id(socket), do: "user_socket:#{socket.assigns.current_user}"

And there you are, a secured connection with an identifier allowing you to know which user is connected.

Hope you enjoyed the quick tutorial, now you can go further and implement some more complex channel handlers.