This is a code-reading and exploration post about the WebSockets side of Phoenix. It builds upon some of the tracing techniques showcased in the previous post, to observe some of the internals of Phoenix. It also features some tricks I commonly employ to debug WebSocket related issues. The title and nature of this post are inspired by the marvellous book Ruby Under a Microscope written by Pat Shaughenessy.

WebSockets

The WebSocket Protocol enables two-way communication between a client running untrusted code in a controlled environment to a remote host that has opted-in to communications from that code The goal of this technology is to provide a mechanism for browser-based applications that need two-way communication with servers that does not rely on opening multiple HTTP connections (e.g., using XMLHttpRequest or <iframe> s and long polling)

Conceptually, WebSocket is really just a layer on top of TCP that does the following:

Adds a web origin-based security model for browsers

Adds an addressing and protocol naming mechanism to support multiple services on one port and multiple host names on one IP address

Layers a framing mechanism on top of TCP to get back to the IP packet mechanism that TCP is built on, but without length limits

Includes an additional closing handshake in-band that is designed to work in the presence of proxies and other intermediaries

(source: RFC 6455)

One of the most radical elements of Phoenix is the ease and productivity it brings when developing WebSocket applications.

We'll start off by creating a simple application to dive in some aspects of Phoenix and understand its inner workings.

The Number Generator Application

We'll quickly build a sample application to start the exploration, leveraging the excellent scaffolding tasks of Phoenix, to start the exploration.

The sample application assigns a random Integer to each connected client and streams numbers down the socket on a pre-configured interval.

mix archive.install https://github.com/phoenixframework/archives/raw/master/phoenix_new.ez mix phoenix.new numbers --no-brunch --no-ecto

With the --no-brunch option, we skip generating Brunch related scaffolding, as none of the examples below require JavaScript and --no-ecto to skip database related configuration and modules.

Phoenix@v1.2.5 is used for all the examples of this post.

By design, a Phoenix channel is an abstraction on sending and receiving messages about topics. Phoenix.Transports.WebSocket is the default transport mechanism, there are others available and you can even write your own. You may read the documentation about channels here.

Let's code our first channel by creating a web/channels/integers_channels.ex .

defmodule Numbers.IntegersChannel do use Numbers.Web , :channel def join ( " numbers: " <> type , _params , socket ) when type in ~w( positive negative ) do with type <- type |> String . to_existing_atom , socket <- socket |> assign ( :number , number ( type ) ) |> assign ( :joined_at , NaiveDateTime . utc_now ) do send self ( ) , { :update , type } { :ok , socket } end end def handle_info ( { :update , type } , socket ) , do : socket |> push_update ( type ) defp push_update ( socket , type ) do Process . send_after ( self ( ) , { :update , type } , 1000 ) push socket , " update " , %{ number : number ( type ) } { :noreply , socket } end defp number ( :positive ) , do : :erlang . unique_integer ( [ :positive ] ) defp number ( :negative ) , do : - number ( :positive ) end

Then update the web/channels/user_socket.ex file to route to the channel we just defined:

defmodule Numbers.UserSocket do use Phoenix.Socket ## Channels channel " numbers:* " , Numbers.IntegersChannel ## Transports transport :websocket , Phoenix.Transports.WebSocket def connect ( _params , socket ) do { :ok , socket } end def id ( _socket ) , do : nil end

Two topics are defined, numbers:positive streaming positive numbers and numbers:negative for negative ones.

You may notice that we're assigning a :joined_at attribute to each client. This is just a convention which can come handy to calculate for how long a client has been connected. For real-time join/leave tracking, Phoenix.Presence should be preferred.

We can now start the application using:

iex -S mix phoenix.server

Connecting to WebSockets from the terminal

Now that we have the server up and running, we can initiate some connections using wsta, a cli tool written in Rust ⚙️, which follows the Unix philosophy letting you pipe streams to and from to other scripts or files.

Installation

Debian ❤️

echo 'deb http://download.opensuse.org/repositories/home:/esphen/Debian_8.0/ /' > /etc/apt/sources.list.d/wsta.list apt-get update apt-get install wsta

Mac OS X

brew tap esphen/wsta https://github.com/esphen/wsta.git brew install wsta

Windows

Ensure you have a command prompt with GNU libraries, for example the git prompt, and run the provided binary file from there. You can find binary releases on the releases page.

With wsta installed, we can connect to the positive numbers topic using:

wsta 'ws://localhost:4000/socket/websocket' \ '{"topic":"numbers:positive","event":"phx_join","payload":{},"ref":"1"}'

The output will look like:

Connected to ws://localhost:4000/socket/websocket { "topic" : "numbers:positive" , "ref" : "1" , "payload" : { "status" : "ok" , "response" : { } } , "event" : "phx_reply" } { "topic" : "numbers:positive" , "ref" :null, "payload" : { "number" :62402 } , "event" : "update" } { "topic" : "numbers:positive" , "ref" :null, "payload" : { "number" :62434 } , "event" : "update" } { "topic" : "numbers:positive" , "ref" :null, "payload" : { "number" :62466 } , "event" : "update" }

and for the negative ones:

wsta 'ws://localhost:4000/socket/websocket' \ '{"topic":"numbers:negative","event":"phx_join","payload":{},"ref":"1"}'

The output will look like:

Connected to ws://localhost:4000/socket/websocket { "topic" : "numbers:negative" , "ref" : "1" , "payload" : { "status" : "ok" , "response" : { } } , "event" : "phx_reply" } { "topic" : "numbers:negative" , "ref" :null, "payload" : { "number" :-61890 } , "event" : "update" } { "topic" : "numbers:negative" , "ref" :null, "payload" : { "number" :-61922 } , "event" : "update" } { "topic" : "numbers:negative" , "ref" :null, "payload" : { "number" :-61954 } , "event" : "update" }

You'll notice that wsta will print Disconnected! and stop streaming numbers after around 1 minute. This is due to a default Phoenix.Transports.WebSocket :timeout configuration which specifies:

The timeout for keeping websocket connections open after it last received data, defaults to 60_000ms

To make it easier to interact with WebSockets in development you may set it to :infinity .

Differences between push/3 , broadcast_from/3 and broadcast/3

As you may read in the Phoenix.Channel documentation there are 3 functions push/3 , broadcast_from3/ and broadcast/3 which can be used to send data to connected clients, but have some not so subtle differences.

push/3 Broadcasts the event to the current subscriber

broadcast_from/3 Broadcasts the event to all but the current subscribers of a given topic

broadcast/3 Broadcasts an event to all the subscribers of a given topic

You don't want to use broadcast/3 if you're only interested to send updates to a single client, use push/3 instead 😇.

Phoenix WebSocket Wiring

The supervision tree of the application is the following:

I've changed the number of ranch HTTP acceptor processes to just 5 to reduce clutter, using the following config/config.exs :

config :numbers , Numbers.Endpoint , url : [ host : " localhost " ] , http : [ acceptors : 5 ] , # 👈 secret_key_base : " something-secret " , render_errors : [ view : Numbers.ErrorView , accepts : ~w( html json ) ] , pubsub : [ name : Numbers.PubSub , adapter : Phoenix.PubSub.PG2 ]

With a connected WebSocket client, the above supervision tree will look like:

In the above image, 0.643.0 is a GenServer which holds Socket assigns for a client, it is created by 0.641.0 which loops in cowboy_websocket:handler_loop/4 and monitored by Numbers.PubSub.Local0 .

The cowboy_websocket handler in turn is spawned by 0.340.0 , a ranch_conns_sup which is supervised by 0.339.0 a ranch_listener_sup.

What about all those wild-west themed applications and modules?

Where do they come from?

How do our socket pushes manage to reach the client?

The following passages unravel some of the most important function calls which take place for a WebSocket connection to be established and a message to be pushed from the server to the client.

If you find it hard to catch on the whys and the hows of all those function calls, don't fret, skip to the Phoenix and Cowboy section.

Cowboy 🤠

This is the default web server for Phoenix. It depends on ranch to handle TCP connections and cowlib for HTTP protocol parsing and utility functions.

The HTTP parsing code of cowlib is especially interesting as a demonstration of what can be achieved using pattern-matching.

🎉 Cowboy 2.0 was recently released with support for HTTP/2 and lots of other exciting changes (see announcement and this talk). Phoenix support is on the way (see here).

HTTP/2 support is being finalized thanks to @TheGazler's work in Plug & Phoenix, and of course @lhoguin's continued fantastic work on Cowboy — Phoenix Framework (@elixirphoenix) November 7, 2017

Further below there are code references, which show how Phoenix uses Cowboy to respond to HTTP/HTTPS requests.

Phoenix.PubSub

phoenix_pubsub is an application dependency of Phoenix handling distributed PubSub messaging and Presence.

In order to understand how essential it is and how things are tied to together, we'll trace the framework's data flows.

In our example, a client connects to the channel "numbers:42" , then from the server a message {"number": 42} is broadcasted to all connected clients.

Take a deep breath and let's probe 🔎 Phoenix to find out what goes on under the hood, for our message to reach its recipients.

Subscribing

When a client joins a channel, Phoenix.Socket.Transport.connect/6 is called. It builds a Phoenix.Socket with pubsub_server set to the configured name in the :pubsub setting of the Endpoint. For this sample application it defaults to Numbers.PubSub .

# Signature connect ( endpoint , handler , transport_name , transport , serializer , params ) # Actual arguments connect ( Numbers.Endpoint , Numbers.UserSocket , :websocket , Phoenix.Transports.WebSocket , Phoenix.Transports.WebSocketSerializer , %{ }

It will call Numbers.UserSocket.connect/2 with the following arguments:

connect ( %{ } , % Phoenix.Socket { assigns : %{ } , channel : nil , channel_pid : nil , endpoint : Numbers.Endpoint , handler : Numbers.UserSocket , id : nil , joined : false , pubsub_server : Numbers.PubSub , ref : nil , serializer : Phoenix.Transports.WebSocketSerializer , topic : nil , transport Phoenix.Transports.WebSocket , transport_name : websocket , transport_pid : < 0.421 . 0 > # This is a :cowboy_websocket process } )

Then a Phoenix.Channel.Server process is started by Phoenix.Channel.Server.join/2 .

File: lib/phoenix/channel/server.ex (link)

defmodule Phoenix.Channel.Server do use GenServer def join ( socket , auth_payload ) do Phoenix.Endpoint . instrument socket , :phoenix_channel_join , %{ params : auth_payload , socket : socket } , fn -> ref = make_ref ( ) # 👀 A Phoenix.Channel.Server is started here case GenServer . start_link ( __MODULE__ , { socket , auth_payload , self ( ) , ref } ) do { :ok , pid } -> receive do : ( { ^ ref , reply } -> { :ok , reply , pid } ) :ignore -> receive do : ( { ^ ref , reply } -> { :error , reply } ) { :error , reason } -> Logger . error fn -> Exception . format_exit ( reason ) end { :error , %{ reason : " join crashed " } } end end end def init ( { socket , auth_payload , parent , ref } ) do socket = %{ socket | channel_pid : self ( ) } # 👀 socket.channel will be Numbers.IntegersChannel here case socket . channel . join ( socket . topic , auth_payload , socket ) do { :ok , socket } -> join ( socket , %{ } , parent , ref ) { :ok , reply , socket } -> join ( socket , reply , parent , ref ) { :error , reply } -> send ( parent , { ref , reply } ) :ignore other -> # ✂️ error handling omitted end end defp join ( socket , reply , parent , ref ) do PubSub . subscribe ( socket . pubsub_server , socket . topic , link : true , fastlane : { socket . transport_pid , socket . serializer , socket . channel . __intercepts__ ( ) } ) send ( parent , { ref , reply } ) { :ok , %{ socket | joined : true } } end end

Phoenix.Channel.Server.join/2 is called with arguments:

# Signature join ( socket , auth_payload ) # Actual Arguments join ( % Phoenix.Socket { assigns : %{ } , channel : Numbers.IntegersChannel , channel_pid : nil , endpoint : Numbers.Endpoint , handler : Numbers.UserSocket , id : nil , joined : false , pubsub_server : Numbers.PubSub , ref : nil , serializer : Phoenix.Transports.WebSocketSerializer , topic : " numbers:42 " , transport : Phoenix.Transports.WebSocket , transport_name : websocket , transport_pid : < 0.438 . 0 > } , %{ } )

and the Phoenix.Channel.Server.init/1 callback is invoked with arguments:

# Signature init ( { socket , auth_payload , parent , ref } ) # Actual Arguments init ( { % Phoenix.Socket { assigns : %{ } , channel : Numbers.IntegersChannel , channel_pid : nil , endpoint : Numbers.Endpoint , handler : Numbers.UserSocket , id : nil , joined : false , pubsub_server : Numbers.PubSub , ref : nil , serializer : Phoenix.Transports.WebSocketSerializer , topic : " numbers:42 " , transport :Phoenix.Transports.WebSocket , transport_name : :websocket , transport_pid : < 0.438 . 0 > i } , %{ } , < 0.438 . 0 > , #Ref<0.3148875998.1158676484.67712>})

Then Phoenix.Channel.Server.init/1 will call Phoenix.Channel.Server.join/4

which calls PubSub.subscribe/3 with arguments:

# Signature subscribe ( server , topic , opts ) # Actual Arguments subscribe ( Numbers.PubSub , " numbers:42 " , link : true , fastlane : { < 0.291 . 0 > , Phoenix.Transports.WebSocketSerializer , [ ] } )

File: lib/phoenix/pubsub.ex (link)

defmodule Phoenix.PubSub do def subscribe ( server , topic , opts ) when is_atom ( server ) and is_binary ( topic ) and is_list ( opts ) do call ( server , :subscribe , [ self ( ) , topic , opts ] ) end # 👇 For the subscription call/3 will be called with: # server here is Numbers.PubSub # kind is :subscribe defp call ( server , kind , args ) do [ { ^ kind , module , head } ] = :ets . lookup ( server , kind ) # :ets.lookup(Numbers.PubSub, :subscribe) apply ( module , kind , head ++ args ) end end

The Numbers.PubSub named ETS table will have the following entry for :subscribe :

{ :subscribe , Phoenix.PubSub.Local , [ Numbers.PubSub , 1 ] }

So apply(module, kind, head ++ args) is in this case is a Phoenix.PubSub.Local.subscribe/5 call.

File: lib/phoenix/pubsub/local.ex (link)

defmodule Phoenix.PubSub.Local do # 👇 Will be called with: # subscribe(Numbers.PubSub, 1, <0.332.0>, "numbers:42", # link: true, fastlane: {<0.330.0>,Phoenix.Transports.WebSocketSerializer,[]}) # Also the default pool_size is 1 def subscribe ( pubsub_server , pool_size , pid , topic , opts \\ [ ] ) when is_atom ( pubsub_server ) do { local , gc } = pid |> :erlang . phash2 ( pool_size ) |> pools_for_shard ( pubsub_server ) :ok = GenServer . call ( local , { :monitor , pid , opts } ) true = :ets . insert ( gc , { pid , topic } ) # 🔎 :ets.insert(Numbers.PubSub.GC0, {<0.344.0>, "numbers:42"}) true = :ets . insert ( local , { topic , { pid , opts [ :fastlane ] } } ) # 🔎 :ets.insert(Numbers.PubSub.Local0, # {"numbers:42", {<0.344.0>, {<0.342.0>, Phoenix.Transports.WebSocketSerializer, []}}}) :ok end end

We ended up with 2 new entries in the Numbers.PubSub.GC0 and Numbers.PubSub.Local0 tables. Hopefully it will become clear further down this post how they're used.

We're done with the subscription part 😅, moving on to the broadcast..

Publishing

When we broadcast/3 from one of our channels, as in the following example:

defmodule Numbers.IntegersChannel do use Numbers.Web , :channel def join ( " numbers: " <> type , _params , socket ) do send self ( ) , { :update , type } { :ok , socket } end end def handle_info ( { :update , " 42 " } , socket ) do # 👉 We broadcast to all the connected clients broadcast socket , " update " , %{ number : 42 } { :noreply , socket } end end

Phoenix.Channel.Server.broadcast/3 will be called.

File: lib/phoenix/channel/server (link)

defmodule Phoenix.Channel.Server do # 👇 Will be called with arguments: # broadcast(Numbers.PubSub, "numbers:42", "update", %{number: 42}) def broadcast ( pubsub_server , topic , event , payload ) when is_binary ( topic ) and is_binary ( event ) and is_map ( payload ) do PubSub . broadcast pubsub_server , topic , % Broadcast { topic : topic , event : event , payload : payload } end end

It calls PubSub.broadcast/3 .

File: lib/phoenix/pubsub.ex (link)

defmodule Phoenix.PubSub do def broadcast ( server , topic , message ) when is_atom ( server ) or is_tuple ( server ) , do : call ( server , :broadcast , [ :none , topic , message ] ) defp call ( server , kind , args ) do [ { ^ kind , module , head } ] = :ets . lookup ( server , kind ) apply ( module , kind , head ++ args ) end end

Our sample application will have the following entry for

:ets.lookup(Numbers.PubSub, :broadcast) :

{ :broadcast , Phoenix.PubSub.PG2Server , [ Phoenix.Channel.Server , Numbers.PubSub , 1 ] }

So apply(module, kind, head ++ args) in this case is a Phoenix.PubSub.PG2Server.broadcast/6 call.

File: lib/phoenix/pubsub/pg2_server.ex (link)

defmodule Phoenix.PubSub.PG2Server do # 👇 Will be called with: # broadcast(Phoenix.Channel.Server, Numbers.PubSub, 1, :none, "numbers:42", # %Phoenix.Socket.Broadcast{event: "update", payload: %{number: 42}, topic: "numbers:42"}) def broadcast ( fastlane , server_name , pool_size , from_pid , topic , msg ) do server_name |> get_members ( ) # 👉 Returns the Numbers.PubSub registered process |> do_broadcast ( fastlane , server_name , pool_size , from_pid , topic , msg ) end defp do_broadcast ( pids , fastlane , server_name , pool_size , from_pid , topic , msg ) when is_list ( pids ) do local_node = Phoenix.PubSub . node_name ( server_name ) Enum . each ( pids , fn pid when is_pid ( pid ) and node ( pid ) == node ( ) -> Local . broadcast ( fastlane , server_name , pool_size , from_pid , topic , msg ) { ^ server_name , node_name } when node_name == local_node -> # 🏃 Next Phoenix.PubSub.Local.broadcast/6 is called Local . broadcast ( fastlane , server_name , pool_size , from_pid , topic , msg ) pid_or_tuple -> send ( pid_or_tuple , { :forward_to_local , fastlane , from_pid , topic , msg } ) end ) :ok end defp get_members ( server_name ) do # 👇 Will be called with: # :pg2.get_members({:phx, Numbers.PubSub}) :pg2 . get_members ( pg2_namespace ( server_name ) ) end end

Next Phoenix.PubSub.Local.broadcast/6 is called.

File: lib/phoenix/pubsub/local.ex (link)

defmodule Phoenix.PubSub.Local do # 👇 Will be called with: # broadcast(Phoenix.Channel.Server, Numbers.PubSub, 1, :none, "numbers:42", # %Phoenix.Socket.Broadcast{event: "update", payload: %{number: 42}, topic: "numbers:42"}) def broadcast ( fastlane , pubsub_server , 1 = _pool_size , from , topic , msg ) when is_atom ( pubsub_server ) do do_broadcast ( fastlane , pubsub_server , _shard = 0 , from , topic , msg ) :ok end defp do_broadcast ( fastlane , pubsub_server , shard , from , topic , msg ) do pubsub_server |> subscribers_with_fastlanes ( topic , shard ) # Returns a List of tuples like: # => {#PID<0.359.0>, {#PID<0.357.0>, Phoenix.Transports.WebSocketSerializer, []}} # Where the <0.359.0> => a Phoenix.Channel.Server process # <0.357.0> a cowboy_websocket process # and calls Phoenix.Channel.Server.fastlane/3 |> fastlane . fastlane ( from , msg ) end end

Next Phoenix.Channel.Server.fastlane/3 is called.

File: lib/phoenix/channel/server.ex (link)

defmodule Phoenix.Channel.Server do # Will be called with: # fastlane([{<0.393.0>, {<0.391.0>, Phoenix.Transports.WebSocketSerializer, []}}], # :none, # %Phoenix.Socket.Broadcast{event: "update", payload: %{number: 42}, topic: "numbers:42"}) def fastlane ( subscribers , from , % Broadcast { event : event } = msg ) do Enum . reduce ( subscribers , %{ } , fn { pid , _fastlanes } , cache when pid == from -> cache { pid , nil } , cache -> send ( pid , msg ) cache { pid , { fastlane_pid , serializer , event_intercepts } } , cache -> # 🤓 Read about message interception # https://github.com/phoenixframework/phoenix/blob/v1.2.5/lib/phoenix/channel.ex#L101 if event in event_intercepts do send ( pid , msg ) cache else case Map . fetch ( cache , serializer ) do { :ok , encoded_msg } -> send ( fastlane_pid , encoded_msg ) cache :error -> # 🔎 serializer here is Phoenix.Transports.WebSocketSerializer encoded_msg = serializer . fastlane! ( msg ) # 🔎 fastlane_pid here is a cowboy_websocket process send ( fastlane_pid , encoded_msg ) Map . put ( cache , serializer , encoded_msg ) end end end ) end end

Next the message is serialized and sent to a cowboy_websocket handler process.

File: lib/phoenix/transports/websocket_serializer.ex (link)

defmodule Phoenix.Transports.WebSocketSerializer do @moduledoc false @behaviour Phoenix.Transports.Serializer alias Phoenix.Socket.Message alias Phoenix.Socket.Broadcast @doc """ Translates a `Phoenix.Socket.Broadcast` into a `Phoenix.Socket.Message`. """ def fastlane! ( % Broadcast { } = msg ) do { :socket_push , :text , Poison . encode_to_iodata! ( % Message { topic : msg . topic , event : msg . event , payload : msg . payload } ) } end end

The message is handled by Phoenix.Endpoint.CowboyWebSocket.websocket_info/3 .

File: lib/phoenix/endpoint/cowboy_web_socket.ex (link)

defmodule Phoenix.Endpoint.CowboyWebSocket do # Implementation of the WebSocket transport for Cowboy. @moduledoc false # 🤓 Read about :cowboy_webocket_handler => https://github.com/ninenines/cowboy/blob/1.1.2/src/cowboy_websocket_handler.erl @behaviour :cowboy_websocket_handler @connection Plug.Adapters.Cowboy.Conn def websocket_info ( message , req , { handler , state } ) do handle_reply req , handler , handler . ws_info ( message , state ) end defp handle_reply ( req , handler , { :reply , { opcode , payload } , new_state } ) do { :reply , { opcode , payload } , req , { handler , new_state } } end end

Phoenix.Endpoint.CowboyWebSocket.websocket_info/3 is called with arguments:

websocket_info ( # message = { :socket_push , :text , [ 123 , [ [ 34 , [ " topic " ] , 34 ] , 58 , [ 34 , [ " numbers:42 " ] , 34 ] , 44 , [ 34 , [ " ref " ] , 34 ] , 58 , " null " , 44 , [ 34 , [ " payload " ] , 34 ] , 58 , [ 123 , [ [ 34 , [ " number " ] , 34 ] , 58 , " 42 " ] , 125 ] , 44 , [ 34 , [ " event " ] , 34 ] , 58 , [ 34 , [ " update " ] , 34 ] ] , 125 ] } , # message is the charlist format of # {"topic": "numbers:42", "ref": "null", "payload": "{\"number\": \"42\"}", "event": "update"} # req = { :http_req , #Port<0.12968>, :ranch_tcp, :keepalive, <0.381.0>, "GET", 'HTTP/1.1', { { 127 , 0 , 0 , 1 } , 50079 } , " localhost " , :undefined , 4000 , " /socket/websocket " , :undefined , << >> , :undefined , :undefined , [ ] , [ ] , [ ] , [ websocket_version : 13 , websocket_compress : false ] , :waiting , << >> , undefined , false , :done , [ ] , << >> , undefined } , # {handler, state} = { Phoenix.Transports.WebSocket , %{ channels : %{ " numbers:42 " = > < 0.383 . 0 > } , channels_inverse : %{ < 0.383 . 0 > = > { " numbers:42 " , " 1 " } } , serializer : Phoenix.Transports.WebSocketSerializer , socket : % Phoenix.Socket { assigns : %{ } , channel : nil , channel_pid : nil , endpoint : Numbers.Endpoint , handler : Numbers.UserSocket , id : nil , joined : false , pubsub_server : Numbers.PubSub , ref : nil , serializer : Phoenix.Transports.WebSocketSerializer , topic : nil , transport : Phoenix.Transports.WebSocket , transport_name : :websocket , transport_pid : < 0.381 . 0 > } } } )

Finally Phoenix.Transports.Websocket.ws_info/2 is called.

File: lib/phoenix/transports/websocket.ex (link)

defmodule Phoenix.Transports.WebSocket do @behaviour Phoenix.Socket.Transport def ws_info ( { :socket_push , _ , _encoded_payload } = msg , state ) do format_reply ( msg , state ) end defp format_reply ( { :socket_push , encoding , encoded_payload } , state ) do { :reply , { encoding , encoded_payload } , state } end end

At this point it may still not be evident how the return value of Phoenix.Endpoint.CowboyWebSocket.websocket_info/3 manages to send data down the socket. For this to be demystified, proceed to the next section.

Phoenix and Cowboy

When the numbers application is started, its Numbers.Endpoint supervisor is started, supervising the following children:

Supervisor . which_children Numbers.Endpoint #=> [{Phoenix.CodeReloader.Server, #PID<0.354.0>, :worker, #=> [Phoenix.CodeReloader.Server]}, #=> #=> {{:node, #=> ["node_modules/brunch/bin/brunch", "watch", "--stdin", #=> {:cd, "/Users/zorbash/dev/opensource/blog_examples/numbers"}]}, #=> #PID<0.353.0>, :worker, [Phoenix.Endpoint.Watcher]}, #=> #=> 👇 We'll focus on this supervisor. #=> It's responsible for the HTTP part of the application #=> {Phoenix.Endpoint.Server, #PID<0.344.0>, :supervisor, #=> [Phoenix.Endpoint.Server]}, #=> #=> {Phoenix.PubSub.PG2, #PID<0.337.0>, :supervisor, [Phoenix.PubSub.PG2]}, #=> #=> {Phoenix.Config, #PID<0.336.0>, :worker, [Phoenix.Config]}]

File: lib/phoenix/endpoint/server.ex (link)

defmodule Phoenix.Endpoint.Server do # The supervisor for the underlying handlers. @moduledoc false use Supervisor require Logger def start_link ( otp_app , endpoint , opts \\ [ ] ) do Supervisor . start_link ( __MODULE__ , { otp_app , endpoint } , opts ) end def init ( { otp_app , endpoint } ) do handler = endpoint . config ( :handler ) # 👉 returns Phoenix.Endpoint.CowboyHandler children = for { scheme , port } <- [ http : 4000 , https : 4040 ] , config = endpoint . config ( scheme ) do handler . child_spec ( scheme , endpoint , default ( config , otp_app , port ) ) # 👇 Phoenix.Endpoint.child_spec/3 will return: # {{:ranch_listener_sup, Numbers.Endpoint.HTTP}, # {Phoenix.Endpoint.CowboyHandler, :start_link, # [:http, Numbers.Endpoint, # {:ranch_listener_sup, :start_link, # [Numbers.Endpoint.HTTP, 5, :ranch_tcp, [max_connections: 16384, port: 4000], # :cowboy_protocol, # [env: [dispatch: [{:_, [], # [{["socket", "websocket"], [], Phoenix.Endpoint.CowboyWebSocket, # {Phoenix.Transports.WebSocket, # {Numbers.Endpoint, Numbers.UserSocket, :websocket}}}, # {:_, [], Plug.Adapters.Cowboy.Handler, # {Numbers.Endpoint, []}}]}]]]]}]}, # :permanent, :infinity, :supervisor, [:ranch_listener_sup]} end supervise ( children , strategy : :one_for_one ) end end

Phoenix.Endpoint.CowboyWebSocket.start_link/3 will be called by the Phoenix.Endpoint.Server supervisor:

File: lib/phoenix/endpoint/cowboy_handler.ex (link)

defmodule Phoenix.Endpoint.CowboyHandler do @behaviour Phoenix.Endpoint.Handler require Logger @doc """ Generates a childspec to be used in the supervision tree. """ def child_spec ( scheme , endpoint , config ) do if scheme == :https do Application . ensure_all_started ( :ssl ) end dispatches = for { path , socket } <- endpoint . __sockets__ , { transport , { module , config } } <- socket . __transports__ , # Allow handlers to be configured at the transport level handler = config [ :cowboy ] || default_for ( module ) , do : { Path . join ( path , Atom . to_string ( transport ) ) , handler , { module , { endpoint , socket , transport } } } dispatches = dispatches ++ [ { :_ , Plug.Adapters.Cowboy.Handler , { endpoint , [ ] } } ] # Use put_new to allow custom dispatches config = Keyword . put_new ( config , :dispatch , [ { :_ , dispatches } ] ) { ref , mfa , type , timeout , kind , modules } = Plug.Adapters.Cowboy . child_spec ( scheme , endpoint , [ ] , config ) # Rewrite MFA for proper error reporting mfa = { __MODULE__ , :start_link , [ scheme , endpoint , mfa ] } { ref , mfa , type , timeout , kind , modules } end defp default_for ( Phoenix.Transports.LongPoll ) , do : Plug.Adapters.Cowboy.Handler defp default_for ( _ ) , do : nil @doc """ Callback to start the Cowboy endpoint. """ # 👉 start_link/3 will be called with: # [ # :http, # Numbers.Endpoint, # { # :ranch_listener_sup, :start_link, [ # Numbers.Endpoint.HTTP, 5, :ranch_tcp, [max_connections: 16384, port: 4000], # :cowboy_protocol, [env: [dispatch: [{:_, [], # [{["socket", "websocket"], [], Phoenix.Endpoint.CowboyWebSocket, # {Phoenix.Transports.WebSocket, # {Numbers.Endpoint, Numbers.UserSocket, :websocket}}}, # {:_, [], Plug.Adapters.Cowboy.Handler, # {Numbers.Endpoint, []}}]}]]] # ] # } # ] # ✂️ Parts of the env responsible for livereloading have been omitted. def start_link ( scheme , endpoint , { m , f , [ ref | _ ] = a } ) do # ref is used by Ranch to identify its listeners, defaulting # to plug.HTTP and plug.HTTPS and overridable by users. # 👀 apply/3 becomes :ranch_listener_sup.start_link/6 case apply ( m , f , a ) do { :ok , pid } -> Logger . info info ( scheme , endpoint , ref ) { :ok , pid } { :error , { :shutdown , { _ , _ , { { _ , { :error , :eaddrinuse } } , _ } } } } = error -> Logger . error [ info ( scheme , endpoint , ref ) , " failed, port already in use " ] error { :error , _ } = error -> error end end end

Next :ranch_listener_sup.start_link/6 is called.

File: src/ranch_listener_sup.erl (link)

- module ( ranch_listener_sup ) . - behaviour ( supervisor ) . - export ( [ start_link / 6 ] ) . - export ( [ init / 1 ] ) . - spec start_link ( ranch : ref ( ) , non_neg_integer ( ) , module ( ) , any ( ) , module ( ) , any ( ) ) - > { ok , pid ( ) } . % 👉 start_link/6 is called as: % start_link( % Ref = Numbers.Endpoint.HTTP, % NbAcceptors = 5, % Transport = :ranch_tcp, % TransOpts = [max_connections: 16384, port: 4000], % Protocol = :cowboy_protocol, % ProtoOpts = [env: % [dispatch: [{:_, [], % [{["socket", "websocket"], [], Phoenix.Endpoint.CowboyWebSocket, % {Phoenix.Transports.WebSocket, % {Numbers.Endpoint, Numbers.UserSocket, :websocket}}}, % {:_, [], Plug.Adapters.Cowboy.Handler, % {Numbers.Endpoint, []}}]}]] % ] start_link ( Ref , NbAcceptors , Transport , TransOpts , Protocol , ProtoOpts ) - > MaxConns = proplists : get_value ( max_connections , TransOpts , 1024 ) , % 👀 The options of the new listener are kept in an ETS table ranch_server : set_new_listener_opts ( Ref , MaxConns , ProtoOpts ) , supervisor : start_link ( ? MODULE , { Ref , NbAcceptors , Transport , TransOpts , Protocol } ) . init ( { Ref , NbAcceptors , Transport , TransOpts , Protocol } ) - > AckTimeout = proplists : get_value ( ack_timeout , TransOpts , 5000 ) , ConnType = proplists : get_value ( connection_type , TransOpts , worker ) , Shutdown = proplists : get_value ( shutdown , TransOpts , 5000 ) , ChildSpecs = [ { ranch_conns_sup , { ranch_conns_sup , start_link , [ Ref , ConnType , Shutdown , Transport , AckTimeout , Protocol ] } , permanent , infinity , supervisor , [ ranch_conns_sup ] } , { ranch_acceptors_sup , { ranch_acceptors_sup , start_link , [ Ref , NbAcceptors , Transport , TransOpts ] } , permanent , infinity , supervisor , [ ranch_acceptors_sup ] } ] , { ok , { { rest_for_one , 10 , 10 } , ChildSpecs } } .

We can fetch (or even modify 🙈) the listener opts:

:ets . lookup :ranch_server , { :opts , Numbers.Endpoint.HTTP } [ { { :opts , Numbers.Endpoint.HTTP } , [ env : [ dispatch : [ { :_ , [ ] , [ { [ " socket " , " websocket " ] , [ ] , Phoenix.Endpoint.CowboyWebSocket , { Phoenix.Transports.WebSocket , { Numbers.Endpoint , Numbers.UserSocket , :websocket } } } , { :_ , [ ] , Plug.Adapters.Cowboy.Handler , { Numbers.Endpoint , [ ] } } ] } ] ] ] } ]

The Phoenix.Endpoint.Server will supervise the following children:

[ { { :ranch_listener_sup , Numbers.Endpoint.HTTP } , #PID<0.345.0>, :supervisor, [ :ranch_listener_sup ] } ]

and that child will supervise:

[ { :ranch_acceptors_sup , #PID<0.350.0>, :supervisor, [:ranch_acceptors_sup]}, { :ranch_conns_sup , #PID<0.349.0>, :supervisor, [:ranch_conns_sup]}]

A ranch_acceptors_sup uses ranch_tcp.listen/1 to start listening.

ranch_tcp is a wrapper around gen_tcp and gen_tcp.listen/2 sets up a socket to listen on a random port on the local host.

ranch_acceptors_sup then starts a pool of acceptor supervised processes.

Each one of them waits (with :infinity timeout) to accept a connection on the listening socket.

If a connection is established, the ranch_acceptor will transfer the control of the socket to the connections supervisor so that it receives messages from the socket.

When a TCP connection is established ranch_conns_sup will start a cowboy_protocol process using :cowboy_protocol.start_link/4. When the cowboy_protocol process is started successfully, ranch_conns_sup will transfer the control of the socket to that process so that it receives messages from the socket.

The cowboy_protocol handler is responsible for receiving and parsing messages of the HTTP protocol. It handles requests by executing all the layers of its middleware stack. By default this stack contains cowboy_router .

cowboy_router.execute/2 will be called as:

:cowboy_router . execute ( { :http_req , #Port<0.12695>, :ranch_tcp, :keepalive, <0.371.0>, "GET", 'HTTP/1.1', { { 127 , 0 , 0 , 1 } , 61936 } , " localhost>, :undefined, 4000, " / socket / websocket " , :undefined, <<>>, :undefined, :undefined, [{ " host " , " localhost : 4000 " }, { " connection " , " Upgrade " }, { " upgrade " , " websocket " }, { " sec - websocket - version " , " 13 " }, { " sec - websocket - key " , " IpbxMsfW2JriERk8clVssw == " }, { " origin " , " http :/ / localhost " }], [{ " connection " , [ " upgrade " ]}], :undefined, [], :waiting, <<>>, :undefined, false, :waiting, [], <<>>, :undefined}, [{:listener, Numbers.Endpoint.HTTP}, {:dispatch, [{'_',[], [{[ " socket " , " websocket " ], [], Phoenix.Endpoint.CowboyWebSocket, {Phoenix.Transports.WebSocket, {Numbers.Endpoint, Numbers.UserSocket, :websocket}}}, {'_', [], Plug.Adapters.Cowboy.Handler, {Numbers.Endpoint, []}}]}]}] )

It will dispatch the GET /socket/websocket request to Phoenix.Endpoint.CowboyWebSocket which has the behaviour of :cowboy_websocket_handler .

File: lib/phoenix/endpoint/cowboy_websocket.ex (link)

defmodule Phoenix.Endpoint.CowboyWebSocket do # Implementation of the WebSocket transport for Cowboy. @moduledoc false @behaviour :cowboy_websocket_handler @connection Plug.Adapters.Cowboy.Conn @already_sent { :plug_conn , :sent } # 👇 Phoenix.Endpoint.CowboyWebSocket.init/3 will be called with: # {transport, :http} = {:tcp, :http}, # # req = {:http_req, #Port<0.12671>, :ranch_tcp, :keepalive, #PID<0.356.0>, "GET", # :"HTTP/1.1", {{127, 0, 0, 1}, 60935}, "localhost", :undefined, 4000, # "/socket/websocket", :undefined, "", :undefined, [], # [{"host", "localhost:4000"}, {"connection", "Upgrade"}, # {"upgrade", "websocket"}, {"sec-websocket-version", "13"}, # {"sec-websocket-key", "Is5bfu9A9tnZ9AO/V0vrEg=="}, # {"origin", "http://localhost"}], [{"connection", ["upgrade"]}], :undefined, # [], :waiting, "", :undefined, false, :waiting, [], "", :undefined}, # # {module, opts} = {Phoenix.Transports.WebSocket, {Numbers.Endpoint, Numbers.UserSocket, :websocket}} def init ( { transport , :http } , req , { module , opts } ) when transport in [ :tcp , :ssl ] do conn = @connection . conn ( req , transport ) try do # 👀 Phoenix.Transports.WebSocket.init/2 is called here case module . init ( conn , opts ) do { :ok , %{ adapter : { @connection , req } } , args } -> { :upgrade , :protocol , __MODULE__ , req , args } { :error , %{ adapter : { @connection , req } } } -> { :shutdown , req , :no_state } end catch kind , reason -> # Although we are not performing a call, we are using the call # function for now so it is properly handled in error reports. mfa = { module , :call , [ conn , opts ] } { :upgrade , :protocol , __MODULE__ , req , { :error , mfa , kind , reason , System . stacktrace } } after receive do @already_sent -> :ok after 0 -> :ok end end end def upgrade ( req , env , __MODULE__ , { handler , opts } ) do args = [ req , env , __MODULE__ , { handler , opts } ] resume ( :cowboy_websocket , :upgrade , args ) end def resume ( module , fun , args ) do apply ( module , fun , args ) # ✂️ error handling omitted end ## Websocket callbacks # 👇 websocket_init/3 will be called with: # _transport = tcp, # # req = { # :http_req, #Port<0.12964>, :ranch_tcp, :keepalive, <0.373.0>, "GET", 'HTTP/1.1', # {{127,0,0,1},50034}, # "localhost", :undefined, 4000, "/socket/websocket",undefined, # <<>>, :undefined, [], # [{"host", "localhost:4000"}, # {"connection", "Upgrade"}, # {"upgrade", "websocket"}, # {"sec-websocket-version", "13"}, # {"sec-websocket-key", "kxCvlKM3nApp8UdFrqh9Ug=="}, # {"origin", "http://localhost">>}], # [{"upgrade",["websocket"]}, # {"connection",["upgrade"]}], # :undefined, # [{:websocket_version,13},{:websocket_compress,false}], # :waiting,<<>>,:undefined,false,:waiting,[],<<>>,:undefined}, # # {handler, args} = {Phoenix.Transports.WebSocket, # {%Phoenix.Socket{assigns => %{},channel => nil, # channel_pid => nil,endpoint => Numbers.Endpoint, # handler => Numbers.UserSocket, id => nil,joined => false, # pubsub_server => Numbers.PubSub,ref => nil, # serializer => Phoenix.Transports.WebSocketSerializer, # topic => nil, transport => Phoenix.Transports.WebSocket, # transport_name => websocket,transport_pid => <0.373.0>}, # [{:serializer, Phoenix.Transports.WebSocketSerializer}, # {:transport_log,false}, # {:timeout, :infinity}, # {:acceptors,5}]}}) def websocket_init ( _transport , req , { handler , args } ) do { :ok , state , timeout } = handler . ws_init ( args ) { :ok , :cowboy_req . compact ( req ) , { handler , state } , timeout } end end

Phoenix will respond with:

HTTP/1.1 101 Switching Protocols Upgrade: websocket Connection: Upgrade Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=

which is generated from (link) and concludes the WebSocket handshake, so that the data transfer starts. It is now a two-way communication channel where each side can, independently from the other, send data at will.

After a successful handshake, clients and servers transfer data back and forth in conceptual units referred to in this specification as “messages”. On the wire, a message is composed of one or more frames

With an established WebSocket connection, a client ( phoenix.js , see link) will push a:

{ "topic" : "numbers:42" , "event" : "phx_join" , "payload" : { } , "ref" : "1" }

The cowboy_websocket processs, which has the control of the socket, receives a message in its mailbox, decodes it (WebSocket Protocol) and calls Phoenix.Endpoint.CowboyWebSocket.websocket_handle/3 as follows:

Phoenix.Endpoint.CowboyWebSocket . websocket_handle { opcode , payload } = { :text , " { \" topic \" : \" numbers:42 \" , \" event \" : \" phx_join \" , \" payload \" :{}, \" ref \" : \" 1 \" } " } req = { :http_req , #Port<0.12964>, :ranch_tcp, :keepalive, <0.373.0>, "GET",'HTTP/1.1', { { 127 , 0 , 0 , 1 } , 50034 } , " localhost " , :undefined , 4000 , " /socket/websocket " , :undefined , << >> , :undefined , :undefined , [ ] , [ ] , [ ] , [ { websocket_version , 13 } , { websocket_compress , false } ] , waiting , << >> , :undefined , false , done , [ ] , << >> , :undefined } { handler , state } = { Phoenix.Transports.WebSocket , %{ channels : %{ } , channels_inverse : %{ } , serializer : Phoenix.Transports.WebSocketSerializer , socket : % Phoenix.Socket { assigns : %{ } , channel : nil , channel_pid : nil , endpoint = > Numbers.Endpoint , handler = > Numbers.UserSocket , id = > nil , joined = > false , pubsub_server = > Numbers.PubSub , ref = > nil , serializer = > Phoenix.Transports.WebSocketSerializer , topic = > nil , transport = > Phoenix.Transports.WebSocket , transport_name = > websocket , transport_pid = > < 0.373 . 0 > } } } )

Then Phoenix.Transports.WebSocket.ws_handle/3 is called which parses the message as JSON and calls Phoenix.Socket.Transport.dispatch/3 which will pattern-match on the event part of the message so that phx_join will finally call Phoenix.Channel.Server.join/2 . What happens next has already been described in the subscribing section.

Debugging WebSockets Essentials

Armed with some knowledge about the inner-workings of Phoenix, we can have some common questions answered. Use the following code snippets with caution in production environments, as some of the Phoenix functions used are documented as:

This is an expensive and private operation. DO NOT USE IT IN PROD.

Getting a list of all the topics which have subscribers:

pool_size = Application . get_env ( :numbers , Numbers.Endpoint ) [ :pubsub ] [ :pool_size ] || 1 0 . . ( pool_size - 1 ) |> Enum . flat_map & Phoenix.PubSub.Local . list ( Numbers.PubSub , &1 ) #=> ["numbers:42"]

💡 Keep in mind that the default pool_size is 1.

Getting the subscribers for a topic:

pool_size = Application . get_env ( :numbers , Numbers.Endpoint ) [ :pubsub ] [ :pool_size ] || 1 0 . . ( pool_size - 1 ) |> Enum . flat_map & Phoenix.PubSub.Local . subscribers ( Numbers.PubSub , " numbers:42 " , &1 ) #=> [#PID<0.273.0>]

Getting the subscribers for a topic across nodes:

# With the assumptions that all the connected nodes are members of the PubSub cluster: pool_size = Application . get_env ( :numbers , Numbers.Endpoint ) [ :pubsub ] [ :pool_size ] || 1 fun = & Phoenix.PubSub.Local . subscribers ( Numbers.PubSub , " numbers:42 " , &1 ) :rpc . multicall [ node ( ) | Node . list ] , Enum , :flat_map , [ 0 . . ( pool_size - 1 ) , fun ] #=> {[[#PID<0.19570.2363>, #PID<0.4457.2620>, #PID<0.18260.2750>, #PID<0.6396.2762>, #=> #PID<0.7481.2801>, #PID<0.2731.2866>, #PID<0.13608.2904>, #=> #PID<0.20333.2910>], #=> [#PID<37723.31494.2599>, #PID<37723.20571.2660>, #PID<37723.26132.2671>, #=> #PID<37723.30777.2690>, #PID<37723.23804.2712>, #PID<37723.27526.2731>, #=> #PID<37723.7742.2776>]], []} # If you don't like assumptions you can use this to get the actual nodes (with Phoenix.PubSub.PG2 adapter): nodes = :pg2 . get_members ( { :phx , Numbers.PubSub } ) |> Enum . map ( & node / 1 ) [ :" numbers1@autoverse " , :" numbers2@autoverse " ]

Getting the socket state for a subscriber:

# Using the pid from the previous example :recon . get_state ' <0.273.0> ' #=> %Phoenix.Socket{assigns: %{}, channel: Numbers.IntegersChannel, #=> channel_pid: #PID<0.273.0>, endpoint: Numbers.Endpoint, #=> handler: Numbers.UserSocket, id: "users_socket:42", #=> joined: true, pubsub_server: Numbers.PubSub, ref: nil, #=> serializer: Phoenix.Transports.WebSocketSerializer, topic: "numbers:42", #=> transport: Phoenix.Transports.WebSocket, transport_name: :websocket, #=> transport_pid: #PID<0.271.0>} # You may as well use :sys.get_state/2 but always mind to provide a timeout :sys . get_state pid ( " 0.273.0 " ) , 1000

Getting the subscriptions of a client by IP address:

You can use :inet.i

Lists all TCP and UDP sockets, including those that the Erlang runtime system uses as well as those you have created

iex ( numbers @autoverse ) 2 > :inet . i Port Module Recv Sent Owner Local Address Foreign Address State Type 57824 inet_tcp 1932 242 < 0.9779 . 2801 > 10.16 . 194.182 : 4000 10.16 . 74.135 : 54514 ?? ?? STREAM 580432 inet_tcp 1095 130 < 0.9683 . 2817 > 10.16 . 194.182 : 4000 10.16 . 74.135 : 55660 ?? ?? STREAM 581976 inet_tcp 1653 110 < 0.9693 . 2818 > 10.16 . 194.182 : 4000 10.16 . 74.135 : 59498 ?? ?? STREAM 583824 inet_tcp 474 129 < 0.239 . 0 > 10.16 . 194.182 : 4000 10.16 . 10.150 : 59540 ?? ?? STREAM 584016 inet_tcp 565 218 < 0.22683 . 2819 > 10.16 . 194.182 : 4000 10.16 . 74.135 : 36924 ?? ?? STREAM

Let's say that we're interested for the subscriptions of 10.16.74.150 . A heuristic would be to get the processes spawned by the owner process of the socket:

owner = pid " 0.239.0 " links = Process . info ( owner , [ :links ] ) [ :links ] pids = for pid <- links , pid > owner , do : pid #=> [#PID<0.241.0>, #PID<0.246.0>]

Alternatively you can use :recon.tcp to enumerate the TCP ports and keep the ones matching the IP as peername :

search_ip = { 10 , 16 , 74 , 150 } :recon . tcp |> Enum . map ( & :recon . port_info / 1 ) |> Enum . filter ( fn port -> case port [ :type ] [ :peername ] do { ^ search_ip , _ } -> true _ -> false end end ) |> Enum . map ( & ( &1 [ :signals ] [ :connected ] ) )

Recon

Recon is a library to be dropped into any other Erlang project, to be used to assist DevOps people diagnose problems in production nodes.

I've found recon to be an indispensable tool for both regular debugging and critical production issues.

A handful of WebSocket related traces you can attempt with recon are:

# Trace the next 100 subscriptions for the Numbers Phoenix app :recon_trace . calls { :ets , :insert , [ { [ :" Elixir.Numbers.PubSub.Local0 " , :_ ] , [ ] , [ :" $_ " ] } ] } , 100 #=> 0:28:39.377466 <0.621.0> ets:insert('Elixir.Numbers.PubSub.Local0', {<<"numbers:42">>, #=> {<0.621.0>,{<0.619.0>,'Elixir.Phoenix.Transports.WebSocketSerializer',[]}}})

# Trace the next 50 subscriptions for the Numbers Phoenix app :recon_trace . calls { :ets , :insert , [ { [ :" Elixir.Numbers.PubSub.Local0 " , :_ ] , [ ] , [ :" $_ " ] } ] } , 50 #=> 0:28:39.377466 <0.621.0> ets:insert('Elixir.Numbers.PubSub.Local0', {<<"numbers:42">>, #=> {<0.621.0>,{<0.619.0>,'Elixir.Phoenix.Transports.WebSocketSerializer',[]}}})

:recon . get_state ' <0.621.0> ' #=> %Phoenix.Socket{assigns: %{}, channel: Numbers.IntegersChannel, #=> channel_pid: #PID<0.621.0>, endpoint: Numbers.Endpoint, #=> handler: Numbers.UserSocket, id: nil, joined: true, #=> pubsub_server: Numbers.PubSub, ref: nil, #=> serializer: Phoenix.Transports.WebSocketSerializer, topic: "numbers:42", #=> transport: Phoenix.Transports.WebSocket, transport_name: :websocket, #=> transport_pid: #PID<0.619.0>}

# Trace the next 10 calls to pg2.get_members/1 and show the return value :recon_trace . calls { :pg2 , :get_members , [ { :_ , [ ] , [ { :return_trace } ] } ] } , 10 #=> 1:15:03.621778 <0.706.0> pg2:get_members({phx,'Elixir.Numbers.PubSub'}) #=> 1:15:03.621956 <0.706.0> pg2:get_members/1 --> [<0.279.0>] # In a distributed Phoenix scenario, we'd get something like: #=> 1:30:41.842307 <0.257.0> pg2:get_members({phx,'Elixir.Numbers.PubSub'}) #=> 1:30:41.845794 <0.257.0> pg2:get_members/1 --> [<0.279.0>,<16611.342.0>]

If you find it hard to build complex match specs by hand 🙀, you can try ex2ms which is a rough equivalent of :dbg.fun2ms/1 or :ets.fun2ms/1.

Conclusion

The Phoenix codebase is a very interesting one and there's lot to learn from it. I really hope that sharing my quest to understand and explore Phoenix will be beneficial for others. If you found a mistake please submit a pull request or mention it in the comments.