Trilha Phoenix · Escrevendo um cliente de Channels
Mix.install(
[
{:phoenix, "~> 1.8.0"},
{:phoenix_live_view, "~> 1.1"},
{:phoenix_ecto, "~> 4.6"},
{:ecto_sql, "~> 3.13"},
{:postgrex, ">= 0.0.0"},
{:bandit, "~> 1.5"},
{:jason, "~> 1.4"},
{:kino, "~> 0.14"},
{:swoosh, "~> 1.16"},
{:pbkdf2_elixir, "~> 2.0"}
],
consolidate_protocols: false,
config: [
phoenix: [json_library: Jason, plug_init_mode: :runtime],
swoosh: [api_client: false],
hello: [
{HelloWeb.Endpoint,
[
adapter: Bandit.PhoenixAdapter,
http: [ip: {127, 0, 0, 1}, port: 4010],
server: true,
debug_errors: true,
secret_key_base: String.duplicate("trilha-phoenix-", 5),
render_errors: [formats: [html: HelloWeb.ErrorHTML, json: HelloWeb.ErrorJSON], layout: false],
pubsub_server: Hello.PubSub
]},
{Hello.Mailer, [adapter: Swoosh.Adapters.Local]}
]
]
)
defmodule Trilha do
@moduledoc """
Apoio deste notebook: mostra bytes e mensagens, confere os exercícios com
conexões de verdade ao servidor e guarda as perguntas e as soluções.
Não faz parte do Phoenix. Você não precisa ler este módulo para seguir o guia:
neste notebook, o cliente é escrito nas células, à vista.
"""
@endpoint HelloWeb.Endpoint
# --- exibição ----------------------------------------------------------------
@doc "Mostra os bytes de um binário em hexadecimal, 16 por linha, com o texto ao lado."
def bytes(binario) do
linhas =
binario
|> :binary.bin_to_list()
|> Enum.chunk_every(16)
|> Enum.map(fn pedaco ->
hex = Enum.map_join(pedaco, " ", &(&1 |> Integer.to_string(16) |> String.pad_leading(2, "0")))
texto = Enum.map_join(pedaco, fn b -> if b in 32..126, do: <<b>>, else: "." end)
String.pad_trailing(hex, 49) <> texto
end)
Kino.Markdown.new(cerca(Enum.join(linhas, "\n"), ""))
end
defp cerca(texto, linguagem) do
marca = String.duplicate("`", 3)
marca <> linguagem <> "\n" <> String.trim(texto) <> "\n" <> marca
end
# --- perguntas -------------------------------------------------------------
@doc "Uma pergunta de múltipla escolha, respondida antes de ver a explicação."
def pergunta(id) do
%{texto: texto, opcoes: opcoes, correta: correta, explicacao: explicacao} = Map.fetch!(perguntas(), id)
quadro = Kino.Frame.new()
letras = Enum.map(opcoes, fn {letra, _texto} -> {letra, to_string(letra)} end)
lista = Enum.map_join(opcoes, "\n", fn {letra, opcao} -> "* **#{letra})** #{opcao}" end)
formulario =
Kino.Control.form(
[resposta: Kino.Input.select("Sua resposta", [{nil, "Escolha"} | letras])],
submit: "Conferir"
)
Kino.listen(formulario, fn %{data: %{resposta: resposta}} ->
retorno =
cond do
resposta == nil -> "Escolha uma opção antes de conferir."
resposta == correta -> "**Certo.** " <> explicacao
true -> "**Ainda não.** Releia o trecho acima e tente outra opção."
end
Kino.Frame.render(quadro, Kino.Markdown.new(retorno))
end)
Kino.Layout.grid([Kino.Markdown.new("**Antes de continuar:** " <> texto <> "\n\n" <> lista), formulario, quadro])
end
defp perguntas do
%{
a_qualquer_hora: %{
texto:
"um cliente entrou em `room:lobby` e não enviou mais nada. Outra pessoa escreve na sala. Como a mensagem chega a esse cliente?",
opcoes: [
a: "Não chega: o cliente só recebe algo como resposta a um pedido seu.",
b: "O cliente precisa perguntar de tempos em tempos se há mensagens novas.",
c: "O servidor a envia pela conexão aberta, na hora, sem o cliente pedir.",
d: "O servidor abre uma segunda conexão até o cliente para entregar."
],
correta: :c,
explicacao:
"Um WebSocket é bidirecional: depois de aberto, qualquer lado envia quando quiser. Por isso um cliente não pode ser escrito como \"envio e leio a resposta\": entre um envio e a resposta dele podem chegar mensagens de outros assuntos. O guia diz que os clientes costumam usar callbacks; o cliente deste notebook usa uma caixa de mensagens, que resolve o mesmo problema."
},
mesma_conexao: %{
texto:
"um cliente quer participar de `room:lobby` e de `vitrine:lobby`, os dois servidos pelo mesmo `UserSocket`. De quantas conexões WebSocket ele precisa?",
opcoes: [
a: "Uma: os dois tópicos dividem a conexão, e o nome do tópico em cada mensagem separa as conversas.",
b: "Duas: uma por tópico.",
c: "Duas: uma para enviar e outra para receber.",
d: "Depende de quantos channels o servidor tem."
],
correta: :a,
explicacao:
"O guia diz: um cliente pode entrar em quantos tópicos quiser pela mesma conexão. No servidor, cada tópico vira um processo de channel; no fio, a terceira posição de cada mensagem diz de qual conversa ela é. Tópicos de sockets diferentes (`/socket` e `/mobile`) pedem conexões diferentes."
},
sem_upgrade: %{
texto:
"o caminho `/socket/websocket` existe no servidor. O que acontece com um `GET` comum para ele, sem os cabeçalhos de upgrade?",
opcoes: [
a: "O servidor responde 200 com uma página que explica o protocolo.",
b: "O servidor responde 400: aquele caminho só aceita o pedido de upgrade.",
c: "O servidor responde 404, porque a rota não está no router.",
d: "O servidor aceita e passa a falar WebSocket assim mesmo."
],
correta: :b,
explicacao:
"O caminho não passa pelo router: quem o atende é a declaração `socket` do endpoint. Sem `Connection: Upgrade` e `Upgrade: websocket`, o Bandit recusa com `400 Bad Request` e a mensagem `'connection' header must contain 'upgrade'`. O WebSocket começa como uma requisição HTTP e só vira outra coisa se o cliente pedir e o servidor concordar."
},
recusa: %{
texto:
"o `connect/3` do `MobileSocket` devolve `:error` quando o parâmetro `token` não é o esperado. O que o cliente vê?",
opcoes: [
a: "A conexão abre, e a primeira mensagem do servidor é um erro em JSON.",
b: "A conexão abre, mas todo `phx_join` é recusado.",
c: "O upgrade não acontece: a resposta HTTP é `403 Forbidden`.",
d: "A conexão TCP é cortada sem resposta nenhuma."
],
correta: :c,
explicacao:
"`connect/3` roda durante a requisição HTTP, antes do upgrade. Se devolve `:error`, o servidor responde `403 Forbidden` e não há WebSocket. Os parâmetros da URL são tudo o que o cliente pode dizer nesse momento; por isso o token viaja neles."
},
refs_do_broadcast: %{
texto:
"uma resposta traz a `ref` da mensagem que a causou. E um broadcast, como o `new_msg` que alguém escreveu na sala: que referências ele traz?",
opcoes: [
a: "A `ref` da última mensagem que este cliente enviou.",
b: "Nenhuma: `join_ref` e `ref` vêm `null`.",
c: "A `ref` da mensagem original, do cliente que escreveu.",
d: "Um número de sequência gerado pelo servidor."
],
correta: :b,
explicacao:
"Um broadcast não responde a nada que este cliente fez, então não há `ref` para devolver. No fio ele chega como `[null,null,\"room:lobby\",\"new_msg\",{...}]`. É assim que um cliente separa as duas coisas: mensagem com `ref` é resposta, e vai para quem está esperando; mensagem sem `ref` é novidade, e vai para quem trata os eventos do tópico."
},
ocioso: %{
texto:
"um cliente conecta, entra num tópico e fica só ouvindo. Ninguém escreve nada por um bom tempo. O que acontece com a conexão?",
opcoes: [
a: "Fica aberta enquanto o processo do cliente existir.",
b: "O servidor a fecha depois de um tempo sem receber nada do cliente.",
c: "O servidor envia um ping e, sem resposta, tenta de novo para sempre.",
d: "O sistema operacional a fecha depois de 30 segundos."
],
correta: :b,
explicacao:
"O transporte tem um `:timeout`, 60 segundos por padrão: se o servidor passa esse tempo sem receber nada do cliente, encerra a conexão. Conta o que o cliente envia, não o que ele recebe. O `heartbeat` existe para isso: é uma mensagem que não faz nada, a não ser provar que o cliente continua lá."
},
evento_desconhecido: %{
texto:
"o `RoomChannel` tem `handle_in/3` para `ping`, `shout`, `new_msg` e `explodir`. O cliente envia o evento `\"danca\"`, que não existe. O que volta?",
opcoes: [
a: "Uma resposta com `status: \"error\"` e o motivo `unknown event`.",
b: "Nada: eventos desconhecidos são ignorados.",
c: "Um `phx_error`: nenhuma cláusula casou, e o processo do channel caiu.",
d: "A conexão inteira é fechada."
],
correta: :c,
explicacao:
"Um channel é um processo com cláusulas `handle_in/3`. Um evento sem cláusula levanta `FunctionClauseError`, o processo termina, e o cliente recebe `phx_error` com `{\"reason\":\"channel_crash\"}`. A conexão e os outros tópicos continuam de pé. Quem escreve um cliente precisa saber os nomes dos eventos da aplicação, como o guia diz, e precisa saber reentrar."
},
join_velho: %{
texto:
"o cliente entrou em `room:lobby` com `join_ref` \"1\", o channel caiu, e ele entrou de novo com `join_ref` \"3\". Uma mensagem atrasada chega ao servidor ainda com `join_ref` \"1\". O que o servidor faz?",
opcoes: [
a: "Entrega ao channel novo: o tópico é o mesmo.",
b: "Responde com um erro que cita o `join_ref` antigo.",
c: "Fecha a conexão.",
d: "Descarta a mensagem em silêncio."
],
correta: :d,
explicacao:
"É para isso que o `join_ref` existe: ele diz a qual entrada no tópico a mensagem pertence. Uma mensagem com o `join_ref` de uma entrada que não existe mais é descartada, sem resposta. O guia chama essas mensagens de *stale*. Para o cliente, isso significa não esperar para sempre por uma resposta."
}
}
end
# --- soluções --------------------------------------------------------------
@doc "Um botão que revela a solução de um exercício."
def solucao(id) do
quadro = Kino.Frame.new()
botao = Kino.Control.button("Mostrar a solução")
Kino.listen(botao, fn _evento ->
Kino.Frame.render(quadro, Kino.Markdown.new(cerca(Map.fetch!(solucoes(), id), "elixir")))
end)
Kino.Layout.grid([botao, quadro])
end
defp solucoes do
%{
endereco: ~S'''
defmodule Hello.Endereco do
@doc "O caminho do pedido de upgrade: o caminho do socket, /websocket, a versão e os parâmetros."
def caminho(base, params \\ %{}) do
base <> "/websocket?" <> URI.encode_query([{"vsn", "2.0.0"} | Map.to_list(params)])
end
end
''',
mensagens: ~S'''
defmodule Hello.Mensagens do
@doc "Entrar num tópico."
def join(join_ref, ref, topico, params), do: [join_ref, ref, topico, "phx_join", params]
@doc "Sair de um tópico."
def leave(join_ref, ref, topico), do: [join_ref, ref, topico, "phx_leave", %{}]
@doc "O batimento que mantém a conexão aberta. Não pertence a nenhum join."
def heartbeat(ref), do: [nil, ref, "phoenix", "heartbeat", %{}]
@doc "Um evento da aplicação, para um tópico em que o cliente já entrou."
def push(join_ref, ref, topico, evento, payload), do: [join_ref, ref, topico, evento, payload]
@doc "A lista vira o texto JSON que vai no quadro."
def codificar(mensagem), do: Jason.encode!(mensagem)
@doc "O texto de um quadro vira um mapa com as cinco partes da mensagem."
def decodificar(texto) do
[join_ref, ref, topic, event, payload] = Jason.decode!(texto)
%{join_ref: join_ref, ref: ref, topic: topic, event: event, payload: payload}
end
end
''',
respostas: ~S'''
defmodule Hello.Cliente do
@moduledoc """
Um cliente de Phoenix Channels: uma conexão, o contador de referências, os
tópicos em que entrou (com o join_ref de cada um) e a caixa das mensagens que
chegaram enquanto ele esperava outra coisa.
"""
alias Hello.{Mensagens, WS}
defstruct [:ws, ref: 0, topicos: %{}, caixa: []]
@doc "Conecta ao socket. `caminho` já inclui /websocket e a versão."
def conectar(porta, caminho) do
with {:ok, ws} <- WS.conectar(porta, caminho) do
{:ok, %__MODULE__{ws: ws}}
end
end
@doc "Entra num tópico e espera a resposta. O join_ref é a referência do próprio join."
def entrar(cliente, topico, params \\ %{}) do
{ref, cliente} = nova_ref(cliente)
WS.enviar(cliente.ws, Mensagens.codificar(Mensagens.join(ref, ref, topico, params)))
case esperar_resposta(cliente, ref, 5000) do
{:ok, resposta, cliente} -> {:ok, resposta, %{cliente | topicos: Map.put(cliente.topicos, topico, ref)}}
outro -> outro
end
end
@doc "Envia um evento e não espera nada. Devolve a referência usada."
def enviar(cliente, topico, evento, payload \\ %{}) do
{ref, cliente} = nova_ref(cliente)
mensagem = Mensagens.push(cliente.topicos[topico], ref, topico, evento, payload)
WS.enviar(cliente.ws, Mensagens.codificar(mensagem))
{ref, cliente}
end
@doc "Envia um evento e espera a resposta dele."
def chamar(cliente, topico, evento, payload \\ %{}, tempo \\ 5000) do
{ref, cliente} = enviar(cliente, topico, evento, payload)
esperar_resposta(cliente, ref, tempo)
end
@doc "Sai de um tópico."
def sair(cliente, topico) do
{ref, cliente} = nova_ref(cliente)
WS.enviar(cliente.ws, Mensagens.codificar(Mensagens.leave(cliente.topicos[topico], ref, topico)))
case esperar_resposta(cliente, ref, 5000) do
{:ok, resposta, cliente} -> {:ok, resposta, %{cliente | topicos: Map.delete(cliente.topicos, topico)}}
outro -> outro
end
end
@doc "A próxima mensagem que não é resposta: primeiro as guardadas na caixa, depois o fio."
def proxima(cliente, tempo \\ 1000)
def proxima(%__MODULE__{caixa: [mensagem | resto]} = cliente, _tempo), do: {:ok, mensagem, %{cliente | caixa: resto}}
def proxima(cliente, tempo), do: ler(cliente, tempo)
@doc """
Espera a resposta da mensagem `ref`. Devolve {:ok, resposta, cliente} ou
{:erro, resposta, cliente}, conforme o status; {:tempo_esgotado, cliente} se
ela não chegar; {:fechado, código, cliente} se o servidor encerrar.
"""
def esperar_resposta(cliente, ref, tempo) do
case ler(cliente, tempo) do
{:ok, %{event: "phx_reply", ref: ^ref, payload: %{"status" => status, "response" => resposta}}, cliente} ->
{if(status == "ok", do: :ok, else: :erro), resposta, cliente}
{:ok, outra, cliente} ->
# Não é a resposta esperada: guarda no fim da caixa e continua esperando.
esperar_resposta(%{cliente | caixa: cliente.caixa ++ [outra]}, ref, tempo)
outro ->
outro
end
end
@doc "Lê uma mensagem do fio, já decodificada."
def ler(cliente, tempo) do
case WS.receber(cliente.ws, tempo) do
{:ok, texto, ws} -> {:ok, Mensagens.decodificar(texto), %{cliente | ws: ws}}
{:fechado, codigo, ws} -> {:fechado, codigo, %{cliente | ws: ws}}
{:tempo_esgotado, ws} -> {:tempo_esgotado, %{cliente | ws: ws}}
end
end
@doc "Uma referência nova: o contador, como texto."
def nova_ref(cliente), do: {Integer.to_string(cliente.ref + 1), %{cliente | ref: cliente.ref + 1}}
@doc "Fecha a conexão."
def fechar(cliente), do: WS.fechar(cliente.ws)
end
''',
batimento: ~S'''
defmodule Hello.Batimento do
alias Hello.{Cliente, Mensagens, WS}
@doc "Envia um heartbeat e espera a resposta."
def bater(cliente) do
{ref, cliente} = Cliente.nova_ref(cliente)
WS.enviar(cliente.ws, Mensagens.codificar(Mensagens.heartbeat(ref)))
Cliente.esperar_resposta(cliente, ref, 5000)
end
@doc "Fica `duracao` milissegundos sem fazer nada, batendo a cada `intervalo`."
def manter_vivo(cliente, duracao, _intervalo) when duracao <= 0, do: {:ok, cliente}
def manter_vivo(cliente, duracao, intervalo) do
Process.sleep(min(duracao, intervalo))
case bater(cliente) do
{:ok, _resposta, cliente} -> manter_vivo(cliente, duracao - intervalo, intervalo)
outro -> outro
end
end
end
''',
reentrar: ~S'''
defmodule Hello.Reentrada do
alias Hello.Cliente
@doc """
Olha uma mensagem que chegou. Se for o aviso de que o channel caiu, entra de
novo no tópico e devolve {:reentrou, cliente}; senão, {:ignorada, cliente}.
"""
def tratar(cliente, %{event: "phx_error", topic: topico}) do
{:ok, _resposta, cliente} = Cliente.entrar(cliente, topico)
{:reentrou, cliente}
end
def tratar(cliente, _mensagem), do: {:ignorada, cliente}
end
''',
brecho: ~S'''
defmodule Hello.Robo do
alias Hello.Cliente
@doc "80_00 centavos viram \"R$ 80,00\"."
def reais(centavos) do
"R$ #{div(centavos, 100)},#{centavos |> rem(100) |> Integer.to_string() |> String.pad_leading(2, "0")}"
end
@doc """
Entra em vitrine:lobby e atende `quantas` mensagens (perguntas e vendas).
Devolve {:ok, %{vendas: [...], respostas: n}, cliente}.
"""
def atender(cliente, precos, quantas) do
{:ok, _resposta, cliente} = Cliente.entrar(cliente, "vitrine:lobby")
laco(cliente, precos, quantas, %{vendas: [], respostas: 0})
end
defp laco(cliente, _precos, 0, estado), do: {:ok, estado, cliente}
defp laco(cliente, precos, faltam, estado) do
case Cliente.proxima(cliente, 10_000) do
{:ok, %{event: "pergunta", payload: %{"texto" => "preço? " <> peca}}, cliente} ->
texto =
case precos do
%{^peca => centavos} -> "#{peca} custa #{reais(centavos)}"
_ -> "Não temos #{peca}"
end
{:ok, _resposta, cliente} = Cliente.chamar(cliente, "vitrine:lobby", "resposta", %{"texto" => texto})
laco(cliente, precos, faltam - 1, %{estado | respostas: estado.respostas + 1})
{:ok, %{event: "venda", payload: %{"peca" => peca}}, cliente} ->
laco(cliente, precos, faltam - 1, %{estado | vendas: estado.vendas ++ [peca]})
{:ok, _outra, cliente} ->
# As respostas do próprio robô também voltam pelo tópico: não contam.
laco(cliente, precos, faltam, estado)
{:tempo_esgotado, cliente} ->
{:ok, estado, cliente}
end
end
end
'''
}
end
# --- conferência dos exercícios ----------------------------------------------
@doc "Confere um exercício abrindo conexões de verdade com o servidor."
def conferir(id) do
resultado =
Enum.reduce_while(Map.fetch!(verificacoes(), id), [], fn {descricao, teste}, feitos ->
case executar(teste) do
:ok -> {:cont, ["- Passou: " <> descricao | feitos]}
{:erro, dica} -> {:halt, ["- **Ainda não:** " <> descricao <> "\n\n " <> dica | feitos]}
end
end)
Kino.Markdown.new(resultado |> Enum.reverse() |> Enum.join("\n"))
end
defp executar(teste) do
teste.()
rescue
erro -> {:erro, explicar(erro)}
end
defp verificacoes do
%{
endereco: [
{"`Hello.Endereco.caminho(\"/socket\")` devolve `/socket/websocket?vsn=2.0.0`.",
fn ->
case apply(Hello.Endereco, :caminho, ["/socket"]) do
"/socket/websocket?vsn=2.0.0" -> :ok
outro -> {:erro, "Veio `#{inspect(outro)}`. Sem `vsn=2.0.0`, o Phoenix usa o serializador V1, o dos mapas."}
end
end},
{"Com `%{\"token\" => \"segredo-do-app\"}`, o servidor aceita a conexão em `/mobile`.",
fn ->
caminho = apply(Hello.Endereco, :caminho, ["/mobile", %{"token" => "segredo-do-app"}])
case ws(:conectar, [porta(), caminho]) do
{:ok, conexao} ->
ws(:fechar, [conexao])
:ok
{:erro, linha} ->
{:erro, "Para `#{caminho}`, o servidor respondeu `#{linha}`. Os parâmetros entram na query string, junto de `vsn`."}
end
end},
{"Valores com espaço e `&` chegam inteiros.",
fn ->
caminho = apply(Hello.Endereco, :caminho, ["/socket", %{"nome" => "ana maria & cia"}])
consulta = caminho |> URI.parse() |> Map.get(:query) |> to_string() |> URI.decode_query()
if consulta == %{"vsn" => "2.0.0", "nome" => "ana maria & cia"},
do: :ok,
else: {:erro, "A query de `#{caminho}` foi lida como `#{inspect(consulta)}`. Use `URI.encode_query/1`, que escapa os valores."}
end}
],
mensagens: [
{"O `phx_join` montado por `join/4` é aceito pelo servidor.",
fn ->
conversa("/socket/websocket?vsn=2.0.0", fn conexao ->
case troca(conexao, m(:join, ["7", "7", "room:lobby", %{}])) do
["7", "7", "room:lobby", "phx_reply", %{"status" => "ok"}] -> :ok
outra -> {:erro, "O servidor respondeu `#{inspect(outra)}`."}
end
end)
end},
{"`push/5` monta um `ping`, e a resposta devolve o mesmo payload.",
fn ->
mensagem = m(:push, ["1", "2", "room:lobby", "ping", %{"n" => 20}])
conversa("/socket/websocket?vsn=2.0.0", fn conexao ->
troca(conexao, ["1", "1", "room:lobby", "phx_join", %{}])
case troca(conexao, mensagem) do
["1", "2", "room:lobby", "phx_reply", %{"status" => "ok", "response" => %{"n" => 20}}] -> :ok
outra -> {:erro, "Enviei `#{inspect(mensagem)}` e o servidor respondeu `#{inspect(outra)}`."}
end
end)
end},
{"`heartbeat/1` usa o tópico `phoenix` e `join_ref` nulo, e o servidor responde.",
fn ->
mensagem = m(:heartbeat, ["42"])
if mensagem != [nil, "42", "phoenix", "heartbeat", %{}] do
{:erro, "Veio `#{inspect(mensagem)}`; o guia mostra `[null, \"2\", \"phoenix\", \"heartbeat\", {}]`."}
else
conversa("/socket/websocket?vsn=2.0.0", fn conexao ->
case troca(conexao, mensagem) do
[nil, "42", "phoenix", "phx_reply", %{"status" => "ok"}] -> :ok
outra -> {:erro, "O servidor respondeu `#{inspect(outra)}`."}
end
end)
end
end},
{"Depois do `phx_leave` de `leave/3`, chegam a resposta e o `phx_close`.",
fn ->
conversa("/socket/websocket?vsn=2.0.0", fn conexao ->
troca(conexao, ["1", "1", "room:lobby", "phx_join", %{}])
resposta = troca(conexao, m(:leave, ["1", "2", "room:lobby"]))
fecho = ler(conexao)
case {resposta, fecho} do
{["1", "2", "room:lobby", "phx_reply", %{"status" => "ok"}], ["1", "1", "room:lobby", "phx_close", _]} -> :ok
_ -> {:erro, "Depois do leave vieram `#{inspect(resposta)}` e `#{inspect(fecho)}`."}
end
end)
end},
{"`decodificar/1` devolve um mapa com `join_ref`, `ref`, `topic`, `event` e `payload`.",
fn ->
texto = ~s(["1","2","room:lobby","phx_reply",{"status":"ok","response":{"n":1}}])
esperado = %{join_ref: "1", ref: "2", topic: "room:lobby", event: "phx_reply", payload: %{"status" => "ok", "response" => %{"n" => 1}}}
case m(:decodificar, [texto]) do
^esperado -> :ok
outro -> {:erro, "Para `#{texto}` veio `#{inspect(outro)}`. As chaves do mapa são átomos; o payload fica como o JSON o trouxe, com chaves em texto."}
end
end}
],
respostas: [
{"Um `ping` sozinho devolve `{:ok, payload, cliente}`.",
fn ->
com_cliente("/socket/websocket?vsn=2.0.0", "room:lobby", fn cliente ->
case c(:chamar, [cliente, "room:lobby", "ping", %{"n" => 1}]) do
{:ok, %{"n" => 1}, _cliente} -> :ok
outro -> {:erro, "Veio `#{resumo(outro)}`."}
end
end)
end},
{"Um broadcast que chega antes da resposta não é confundido com ela.",
fn ->
com_cliente("/socket/websocket?vsn=2.0.0", "room:lobby", fn cliente ->
apply(@endpoint, :broadcast, ["room:lobby", "new_msg", %{body: "no meio do caminho"}])
Process.sleep(100)
case c(:chamar, [cliente, "room:lobby", "ping", %{"n" => 2}]) do
{:ok, %{"n" => 2}, _cliente} ->
:ok
outro ->
{:erro,
"Havia um `new_msg` no fio antes da resposta do `ping`, e `chamar/4` devolveu `#{resumo(outro)}`. Só a mensagem com `event: \"phx_reply\"` e a `ref` esperada é a resposta."}
end
end)
end},
{"As mensagens que chegaram durante a espera ficam na caixa, na ordem, e saem por `proxima/2`.",
fn ->
com_cliente("/socket/websocket?vsn=2.0.0", "room:lobby", fn cliente ->
apply(@endpoint, :broadcast, ["room:lobby", "new_msg", %{body: "primeira"}])
apply(@endpoint, :broadcast, ["room:lobby", "new_msg", %{body: "segunda"}])
Process.sleep(100)
{:ok, _, cliente} = c(:chamar, [cliente, "room:lobby", "ping", %{}])
with {:ok, %{payload: %{"body" => "primeira"}}, cliente} <- c(:proxima, [cliente, 300]),
{:ok, %{payload: %{"body" => "segunda"}}, _cliente} <- c(:proxima, [cliente, 300]) do
:ok
else
outro -> {:erro, "Esperava `primeira` e depois `segunda`; `proxima/2` devolveu `#{resumo(outro)}`. Guarde no fim da caixa: `cliente.caixa ++ [mensagem]`."}
end
end)
end},
{"Se a resposta não vem, devolve `{:tempo_esgotado, cliente}`, com o que chegou guardado na caixa.",
fn ->
com_cliente("/socket/websocket?vsn=2.0.0", "room:lobby", fn cliente ->
# "shout" não responde: só volta o broadcast.
case c(:chamar, [cliente, "room:lobby", "shout", %{"a" => 1}, 300]) do
{:tempo_esgotado, %{caixa: [%{event: "shout"}]}} -> :ok
outro -> {:erro, "O evento `shout` não tem resposta, só o broadcast. `chamar/5` devolveu `#{resumo(outro)}`."}
end
end)
end}
],
batimento: [
{"Em `/impaciente`, depois de 4,5 segundos de `manter_vivo/3`, a conexão ainda responde.",
fn ->
com_cliente("/impaciente/websocket?vsn=2.0.0", "room:lobby", fn cliente ->
inicio = System.monotonic_time(:millisecond)
case apply(Hello.Batimento, :manter_vivo, [cliente, 4500, 1000]) do
{:ok, cliente} ->
passou = System.monotonic_time(:millisecond) - inicio
case c(:chamar, [cliente, "room:lobby", "ping", %{"vivo" => true}, 1000]) do
{:ok, %{"vivo" => true}, _} when passou >= 4400 -> :ok
{:ok, _, _} -> {:erro, "A função voltou em #{passou} ms; deveria ocupar os 4500 ms pedidos."}
outro -> {:erro, "Depois de #{passou} ms, o `ping` devolveu `#{resumo(outro)}`: o servidor fechou a conexão. Envie um heartbeat a cada intervalo."}
end
outro ->
{:erro, "`manter_vivo/3` devolveu `#{resumo(outro)}`; o esperado é `{:ok, cliente}`."}
end
end)
end},
{"As respostas dos heartbeats foram consumidas: não sobram na caixa.",
fn ->
com_cliente("/impaciente/websocket?vsn=2.0.0", "room:lobby", fn cliente ->
{:ok, cliente} = apply(Hello.Batimento, :manter_vivo, [cliente, 2500, 1000])
cond do
cliente.ref < 3 -> {:erro, "Em 2,5 segundos com intervalo de 1 segundo, esperava pelo menos 2 heartbeats; o contador de referências está em #{cliente.ref}."}
cliente.caixa != [] -> {:erro, "Sobraram na caixa: `#{inspect(cliente.caixa, limit: 6)}`. Use `bater/1`, que espera a resposta de cada heartbeat."}
true -> :ok
end
end)
end}
],
reentrar: [
{"Uma mensagem comum devolve `{:ignorada, cliente}`.",
fn ->
com_cliente("/socket/websocket?vsn=2.0.0", "room:lobby", fn cliente ->
mensagem = %{join_ref: nil, ref: nil, topic: "room:lobby", event: "new_msg", payload: %{"body" => "oi"}}
case apply(Hello.Reentrada, :tratar, [cliente, mensagem]) do
{:ignorada, ^cliente} -> :ok
outro -> {:erro, "Veio `#{resumo(outro)}`."}
end
end)
end},
{"Depois que o channel cai, `tratar/2` com o `phx_error` devolve `{:reentrou, cliente}`, e o `ping` volta a funcionar.",
fn ->
com_cliente("/socket/websocket?vsn=2.0.0", "room:lobby", fn cliente ->
{queda, cliente} = derrubar(cliente)
case apply(Hello.Reentrada, :tratar, [cliente, queda]) do
{:reentrou, cliente} ->
case c(:chamar, [cliente, "room:lobby", "ping", %{"de" => "volta"}, 1000]) do
{:ok, %{"de" => "volta"}, _} -> :ok
outro -> {:erro, "`tratar/2` disse que reentrou, mas o `ping` devolveu `#{resumo(outro)}`."}
end
outro ->
{:erro, "A mensagem era `#{inspect(queda)}` e `tratar/2` devolveu `#{resumo(outro)}`. Case com `%{event: \"phx_error\", topic: topico}` e chame `Hello.Cliente.entrar/2`."}
end
end)
end},
{"A nova entrada tem um `join_ref` novo.",
fn ->
com_cliente("/socket/websocket?vsn=2.0.0", "room:lobby", fn cliente ->
antigo = cliente.topicos["room:lobby"]
{queda, cliente} = derrubar(cliente)
case apply(Hello.Reentrada, :tratar, [cliente, queda]) do
{:reentrou, novo} ->
if novo.topicos["room:lobby"] not in [nil, antigo],
do: :ok,
else: {:erro, "O `join_ref` de `room:lobby` continua `#{inspect(novo.topicos["room:lobby"])}`. Devolva o cliente que `Hello.Cliente.entrar/2` devolveu."}
outro ->
{:erro, "Veio `#{resumo(outro)}`."}
end
end)
end}
],
brecho: [
{"`Hello.Robo.atender/3` existe.", fn -> exporta(Hello.Robo, atender: 3) end},
{"O robô responde o preço de uma peça que está na tabela.",
fn ->
com_robo(1, fn visitante, _robo ->
troca(visitante, ["1", "2", "vitrine:lobby", "pergunta", %{"texto" => "preço? Jaqueta jeans"}])
case esperar_evento(visitante, "resposta") do
%{"texto" => "Jaqueta jeans custa R$ 80,00"} -> :ok
outro -> {:erro, "Perguntei `preço? Jaqueta jeans` e a resposta no tópico foi `#{inspect(outro)}`; o esperado é `Jaqueta jeans custa R$ 80,00`."}
end
end)
end},
{"Para uma peça que não está na tabela, responde `Não temos ...`.",
fn ->
com_robo(1, fn visitante, _robo ->
troca(visitante, ["1", "2", "vitrine:lobby", "pergunta", %{"texto" => "preço? Chapéu"}])
case esperar_evento(visitante, "resposta") do
%{"texto" => "Não temos Chapéu"} -> :ok
outro -> {:erro, "Perguntei `preço? Chapéu` e a resposta foi `#{inspect(outro)}`; o esperado é `Não temos Chapéu`."}
end
end)
end},
{"Depois de 4 mensagens (2 perguntas e 2 vendas), devolve as vendas na ordem e a contagem de respostas.",
fn ->
com_robo(4, fn visitante, robo ->
troca(visitante, ["1", "2", "vitrine:lobby", "pergunta", %{"texto" => "preço? Bota"}])
esperar_evento(visitante, "resposta")
apply(@endpoint, :broadcast, ["vitrine:lobby", "venda", %{peca: "Bota"}])
troca(visitante, ["1", "3", "vitrine:lobby", "pergunta", %{"texto" => "preço? Vestido floral"}])
esperar_evento(visitante, "resposta")
apply(@endpoint, :broadcast, ["vitrine:lobby", "venda", %{peca: "Vestido floral"}])
receive do
{:robo, ^robo, {:ok, %{vendas: ["Bota", "Vestido floral"], respostas: 2}, _cliente}} ->
:ok
{:robo, ^robo, {:caiu, motivo}} ->
{:erro, "O robô caiu: `#{motivo}`."}
{:robo, ^robo, outro} ->
{:erro,
"O robô devolveu `#{resumo(outro)}`; o esperado é `{:ok, %{vendas: [\"Bota\", \"Vestido floral\"], respostas: 2}, cliente}`. As respostas do próprio robô voltam pelo tópico e não contam como mensagem atendida."}
after
3000 ->
{:erro, "Depois de 2 perguntas e 2 vendas, o robô ainda não tinha terminado. Ele deve parar ao atender `quantas` mensagens."}
end
end)
end}
]
}
end
# --- ajudantes da conferência --------------------------------------------------
# Um segundo cliente, independente do que você escreve nos exercícios.
defp porta, do: apply(@endpoint, :config, [:http])[:port]
defp ws(funcao, argumentos), do: apply(Hello.WS, funcao, argumentos)
defp m(funcao, argumentos), do: apply(Hello.Mensagens, funcao, argumentos)
defp c(funcao, argumentos), do: apply(Hello.Cliente, funcao, argumentos)
defp conversa(caminho, teste) do
{:ok, conexao} = ws(:conectar, [porta(), caminho])
{:ok, agente} = Agent.start_link(fn -> conexao end)
try do
teste.(agente)
after
ws(:fechar, [Agent.get(agente, & &1)])
Agent.stop(agente)
end
end
defp troca(agente, mensagem) do
ws(:enviar, [Agent.get(agente, & &1), Jason.encode!(mensagem)])
ler(agente)
end
defp ler(agente, tempo \\ 1000) do
case ws(:receber, [Agent.get(agente, & &1), tempo]) do
{:ok, texto, conexao} ->
Agent.update(agente, fn _ -> conexao end)
Jason.decode!(texto)
{:tempo_esgotado, _} ->
:nada
{:fechado, codigo, _} ->
{:fechado, codigo}
end
end
defp esperar_evento(agente, evento, tentativas \\ 10) do
case ler(agente, 500) do
[_, _, _, ^evento, payload] -> payload
_ when tentativas > 0 -> esperar_evento(agente, evento, tentativas - 1)
_ -> :nada
end
end
defp com_cliente(caminho, topico, teste) do
{:ok, cliente} = c(:conectar, [porta(), caminho])
try do
{:ok, _, cliente} = c(:entrar, [cliente, topico])
teste.(cliente)
after
c(:fechar, [cliente])
end
end
# Derruba o channel de propósito (sem poluir a saída com o log do erro) e
# devolve o phx_error que chegou.
defp derrubar(cliente) do
nivel = Logger.level()
Logger.configure(level: :none)
try do
{_ref, cliente} = c(:enviar, [cliente, "room:lobby", "explodir", %{}])
{:ok, queda, cliente} = c(:proxima, [cliente, 1000])
Process.sleep(50)
{queda, cliente}
after
Logger.configure(level: nivel)
end
end
# Liga o robô do aluno num processo à parte, espera ele entrar no tópico e abre um visitante.
defp com_robo(quantas, teste) do
precos = %{"Jaqueta jeans" => 80_00, "Vestido floral" => 45_00, "Bota" => 120_00}
antes = inscritos("vitrine:lobby")
pai = self()
marca = make_ref()
pid =
spawn(fn ->
resultado =
try do
{:ok, cliente} = c(:conectar, [porta(), "/socket/websocket?vsn=2.0.0"])
apply(Hello.Robo, :atender, [cliente, precos, quantas])
rescue
erro -> {:caiu, Exception.format_banner(:error, erro)}
end
send(pai, {:robo, marca, resultado})
end)
try do
if esperar_novo_inscrito("vitrine:lobby", antes, 40) do
conversa("/socket/websocket?vsn=2.0.0", fn visitante ->
troca(visitante, ["1", "1", "vitrine:lobby", "phx_join", %{}])
teste.(visitante, marca)
end)
else
receive do
{:robo, ^marca, {:caiu, motivo}} -> {:erro, "O robô caiu antes de entrar no tópico: `#{motivo}`."}
after
0 -> {:erro, "O robô não entrou em `vitrine:lobby`. `atender/3` começa com `Hello.Cliente.entrar(cliente, \"vitrine:lobby\")`."}
end
end
after
Process.exit(pid, :kill)
end
end
defp inscritos(topico), do: Hello.PubSub |> Registry.lookup(topico) |> Enum.map(&elem(&1, 0))
# Espera aparecer um processo inscrito que não estava na lista de antes.
defp esperar_novo_inscrito(_topico, _antes, 0), do: false
defp esperar_novo_inscrito(topico, antes, tentativas) do
if Enum.any?(inscritos(topico), &(&1 not in antes and Process.alive?(&1))) do
true
else
Process.sleep(50)
esperar_novo_inscrito(topico, antes, tentativas - 1)
end
end
defp exporta(modulo, funcoes) do
faltando = Enum.reject(funcoes, fn {nome, aridade} -> Code.ensure_loaded?(modulo) and function_exported?(modulo, nome, aridade) end)
if faltando == [],
do: :ok,
else: {:erro, "Falta definir em `#{inspect(modulo)}`: #{Enum.map_join(faltando, ", ", fn {n, a} -> "`#{n}/#{a}`" end)}."}
end
# Tira do resultado a struct do cliente, que é grande, para a dica caber numa linha.
defp resumo(valor) when is_tuple(valor) do
valor
|> Tuple.to_list()
|> Enum.map(fn
%{__struct__: Hello.Cliente} -> :cliente
outro -> outro
end)
|> List.to_tuple()
|> inspect(limit: 8)
end
defp resumo(valor), do: inspect(valor, limit: 8)
defp explicar(%UndefinedFunctionError{module: modulo, function: funcao, arity: aridade}) do
"`#{inspect(modulo)}.#{funcao}/#{aridade}` não existe. O módulo foi definido e a célula foi avaliada?"
end
defp explicar(%RuntimeError{message: mensagem}), do: "A execução parou com: `#{mensagem}`."
defp explicar(%MatchError{term: termo}) do
"Um casamento de padrões falhou com o valor `#{resumo(termo)}`."
end
defp explicar(erro), do: "Levantou uma exceção:\n\n `" <> Exception.format_banner(:error, erro) <> "`"
endComo usar este notebook
Este notebook acompanha o guia Writing a Channels Client da documentação do Phoenix 1.8, na mesma ordem e com o link de cada seção. O guia é curto: ele descreve o que um cliente precisa enviar e o que recebe de volta, para quem quer escrever um cliente de Channels numa linguagem que ainda não tem um, ou testar um channel à mão.
Aqui você escreve esse cliente, em Elixir. Nos notebooks anteriores, o módulo Trilha escondia um cliente WebSocket para conferir os exercícios. Neste, o cliente fica à vista, em células, e cresce em três camadas:
Hello.WS: o WebSocket em si, sobre uma conexão TCP. Vem pronto, e o texto explica cada parte;Hello.Mensagens: as mensagens do protocolo dos Channels. Você escreve;Hello.Cliente: o cliente que entra em tópicos, envia eventos e separa respostas de novidades. Vem quase pronto; a parte central é sua.
As convenções são as de sempre: uma célula equivale a um arquivo, as perguntas "Antes de continuar" vêm antes da explicação, e os erros fazem parte do conteúdo. As conferências abrem conexões de verdade com o servidor do notebook, com um segundo cliente, independente do seu.
Os exercícios se apoiam uns nos outros. Hello.Cliente usa Hello.Mensagens; por isso, enquanto o exercício "As mensagens" não estiver feito, as células que usam o cliente param com Hello.Mensagens.decodificar/1 ainda não foi escrita, e as que vêm logo depois delas param com undefined variable "cliente", porque a variável não chegou a ser criada. Faça os exercícios na ordem e reavalie as células seguintes.
Este notebook não usa o PostgreSQL. A instalação é a dos notebooks 13 a 17; se você vem do 12, o Livebook faz uma instalação nova na primeira avaliação, uma vez só.
A aplicação fica em http://localhost:4010. Se outro notebook da trilha estiver com o runtime ligado, desconecte-o antes. Avalie a primeira célula, no topo, antes de tudo.
A aplicação Hello dentro do notebook
O servidor é o do guia de Channels: um UserSocket com a sala room:lobby. Em volta dele há três peças a mais, que existem para este notebook: um segundo socket com o exemplo que o guia usa (miami:weather), um channel para o desafio do Brechó e um socket impaciente, para a seção do heartbeat.
lib/hello_web.ex
defmodule HelloWeb do
def static_paths, do: ~w(assets fonts images favicon.ico robots.txt)
def router do
quote do
use Phoenix.Router, helpers: false
import Plug.Conn
import Phoenix.Controller
end
end
def channel do
quote do
use Phoenix.Channel
end
end
def controller do
quote do
use Phoenix.Controller, formats: [:html, :json]
import Plug.Conn
end
end
defmacro __using__(which) when is_atom(which) do
apply(__MODULE__, which, [])
end
endlib/hello_web/channels/user_socket.ex: o socket gerado por mix phx.gen.socket User, sem os comentários, com as rotas dos channels.
defmodule HelloWeb.UserSocket do
use Phoenix.Socket
## Channels
channel "room:*", HelloWeb.RoomChannel
channel "vitrine:*", HelloWeb.VitrineChannel
@impl true
def connect(_params, socket, _connect_info) do
{:ok, socket}
end
@impl true
def id(_socket), do: nil
endlib/hello_web/channels/room_channel.ex
defmodule HelloWeb.RoomChannel do
use HelloWeb, :channel
@impl true
def join("room:lobby", payload, socket) do
if authorized?(payload) do
{:ok, socket}
else
{:error, %{reason: "unauthorized"}}
end
end
def join("room:" <> _private_room_id, _params, _socket) do
{:error, %{reason: "unauthorized"}}
end
# Channels can be used in a request/response fashion
# by sending replies to requests from the client
@impl true
def handle_in("ping", payload, socket) do
{:reply, {:ok, payload}, socket}
end
# It is also common to receive messages from the client and
# broadcast to everyone in the current topic (room:lobby).
@impl true
def handle_in("shout", payload, socket) do
broadcast(socket, "shout", payload)
{:noreply, socket}
end
def handle_in("new_msg", %{"body" => body}, socket) do
broadcast!(socket, "new_msg", %{body: body})
{:noreply, socket}
end
# Só neste notebook: um evento que derruba o processo do channel.
def handle_in("explodir", _payload, _socket) do
raise "o channel caiu de propósito"
end
# Add authorization logic here as required.
defp authorized?(_payload) do
true
end
endlib/hello_web/channels/mobile_socket.ex e weather_channel.ex: o exemplo que o guia usa, miami:weather, num segundo socket.
defmodule HelloWeb.MobileSocket do
use Phoenix.Socket
channel "miami:*", HelloWeb.WeatherChannel
@impl true
def connect(%{"token" => "segredo-do-app"}, socket, _connect_info) do
{:ok, assign(socket, :app, "mobile")}
end
def connect(_params, _socket, _connect_info), do: :error
@impl true
def id(_socket), do: nil
end
defmodule HelloWeb.WeatherChannel do
use HelloWeb, :channel
@impl true
def join("miami:weather", params, socket) do
send(self(), :after_join)
{:ok, %{you_sent: params}, socket}
end
@impl true
def handle_info(:after_join, socket) do
push(socket, "forecast", %{today: "sunny"})
{:noreply, socket}
end
@impl true
def handle_in("report_emergency", payload, socket) do
Phoenix.PubSub.broadcast(Hello.PubSub, "emergencies", {:emergency, payload})
{:reply, :ok, socket}
end
endlib/hello_web/channels/vitrine_channel.ex: o channel do desafio do Brechó.
defmodule HelloWeb.VitrineChannel do
use HelloWeb, :channel
@impl true
def join("vitrine:lobby", _payload, socket) do
{:ok, socket}
end
# Uma pergunta de um visitante vai para todos no tópico, inclusive o robô.
@impl true
def handle_in("pergunta", %{"texto" => texto}, socket) do
broadcast!(socket, "pergunta", %{texto: texto})
{:reply, :ok, socket}
end
def handle_in("resposta", %{"texto" => texto}, socket) do
broadcast!(socket, "resposta", %{texto: texto})
{:reply, :ok, socket}
end
endlib/hello_web/router.ex
defmodule HelloWeb.Router do
use HelloWeb, :router
pipeline :browser do
plug :accepts, ["html"]
plug :fetch_session
plug :protect_from_forgery
plug :put_secure_browser_headers
end
scope "/", HelloWeb do
pipe_through :browser
get "/", PageController, :home
end
endlib/hello_web/endpoint.ex: três declarações socket. A primeira é a que mix phx.gen.socket User pede para acrescentar.
defmodule HelloWeb.Endpoint do
use Phoenix.Endpoint, otp_app: :hello
@session_options [
store: :cookie,
key: "_hello_key",
signing_salt: "trilha00",
same_site: "Lax"
]
socket "/socket", HelloWeb.UserSocket,
websocket: true,
longpoll: false
socket "/mobile", HelloWeb.MobileSocket,
websocket: true,
longpoll: false
# Só neste notebook: o mesmo socket, com um tempo de espera de 2 segundos.
socket "/impaciente", HelloWeb.UserSocket,
websocket: [timeout: 2_000],
longpoll: false
plug Plug.Static, at: "/assets", from: {:phoenix, "priv/static"}, only: ~w(phoenix.min.js)
plug Plug.RequestId
plug Plug.Telemetry, event_prefix: [:phoenix, :endpoint]
plug Plug.Parsers,
parsers: [:urlencoded, :multipart, :json],
pass: ["*/*"],
json_decoder: Phoenix.json_library()
plug Plug.MethodOverride
plug Plug.Head
plug Plug.Session, @session_options
plug HelloWeb.Router
endlib/hello_web/controllers/: as páginas de erro e a página inicial, um chat mínimo em room:lobby com o cliente JavaScript oficial (a global Phoenix vem do phoenix.min.js que o endpoint serve).
defmodule HelloWeb.ErrorHTML do
def render(template, _assigns) do
Phoenix.Controller.status_message_from_template(template)
end
end
defmodule HelloWeb.ErrorJSON do
def render(template, _assigns) do
%{errors: %{detail: Phoenix.Controller.status_message_from_template(template)}}
end
end
defmodule HelloWeb.PageController do
use HelloWeb, :controller
def home(conn, _params) do
html(conn, """
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8" />
<title>Hello · room:lobby</title>
<script src="/assets/phoenix.min.js"></script>
</head>
<body>
<h1>room:lobby</h1>
<input id="chat-input" type="text" placeholder="Escreva e aperte Enter" />
<div id="messages" role="log" aria-live="polite"></div>
<script>
const socket = new Phoenix.Socket("/socket", {params: {}})
socket.connect()
const channel = socket.channel("room:lobby", {})
const chatInput = document.querySelector("#chat-input")
const messages = document.querySelector("#messages")
const mostrar = (texto) => {
const item = document.createElement("p")
item.innerText = texto
messages.appendChild(item)
}
chatInput.addEventListener("keypress", event => {
if (event.key === "Enter") {
channel.push("new_msg", {body: chatInput.value})
chatInput.value = ""
}
})
channel.on("new_msg", payload => mostrar(payload.body))
channel.on("shout", payload => mostrar("shout: " + JSON.stringify(payload)))
channel.join()
.receive("ok", resp => { console.log("Joined successfully", resp) })
.receive("error", resp => { console.log("Unable to join", resp) })
</script>
</body>
</html>
""")
end
endlib/hello/application.ex: a árvore de supervisão inicia o PubSub e o endpoint. Este notebook não usa o banco.
{:ok, _} = Kino.start_child({Phoenix.PubSub, name: Hello.PubSub})
{:ok, _} = Kino.start_child(HelloWeb.Endpoint)
Kino.Markdown.new("Aplicação no ar em <http://localhost:4010>.")O router aponta para o PageController antes de ele existir, por isso a célula do router mostra o aviso HelloWeb.PageController.init/1 is undefined; ele some quando a célula dos controllers é avaliada.
Resumo do que está no ar:
| Caminho no endpoint | Socket | Tópicos | Para que serve aqui |
|---|---|---|---|
/socket |
HelloWeb.UserSocket |
room:*, vitrine:* |
a sala do guia de Channels e o desafio |
/mobile |
HelloWeb.MobileSocket |
miami:* |
o exemplo do guia; exige o parâmetro token |
/impaciente |
HelloWeb.UserSocket |
os mesmos de /socket |
fecha conexões ociosas em 2 segundos |
O RoomChannel é o que mix phx.gen.channel Room gera (o ping que responde e o shout que avisa a sala), mais o new_msg do guia de Channels, a recusa de salas privadas e um evento explodir, que derruba o processo de propósito.
A página http://localhost:4010 é um chat mínimo em room:lobby, com o cliente JavaScript oficial. Deixe-a aberta numa aba: o que o seu cliente enviar para a sala aparece nela.
Visão geral
Na documentação: Overview
Já existem clientes de Channels em várias linguagens. O guia serve para quem precisa escrever mais um e, como ele mesmo diz, também para testar um channel à mão com um cliente WebSocket qualquer.
Trilha.pergunta(:a_qualquer_hora)WebSockets são bidirecionais: as mensagens podem ir em qualquer direção, a qualquer momento. Por isso, diz o guia, os clientes costumam usar callbacks para tratar as mensagens que chegam, quando chegarem.
Trilha.pergunta(:mesma_conexao)Um cliente precisa entrar em pelo menos um tópico para começar a enviar e receber mensagens, e pode entrar em vários pela mesma conexão.
O Phoenix traz dois transportes, WebSocket e long polling, escolhidos na declaração socket do endpoint. O guia trata do WebSocket, e este notebook também; os sockets da aplicação estão com longpoll: false.
Conectando
Na documentação: Connecting
Para abrir uma conexão, o guia manda olhar a declaração socket no módulo Endpoint da aplicação. Se ela diz socket "/mobile", MyAppWeb.MobileSocket, o caminho da requisição HTTP inicial é:
[host]:[port]/mobile/websocket?vsn=2.0.0
São três partes: o caminho do socket, o sufixo /websocket, que escolhe o transporte, e vsn=2.0.0, que escolhe o Phoenix.Socket.V2.JSONSerializer. Esse serializador vem com o Phoenix e troca mensagens em forma de lista.
A requisição também precisa levar os cabeçalhos que pedem a troca de protocolo, de HTTP para WebSocket. O guia sugere usar uma biblioteca que cuide disso; em Elixir, cita a mint_web_socket. Aqui você faz à mão uma vez, para ver o que a biblioteca faria.
O pedido de upgrade, à mão
Um WebSocket começa como uma requisição HTTP comum, numa conexão TCP. A célula abaixo abre a conexão com o :gen_tcp, do Erlang, escreve a requisição como texto e lê a resposta:
porta = HelloWeb.Endpoint.config(:http)[:port]
{:ok, tcp} = :gen_tcp.connect(~c"127.0.0.1", porta, [:binary, active: false])
:ok =
:gen_tcp.send(tcp, """
GET /socket/websocket?vsn=2.0.0 HTTP/1.1\r
Host: localhost:#{porta}\r
Upgrade: websocket\r
Connection: Upgrade\r
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r
Sec-WebSocket-Version: 13\r
\r
""")
{:ok, resposta} = :gen_tcp.recv(tcp, 0, 2000)
:gen_tcp.close(tcp)
IO.puts(resposta)A resposta é HTTP/1.1 101 Switching Protocols, com upgrade: websocket e connection: Upgrade. A partir da linha em branco, a mesma conexão TCP deixa de falar HTTP. No log, o Phoenix registra CONNECTED TO HelloWeb.UserSocket, com o transporte, o serializador e os parâmetros.
Os quatro cabeçalhos do pedido:
Upgrade: websocketeConnection: Upgradepedem a troca de protocolo;Sec-WebSocket-Version: 13é a versão do protocolo WebSocket (a única em uso);Sec-WebSocket-Keyé um valor aleatório, 16 bytes em Base64. O servidor prova que entendeu o pedido devolvendo, emsec-websocket-accept, o SHA-1 dessa chave concatenada com um texto fixo do padrão.
A chave acima é a do exemplo da especificação (RFC 6455). A conta que o servidor fez:
:crypto.hash(:sha, "dGhlIHNhbXBsZSBub25jZQ==" <> "258EAFA5-E914-47DA-95CA-C5AB0DC85B11")
|> Base.encode64()É o mesmo valor do cabeçalho sec-websocket-accept da resposta: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=.
Trilha.pergunta(:sem_upgrade){:ok, tcp} = :gen_tcp.connect(~c"127.0.0.1", porta, [:binary, active: false])
:ok = :gen_tcp.send(tcp, "GET /socket/websocket?vsn=2.0.0 HTTP/1.1\r\nHost: localhost:#{porta}\r\n\r\n")
{:ok, resposta} = :gen_tcp.recv(tcp, 0, 2000)
:gen_tcp.close(tcp)
IO.puts(resposta)400 Bad Request, com o corpo 'connection' header must contain 'upgrade', got []. Esse caminho não passa pelo router; quem o atende é a declaração socket do endpoint, e ela só aceita o pedido de upgrade.
Os quadros
Depois do 101, os dois lados trocam quadros (frames). Cada quadro tem um cabeçalho pequeno e os dados:
- o primeiro byte traz o bit FIN (último pedaço da mensagem) e o opcode:
1para texto,8para encerrar,9para ping,10para pong; - o segundo byte traz o bit de máscara e o tamanho dos dados. Até 125 bytes, o tamanho cabe ali; se o valor é 126, o tamanho vem nos 2 bytes seguintes; se é 127, nos 8 seguintes;
- quadros do cliente para o servidor são sempre mascarados: vêm 4 bytes aleatórios de máscara, e cada byte dos dados passa por um XOR com um deles. Quadros do servidor para o cliente vêm sem máscara.
lib/hello/ws.ex: o WebSocket deste notebook. São cerca de cem linhas, e bastam para falar com o Phoenix. Não tratam mensagens divididas em vários quadros nem compressão, que o Phoenix não usa por padrão.
defmodule Hello.WS do
@moduledoc """
O mínimo de WebSocket (RFC 6455) para conversar com o Phoenix: o pedido de
upgrade, quadros de texto mascarados na ida e a leitura dos quadros na volta.
"""
defstruct [:socket, buffer: ""]
@guid "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
@doc "Abre a conexão TCP e pede o upgrade. Devolve {:ok, ws} ou {:erro, linha_de_status}."
def conectar(porta, caminho) do
{:ok, socket} = :gen_tcp.connect(~c"127.0.0.1", porta, [:binary, active: false])
chave = Base.encode64(:crypto.strong_rand_bytes(16))
pedido =
"GET #{caminho} HTTP/1.1\r\n" <>
"Host: localhost:#{porta}\r\n" <>
"Upgrade: websocket\r\n" <>
"Connection: Upgrade\r\n" <>
"Sec-WebSocket-Key: #{chave}\r\n" <>
"Sec-WebSocket-Version: 13\r\n\r\n"
:ok = :gen_tcp.send(socket, pedido)
{:ok, resposta} = :gen_tcp.recv(socket, 0, 5000)
[cabeca, resto] = String.split(resposta, "\r\n\r\n", parts: 2)
[linha_de_status | _] = String.split(cabeca, "\r\n")
aceite = :crypto.hash(:sha, chave <> @guid) |> Base.encode64()
cond do
not String.contains?(linha_de_status, " 101 ") ->
:gen_tcp.close(socket)
{:erro, linha_de_status}
not String.contains?(String.downcase(cabeca), "sec-websocket-accept: " <> String.downcase(aceite)) ->
:gen_tcp.close(socket)
{:erro, "o servidor não devolveu o Sec-WebSocket-Accept esperado"}
true ->
{:ok, %__MODULE__{socket: socket, buffer: resto}}
end
end
@doc "Envia um quadro de texto. Quadros do cliente para o servidor são sempre mascarados."
def enviar(%__MODULE__{socket: socket}, texto), do: :gen_tcp.send(socket, quadro(1, texto))
@doc """
Espera o próximo quadro de texto. Devolve {:ok, texto, ws}, {:fechado, código, ws}
quando o servidor encerra, ou {:tempo_esgotado, ws} se nada chegar a tempo.
"""
def receber(%__MODULE__{} = ws, tempo \\ 1000) do
case ler_quadro(ws.buffer) do
{:ok, 1, dados, resto} ->
{:ok, dados, %{ws | buffer: resto}}
{:ok, 8, dados, resto} ->
codigo =
case dados do
<<codigo::16, _motivo::binary>> -> codigo
_ -> nil
end
{:fechado, codigo, %{ws | buffer: resto}}
{:ok, 9, dados, resto} ->
# ping: responde com um pong (opcode 10) e continua esperando
:gen_tcp.send(ws.socket, quadro(10, dados))
receber(%{ws | buffer: resto}, tempo)
{:ok, _outro, _dados, resto} ->
receber(%{ws | buffer: resto}, tempo)
:incompleto ->
case :gen_tcp.recv(ws.socket, 0, tempo) do
{:ok, dados} -> receber(%{ws | buffer: ws.buffer <> dados}, tempo)
{:error, :timeout} -> {:tempo_esgotado, ws}
{:error, :closed} -> {:fechado, nil, ws}
end
end
end
@doc "Fecha a conexão TCP."
def fechar(%__MODULE__{socket: socket}), do: :gen_tcp.close(socket)
@doc "Monta um quadro mascarado: FIN + opcode, bit de máscara + tamanho, a máscara e os dados."
def quadro(opcode, dados) do
mascara = :crypto.strong_rand_bytes(4)
tamanho = byte_size(dados)
cabeca =
cond do
tamanho < 126 -> <<1::1, 0::3, opcode::4, 1::1, tamanho::7>>
tamanho < 65_536 -> <<1::1, 0::3, opcode::4, 1::1, 126::7, tamanho::16>>
true -> <<1::1, 0::3, opcode::4, 1::1, 127::7, tamanho::64>>
end
cabeca <> mascara <> mascarar(dados, mascara)
end
@doc "Aplica a máscara: XOR de cada byte com um dos 4 bytes da máscara, em rodízio."
def mascarar(dados, <<_::32>> = mascara) do
chaves = :binary.bin_to_list(mascara)
dados
|> :binary.bin_to_list()
|> Enum.with_index()
|> Enum.map(fn {byte, i} -> Bitwise.bxor(byte, Enum.at(chaves, rem(i, 4))) end)
|> :binary.list_to_bin()
end
# Quadros do servidor para o cliente vêm sem máscara (bit 0).
defp ler_quadro(<<_fin::1, _::3, op::4, 0::1, 126::7, n::16, dados::binary-size(n), resto::binary>>), do: {:ok, op, dados, resto}
defp ler_quadro(<<_fin::1, _::3, op::4, 0::1, 127::7, n::64, dados::binary-size(n), resto::binary>>), do: {:ok, op, dados, resto}
defp ler_quadro(<<_fin::1, _::3, op::4, 0::1, n::7, dados::binary-size(n), resto::binary>>) when n < 126, do: {:ok, op, dados, resto}
defp ler_quadro(_), do: :incompleto
endUm quadro de texto com oi, visto byte a byte:
Trilha.bytes(Hello.WS.quadro(1, "oi"))São 8 bytes. 81 é FIN mais o opcode 1 (texto). 82 é o bit de máscara mais o tamanho, 2. Os quatro seguintes são a máscara, diferente a cada avaliação, e os dois últimos são oi mascarado. A máscara não é criptografia: quem lê o quadro lê a máscara junto. Ela existe para que um proxy antigo no caminho não confunda os dados com uma requisição HTTP. Aplicar a mesma máscara duas vezes devolve o texto:
mascara = <<1, 2, 3, 4>>
"oi" |> Hello.WS.mascarar(mascara) |> Hello.WS.mascarar(mascara)Agora a conexão, com o módulo:
{:ok, ws} = Hello.WS.conectar(porta, "/socket/websocket?vsn=2.0.0")Parâmetros e recusas
O guia avisa: outros parâmetros ou cabeçalhos podem ser esperados ou exigidos pela função connect/3 do socket da aplicação. O MobileSocket deste notebook exige um:
def connect(%{"token" => "segredo-do-app"}, socket, _connect_info) do
{:ok, assign(socket, :app, "mobile")}
end
def connect(_params, _socket, _connect_info), do: :error
Trilha.pergunta(:recusa)Hello.WS.conectar(porta, "/mobile/websocket?vsn=2.0.0"){:erro, "HTTP/1.1 403 Forbidden"}, e no log REFUSED CONNECTION TO HelloWeb.MobileSocket. O connect/3 roda durante a requisição HTTP, antes do upgrade; se ele recusa, não chega a existir WebSocket. Com o token na query string, a conexão abre:
{:ok, mobile} = Hello.WS.conectar(porta, "/mobile/websocket?vsn=2.0.0&token=segredo-do-app")No log, o parâmetro aparece como "token" => "[FILTERED]": o Phoenix esconde dos logs os parâmetros com nomes como token e password.
A versão também pode ser recusada. O servidor só conhece dois serializadores:
Hello.WS.conectar(porta, "/socket/websocket?vsn=3.0.0")403 de novo, e um aviso no log: The client's requested transport version "3.0.0" does not match server's version requirements.
Sua vez: o endereço
A função abaixo monta o caminho do pedido de upgrade a partir do caminho do socket e dos parâmetros. Do jeito que está, ela esquece a versão e ignora os parâmetros. Complete-a: o resultado deve ter vsn=2.0.0 e os parâmetros, escapados, na query string. URI.encode_query/1 faz o escape.
defmodule Hello.Endereco do
@doc "O caminho do pedido de upgrade: o caminho do socket, /websocket, a versão e os parâmetros."
def caminho(base, _params \\ %{}) do
# Falta a versão do protocolo e faltam os parâmetros.
base <> "/websocket"
end
endTrilha.solucao(:endereco)Trilha.conferir(:endereco)O formato das mensagens
Na documentação: Message Format
O formato das mensagens depende do serializador configurado na aplicação. O guia supõe o Phoenix.Socket.V2.JSONSerializer, o de vsn=2.0.0. Com ele, toda mensagem que o cliente envia é uma lista JSON de cinco posições:
[join_reference, message_reference, topic_name, event_name, payload]
join_reference: escolhida pelo cliente, deve ser única. É obrigatória no eventophx_join, e o cliente deve repeti-la nas mensagens seguintes para o mesmo channel, para que o servidor descarte mensagens velhas, de uma entrada anterior. O servidor aceitanull, por compatibilidade.message_reference: escolhida pelo cliente, também única. O servidor a devolve na resposta, e é assim que o cliente sabe a que mensagem a resposta se refere.topic_name: um tópico que o socket conhece. O cliente precisa entrar nele antes de enviar qualquer mensagem.event_name: o primeiro argumento de umhandle_in/3no channel.payload: um valor serializável, em geral JSON; é o segundo argumento dohandle_in/3.
Um quadro de texto leva uma mensagem. A primeira, à mão, na conexão ws aberta acima: entrar em room:lobby.
:ok = Hello.WS.enviar(ws, ~s(["1", "1", "room:lobby", "phx_join", {}]))
{:ok, texto, ws} = Hello.WS.receber(ws)
textoA resposta é ["1","1","room:lobby","phx_reply",{"status":"ok","response":{}}]. As duas referências voltaram: a do join e a da mensagem. O evento é phx_reply, e o payload traz o status e a response, que é o que o join/3 do channel devolveu além do socket (aqui, nada). No log: JOINED room:lobby.
Para encurtar as próximas células, uma função que envia um texto, espera um quadro e o mostra:
falar = fn ws, texto ->
:ok = Hello.WS.enviar(ws, texto)
{:ok, resposta, ws} = Hello.WS.receber(ws)
IO.puts(resposta)
ws
end
ws = falar.(ws, ~s(["1", "2", "room:lobby", "ping", {"oi": 1}]))O ping do RoomChannel responde com o payload que recebeu: ["1","2","room:lobby","phx_reply",{"status":"ok","response":{"oi":1}}]. A ref agora é "2", a desta mensagem; o join_ref continua "1".
Trilha.pergunta(:refs_do_broadcast)O new_msg não responde ({:noreply, socket}); ele faz um broadcast!/3 para todos no tópico, inclusive quem enviou:
ws = falar.(ws, ~s(["1", "3", "room:lobby", "new_msg", {"body": "olá"}]))O que chegou foi [null,null,"room:lobby","new_msg",{"body":"olá"}]: sem join_ref e sem ref. Se a aba do chat está aberta, a mensagem apareceu nela. O mesmo vale para um broadcast que parte do servidor, sem cliente nenhum na origem:
HelloWeb.Endpoint.broadcast("room:lobby", "new_msg", %{body: "do servidor"})
{:ok, texto, ws} = Hello.WS.receber(ws)
textoEssa é a regra que o seu cliente vai usar: mensagem com ref é resposta a algo que ele enviou; mensagem sem ref é novidade do tópico.
O guia diz que a join_reference também serve de referência para os pushes do servidor, as mensagens que não respondem a nada. Vale para o que o channel envia com push/3 a um cliente só, e você vê isso logo abaixo, no exemplo de Miami. Os broadcasts chegam com as duas posições nulas, porque o Phoenix serializa a mensagem uma vez só para todos os inscritos.
Quando o servidor diz não
Três recusas, cada uma com uma resposta de status "error":
# O channel recusa a entrada: o join/3 devolveu {:error, %{reason: "unauthorized"}}.
ws = falar.(ws, ~s(["5", "5", "room:secreta", "phx_join", {}]))
# Nenhum channel do socket atende esse tópico.
ws = falar.(ws, ~s(["6", "6", "nada:aqui", "phx_join", {}]))
# Um evento para um tópico em que o cliente não entrou.
ws = falar.(ws, ~s([null, "7", "room:outra", "ping", {}]))A primeira resposta traz o motivo que o channel escolheu, {"reason":"unauthorized"}, e no log aparece REFUSED JOIN room:secreta. As outras duas vêm do próprio Phoenix, com {"reason":"unmatched topic"} e join_ref nulo: não há entrada a que a mensagem pertença. A segunda ainda deixa um aviso no log, Ignoring unmatched topic "nada:aqui" in HelloWeb.UserSocket.
A versão 1, a dos mapas
Sem vsn na URL, ou com vsn=1.0.0, o Phoenix usa o serializador antigo, em que cada mensagem é um objeto JSON:
{:ok, v1} = Hello.WS.conectar(porta, "/socket/websocket?vsn=1.0.0")
:ok = Hello.WS.enviar(v1, ~s({"topic": "room:lobby", "event": "phx_join", "payload": {}, "ref": "1"}))
{:ok, texto, v1} = Hello.WS.receber(v1)
Hello.WS.fechar(v1)
Jason.decode!(texto)A resposta é um objeto com as chaves "event", "payload", "ref" e "topic" (a célula a decodifica, porque a ordem das chaves no texto muda de uma execução para outra): as mesmas informações, com nomes, e sem join_ref. O log da conexão mostra Serializer: Phoenix.Socket.V1.JSONSerializer. O formato de lista da versão 2 é mais curto e trouxe o join_ref. Um cliente novo deve pedir sempre vsn=2.0.0, como o guia faz.
Os três eventos que toda aplicação entende
O guia lista três eventos que qualquer aplicação Phoenix entende, e dá os exemplos no tópico miami:weather. O MobileSocket do notebook tem esse tópico; a conexão mobile, aberta com o token, serve para repeti-los como estão no guia.
Primeiro, phx_join, para entrar num channel:
:ok = Hello.WS.enviar(mobile, ~s(["0", "0", "miami:weather", "phx_join", {"some": "param"}]))
{:ok, resposta, mobile} = Hello.WS.receber(mobile)
{:ok, push, mobile} = Hello.WS.receber(mobile)
IO.puts(resposta)
IO.puts(push)Chegaram duas mensagens. A primeira é a resposta do join; o WeatherChannel devolve nela os parâmetros que recebeu, para você ver que o payload do phx_join é o segundo argumento do join/3. A segunda é um push/3 que esse channel faz logo depois da entrada: ["0",null,"miami:weather","forecast",{"today":"sunny"}]. Ela traz o join_ref da entrada e nenhuma ref.
Uma aplicação define os próprios eventos. O exemplo do guia é um channel que trata report_emergency:
def handle_in("report_emergency", payload, socket) do
MyApp.Emergencies.report(payload) # or whatever
{:reply, :ok, socket}
end
e o cliente envia:
mobile = falar.(mobile, ~s(["0", "3", "miami:weather", "report_emergency", {"category": "sharknado"}])){:reply, :ok, socket} vira {"status":"ok","response":{}}.
Segundo, phx_leave, para sair de um channel:
:ok = Hello.WS.enviar(mobile, ~s(["0", "1", "miami:weather", "phx_leave", {}]))
{:ok, resposta, mobile} = Hello.WS.receber(mobile)
{:ok, fecho, mobile} = Hello.WS.receber(mobile)
IO.puts(resposta)
IO.puts(fecho)De novo duas mensagens: a resposta ao phx_leave e um phx_close, com o join_ref nas duas primeiras posições, que avisa que o processo do channel terminou. A conexão continua aberta; o cliente pode entrar em outro tópico ou voltar a este.
Terceiro, heartbeat, que mantém a conexão. Ele vai para o tópico especial phoenix e não pertence a nenhum join:
mobile = falar.(mobile, ~s([null, "2", "phoenix", "heartbeat", {}]))
Hello.WS.fechar(mobile)A seção "O heartbeat", mais abaixo, mostra o que acontece sem ele.
Sua vez: as mensagens
Escrever JSON à mão cansa e erra. Complete o módulo abaixo, que monta as mensagens como listas do Elixir e transforma o texto que chega num mapa. join/4 e codificar/1 já estão prontas; faltam leave/3, heartbeat/1, push/5 e decodificar/1. Os exemplos do guia, acima, dizem o que cada lista deve conter. decodificar/1 devolve um mapa com as chaves :join_ref, :ref, :topic, :event e :payload.
defmodule Hello.Mensagens do
@doc "Entrar num tópico."
def join(join_ref, ref, topico, params), do: [join_ref, ref, topico, "phx_join", params]
@doc "Sair de um tópico."
def leave(_join_ref, _ref, _topico), do: raise("Hello.Mensagens.leave/3 ainda não foi escrita")
@doc "O batimento que mantém a conexão aberta. Não pertence a nenhum join."
def heartbeat(_ref), do: raise("Hello.Mensagens.heartbeat/1 ainda não foi escrita")
@doc "Um evento da aplicação, para um tópico em que o cliente já entrou."
def push(_join_ref, _ref, _topico, _evento, _payload), do: raise("Hello.Mensagens.push/5 ainda não foi escrita")
@doc "A lista vira o texto JSON que vai no quadro."
def codificar(mensagem), do: Jason.encode!(mensagem)
@doc "O texto de um quadro vira um mapa com as cinco partes da mensagem."
def decodificar(_texto), do: raise("Hello.Mensagens.decodificar/1 ainda não foi escrita")
endTrilha.solucao(:mensagens)Trilha.conferir(:mensagens)Um cliente de verdade
O guia para na descrição das mensagens. Entre elas e um cliente utilizável falta uma camada: alguém que conte as referências, lembre o join_ref de cada tópico e saiba que resposta é de quem. É o que as bibliotecas da lista do guia fazem, cada uma no estilo da sua linguagem. O cliente JavaScript usa callbacks: channel.push("ping").receive("ok", fn) e channel.on("new_msg", fn). O deste notebook é funcional: cada operação recebe o cliente e devolve o cliente atualizado.
lib/hello/cliente.ex: o estado é uma struct com a conexão, o contador de referências, os tópicos em que o cliente entrou (cada um com seu join_ref) e a caixa, uma lista com as mensagens que chegaram enquanto ele esperava outra coisa.
defmodule Hello.Cliente do
@moduledoc """
Um cliente de Phoenix Channels: uma conexão, o contador de referências, os
tópicos em que entrou (com o join_ref de cada um) e a caixa das mensagens que
chegaram enquanto ele esperava outra coisa.
"""
alias Hello.{Mensagens, WS}
defstruct [:ws, ref: 0, topicos: %{}, caixa: []]
@doc "Conecta ao socket. `caminho` já inclui /websocket e a versão."
def conectar(porta, caminho) do
with {:ok, ws} <- WS.conectar(porta, caminho) do
{:ok, %__MODULE__{ws: ws}}
end
end
@doc "Entra num tópico e espera a resposta. O join_ref é a referência do próprio join."
def entrar(cliente, topico, params \\ %{}) do
{ref, cliente} = nova_ref(cliente)
WS.enviar(cliente.ws, Mensagens.codificar(Mensagens.join(ref, ref, topico, params)))
case esperar_resposta(cliente, ref, 5000) do
{:ok, resposta, cliente} -> {:ok, resposta, %{cliente | topicos: Map.put(cliente.topicos, topico, ref)}}
outro -> outro
end
end
@doc "Envia um evento e não espera nada. Devolve a referência usada."
def enviar(cliente, topico, evento, payload \\ %{}) do
{ref, cliente} = nova_ref(cliente)
mensagem = Mensagens.push(cliente.topicos[topico], ref, topico, evento, payload)
WS.enviar(cliente.ws, Mensagens.codificar(mensagem))
{ref, cliente}
end
@doc "Envia um evento e espera a resposta dele."
def chamar(cliente, topico, evento, payload \\ %{}, tempo \\ 5000) do
{ref, cliente} = enviar(cliente, topico, evento, payload)
esperar_resposta(cliente, ref, tempo)
end
@doc "Sai de um tópico."
def sair(cliente, topico) do
{ref, cliente} = nova_ref(cliente)
WS.enviar(cliente.ws, Mensagens.codificar(Mensagens.leave(cliente.topicos[topico], ref, topico)))
case esperar_resposta(cliente, ref, 5000) do
{:ok, resposta, cliente} -> {:ok, resposta, %{cliente | topicos: Map.delete(cliente.topicos, topico)}}
outro -> outro
end
end
@doc "A próxima mensagem que não é resposta: primeiro as guardadas na caixa, depois o fio."
def proxima(cliente, tempo \\ 1000)
def proxima(%__MODULE__{caixa: [mensagem | resto]} = cliente, _tempo), do: {:ok, mensagem, %{cliente | caixa: resto}}
def proxima(cliente, tempo), do: ler(cliente, tempo)
@doc """
Espera a resposta da mensagem `ref`. Devolve {:ok, resposta, cliente} ou
{:erro, resposta, cliente}, conforme o status; {:tempo_esgotado, cliente} se
ela não chegar; {:fechado, código, cliente} se o servidor encerrar.
"""
def esperar_resposta(cliente, _ref, tempo) do
# Versão ingênua: supõe que a próxima mensagem do fio é a resposta.
case ler(cliente, tempo) do
{:ok, mensagem, cliente} ->
status = if mensagem.payload["status"] == "ok", do: :ok, else: :erro
{status, mensagem.payload["response"], cliente}
outro ->
outro
end
end
@doc "Lê uma mensagem do fio, já decodificada."
def ler(cliente, tempo) do
case WS.receber(cliente.ws, tempo) do
{:ok, texto, ws} -> {:ok, Mensagens.decodificar(texto), %{cliente | ws: ws}}
{:fechado, codigo, ws} -> {:fechado, codigo, %{cliente | ws: ws}}
{:tempo_esgotado, ws} -> {:tempo_esgotado, %{cliente | ws: ws}}
end
end
@doc "Uma referência nova: o contador, como texto."
def nova_ref(cliente), do: {Integer.to_string(cliente.ref + 1), %{cliente | ref: cliente.ref + 1}}
@doc "Fecha a conexão."
def fechar(cliente), do: WS.fechar(cliente.ws)
endAs operações:
entrar/3envia ophx_joine espera a resposta. A referência do join serve também dejoin_ref, como faz o cliente JavaScript;enviar/4envia um evento e não espera;chamar/5envia e espera a resposta;proxima/2devolve a próxima novidade do tópico;esperar_resposta/3é quem lê o fio até achar a resposta de umaref.
{:ok, cliente} = Hello.Cliente.conectar(porta, "/socket/websocket?vsn=2.0.0")
{:ok, _resposta, cliente} = Hello.Cliente.entrar(cliente, "room:lobby")
{:ok, resposta, cliente} = Hello.Cliente.chamar(cliente, "room:lobby", "ping", %{"n" => 1})
respostaFunciona: %{"n" => 1}. Mas a esperar_resposta/3 desta célula é ingênua. Ela supõe que a próxima mensagem do fio é a resposta. Veja o que acontece quando alguém escreve na sala um instante antes da resposta chegar:
HelloWeb.Endpoint.broadcast("room:lobby", "new_msg", %{body: "cheguei antes"})
Process.sleep(100)
{status, resposta, cliente} = Hello.Cliente.chamar(cliente, "room:lobby", "ping", %{"n" => 2})
{status, resposta}Com a versão ingênua, o resultado é {:erro, nil}: o cliente leu o new_msg, não achou status nem response no payload, e o entregou como se fosse a resposta do ping. A resposta de verdade ficou no fio e vai ser lida como resposta da próxima chamada. Daí em diante, tudo sai trocado. É o problema que a primeira pergunta deste notebook anunciava.
Sua vez: respostas e novidades
Reescreva esperar_resposta/3, na célula do Hello.Cliente, para que ela:
- leia mensagens do fio, com
ler/2, até achar uma comevent: "phx_reply"e arefpedida; - devolva
{:ok, resposta, cliente}se ostatusfor"ok", e{:erro, resposta, cliente}se não for; - guarde no fim da caixa (
cliente.caixa) toda mensagem que não for a resposta esperada, e continue lendo; - repasse o que
ler/2devolver quando não for mensagem:{:tempo_esgotado, cliente}ou{:fechado, codigo, cliente}.
Depois de avaliar a célula, reavalie as duas células acima. O resultado da segunda passa a ser {:ok, %{"n" => 2}}, e o new_msg fica guardado.
Trilha.solucao(:respostas)Trilha.conferir(:respostas)Com a função certa, as novidades que chegaram durante as esperas saem pela proxima/2, na ordem:
{resultado, mensagem, cliente} = Hello.Cliente.proxima(cliente)
{resultado, mensagem}Repare num efeito colateral: a conferência acima também abriu conexões, entrou em room:lobby e fez broadcasts na sala (no meio do caminho, primeira, segunda). Este cliente está na mesma sala, então essas mensagens também chegaram a ele e estão esperando no fio. Um tópico é compartilhado por todos que entram nele. Para as próximas células, um cliente novo, de caixa vazia:
Hello.Cliente.fechar(cliente)
{:ok, cliente} = Hello.Cliente.conectar(porta, "/socket/websocket?vsn=2.0.0")
{:ok, _resposta, cliente} = Hello.Cliente.entrar(cliente, "room:lobby")
cliente.caixaUm evento que não responde, como o shout, é enviado com enviar/4. O broadcast volta para todos no tópico, inclusive para quem enviou:
{_ref, cliente} = Hello.Cliente.enviar(cliente, "room:lobby", "shout", %{"de" => "Elixir"})
{:ok, mensagem, cliente} = Hello.Cliente.proxima(cliente)
mensagemO cliente oficial, na outra ponta
Com a aba http://localhost:4010 aberta, a mensagem abaixo aparece na página. Quem a recebe lá é o phoenix.js, o cliente oficial; no fio, ele fala exatamente o que o seu cliente fala.
{_ref, cliente} = Hello.Cliente.enviar(cliente, "room:lobby", "new_msg", %{"body" => "Olá do cliente em Elixir"})
{:ok, mensagem, cliente} = Hello.Cliente.proxima(cliente)
mensagem.payloadO caminho inverso também funciona: escreva algo na página, aperte Enter e avalie Hello.Cliente.proxima(cliente, 5000) numa célula nova. As ferramentas do navegador (aba Network, filtro WS) mostram as mensagens do phoenix.js, inclusive um heartbeat a cada 30 segundos.
O heartbeat
O guia diz que o heartbeat só é necessário quando nenhuma outra mensagem está sendo enviada, e que ele impede o Phoenix de fechar a conexão; o tempo exato é o :timeout configurado no Endpoint da aplicação.
Trilha.pergunta(:ocioso)O padrão do :timeout é 60 segundos. Para não esperar tanto, o endpoint deste notebook tem o socket /impaciente, igual ao /socket, com websocket: [timeout: 2_000]. A célula conecta, não envia nada e mede quanto tempo a conexão dura:
{:ok, ocioso} = Hello.WS.conectar(porta, "/impaciente/websocket?vsn=2.0.0")
inicio = System.monotonic_time(:millisecond)
{:fechado, codigo, _ocioso} = Hello.WS.receber(ocioso, 5000)
{codigo, System.monotonic_time(:millisecond) - inicio}Uns dois segundos depois, o servidor envia um quadro de encerramento (opcode 8), com o código 1002, e fecha a conexão. O que conta é o tempo sem receber nada do cliente; mensagens que o servidor envia não zeram o relógio.
Com um heartbeat a cada 1,2 segundo, a mesma conexão passa dos 2 segundos:
{:ok, ocioso} = Hello.WS.conectar(porta, "/impaciente/websocket?vsn=2.0.0")
for n <- 1..4 do
Process.sleep(1200)
:ok = Hello.WS.enviar(ocioso, Jason.encode!([nil, "#{n}", "phoenix", "heartbeat", %{}]))
{:ok, texto, _} = Hello.WS.receber(ocioso)
texto
endQuatro respostas, de [null,"1","phoenix","phx_reply",{"status":"ok","response":{}}] até a de ref "4", em quase 5 segundos de conexão. O cliente JavaScript envia um heartbeat a cada 30 segundos, metade do tempo padrão do servidor. A resposta tem uma segunda utilidade: se ela não chega, o cliente descobre que a conexão morreu sem aviso e reconecta.
Sua vez: manter a conexão viva
Hello.Batimento.bater/1 envia um heartbeat pelo cliente e espera a resposta. Complete manter_vivo/3: ela deve ficar duracao milissegundos sem fazer mais nada, batendo a cada intervalo, e devolver {:ok, cliente}. Do jeito que está, ela só espera, e a conexão impaciente não sobrevive.
defmodule Hello.Batimento do
alias Hello.{Cliente, Mensagens, WS}
@doc "Envia um heartbeat e espera a resposta."
def bater(cliente) do
{ref, cliente} = Cliente.nova_ref(cliente)
WS.enviar(cliente.ws, Mensagens.codificar(Mensagens.heartbeat(ref)))
Cliente.esperar_resposta(cliente, ref, 5000)
end
@doc "Fica `duracao` milissegundos sem fazer nada, batendo a cada `intervalo`."
def manter_vivo(cliente, duracao, _intervalo) do
# Só espera. Falta bater.
Process.sleep(duracao)
{:ok, cliente}
end
end{:ok, paciente} = Hello.Cliente.conectar(porta, "/impaciente/websocket?vsn=2.0.0")
{:ok, _resposta, paciente} = Hello.Cliente.entrar(paciente, "room:lobby")
{:ok, paciente} = Hello.Batimento.manter_vivo(paciente, 3000, 1000)
Hello.Cliente.chamar(paciente, "room:lobby", "ping", %{"ainda" => "aqui"}, 1000) |> Tuple.delete_at(2)Antes do exercício, o resultado é {:fechado, 1002}: quando o ping foi enviado, o servidor já tinha encerrado a conexão. Depois, {:ok, %{"ainda" => "aqui"}}.
Trilha.solucao(:batimento)Trilha.conferir(:batimento)A conferência leva uns 7 segundos: ela deixa a sua função manter duas conexões impacientes de pé.
Quando o channel cai
O guia de Channels, na seção Fault Tolerance and Reliability Guarantees, diz que, se um channel cai, o servidor avisa os clientes, e eles precisam entrar de novo nos tópicos. O cliente JavaScript faz isso sozinho. O seu ainda não.
Trilha.pergunta(:evento_desconhecido)O evento explodir do RoomChannel levanta uma exceção. A célula entra na sala com um cliente novo, envia o evento e lê o que chega. O log mostra o erro do processo, GenServer ... terminating com (RuntimeError) o channel caiu de propósito; é esperado.
{:ok, cliente} = Hello.Cliente.conectar(porta, "/socket/websocket?vsn=2.0.0")
{:ok, _resposta, cliente} = Hello.Cliente.entrar(cliente, "room:lobby")
{_ref, cliente} = Hello.Cliente.enviar(cliente, "room:lobby", "explodir")
{:ok, queda, cliente} = Hello.Cliente.proxima(cliente)
quedaChegou um phx_error, com %{"reason" => "channel_crash"} e o join_ref da entrada nas duas primeiras posições. Não é uma resposta: o evento explodir nunca vai ter resposta. A conexão continua aberta, mas a entrada naquele tópico acabou:
Hello.Cliente.chamar(cliente, "room:lobby", "ping", %{}, 1000) |> Tuple.delete_at(2){:erro, %{"reason" => "unmatched topic"}}: para o servidor, este cliente não está mais em room:lobby. Um evento que o channel não conhece dá no mesmo, porque nenhuma cláusula de handle_in/3 casa e o processo cai com FunctionClauseError.
Sua vez: reentrar
Complete Hello.Reentrada.tratar/2. Ela recebe o cliente e uma mensagem que chegou. Se a mensagem for um phx_error, a função entra de novo no tópico da mensagem, com Hello.Cliente.entrar/2, e devolve {:reentrou, cliente}. Para qualquer outra mensagem, devolve {:ignorada, cliente}.
defmodule Hello.Reentrada do
@doc """
Olha uma mensagem que chegou. Se for o aviso de que o channel caiu, entra de
novo no tópico e devolve {:reentrou, cliente}; senão, {:ignorada, cliente}.
"""
def tratar(cliente, _mensagem) do
# Ainda não reconhece o phx_error.
{:ignorada, cliente}
end
endcase Hello.Reentrada.tratar(cliente, queda) do
{:reentrou, cliente} ->
Hello.Cliente.chamar(cliente, "room:lobby", "ping", %{"de" => "volta"}, 1000) |> Tuple.delete_at(2)
{:ignorada, _cliente} ->
:ignorada
endAntes do exercício, :ignorada. Depois, {:ok, %{"de" => "volta"}}.
Trilha.solucao(:reentrar)Trilha.conferir(:reentrar)entrar/3 usa uma referência nova como join_ref. Isso importa por causa das mensagens atrasadas:
Trilha.pergunta(:join_velho)A célula abaixo entra, derruba o channel, entra de novo e envia um ping com o join_ref da primeira entrada:
{:ok, cliente} = Hello.Cliente.conectar(porta, "/socket/websocket?vsn=2.0.0")
{:ok, _resposta, cliente} = Hello.Cliente.entrar(cliente, "room:lobby")
antigo = cliente.topicos["room:lobby"]
{_ref, cliente} = Hello.Cliente.enviar(cliente, "room:lobby", "explodir")
{:ok, _queda, cliente} = Hello.Cliente.proxima(cliente)
{:ok, _resposta, cliente} = Hello.Cliente.entrar(cliente, "room:lobby")
novo = cliente.topicos["room:lobby"]
:ok = Hello.WS.enviar(cliente.ws, Jason.encode!([antigo, "99", "room:lobby", "ping", %{}]))
{resultado, _cliente} = Hello.Cliente.proxima(cliente, 500)
{antigo, novo, resultado}{"1", "3", :tempo_esgotado}: a mensagem com o join_ref antigo foi descartada, sem resposta. O servidor não tem como saber se ela era para o processo que caiu, então não a entrega ao novo. Um cliente não pode esperar uma resposta para sempre; por isso chamar/5 tem um tempo limite.
O guia de Channels lembra o resto das garantias: o Phoenix entrega cada mensagem no máximo uma vez. Se o cliente estava fora quando a mensagem passou, ele não a recebe depois. Quem precisa de mais guarda as mensagens no servidor e deixa o cliente pedir, ao entrar, o que perdeu.
Sem roteiro: o robô da vitrine
O Brechó ganhou um tópico para quem está olhando a vitrine, vitrine:lobby, servido pelo HelloWeb.VitrineChannel, que já está no ar:
- um visitante envia o evento
pergunta, com%{"texto" => "preço? Jaqueta jeans"}, e o channel repassa a pergunta a todos no tópico; - qualquer participante pode enviar
resposta, com%{"texto" => ...}, que também vai para todos; - quando uma peça é vendida, o servidor avisa o tópico com
HelloWeb.Endpoint.broadcast("vitrine:lobby", "venda", %{peca: "Bota"}).
Falta quem responda. Escreva Hello.Robo.atender(cliente, precos, quantas), um robô feito com o seu cliente:
- recebe um cliente conectado e entra em
vitrine:lobby; - para cada
perguntacom o texto"preço? NOME", envia umaresposta:"NOME custa R$ 80,00"se a peça está emprecos(um mapa de nome para preço em centavos;reais/1formata), ou"Não temos NOME"se não está; - para cada
venda, anota o nome da peça; - as
respostas que voltam pelo tópico, inclusive as dele mesmo, não contam; - depois de atender
quantasmensagens (perguntas e vendas somadas), devolve{:ok, %{vendas: [...], respostas: n}, cliente}, com as vendas na ordem em que aconteceram.
lib/hello/robo.ex
defmodule Hello.Robo do
@doc "80_00 centavos viram \"R$ 80,00\"."
def reais(centavos) do
"R$ #{div(centavos, 100)},#{centavos |> rem(100) |> Integer.to_string() |> String.pad_leading(2, "0")}"
end
# def atender(cliente, precos, quantas) do
# ...
# end
endPara testar à mão: a célula abaixo põe um visitante num processo à parte, que pergunta o preço da bota, espera a resposta e então dispara uma venda. O robô atende as duas mensagens no processo da célula.
precos = %{"Jaqueta jeans" => 80_00, "Vestido floral" => 45_00, "Bota" => 120_00}
celula = self()
spawn(fn ->
{:ok, visitante} = Hello.Cliente.conectar(porta, "/socket/websocket?vsn=2.0.0")
{:ok, _resposta, visitante} = Hello.Cliente.entrar(visitante, "vitrine:lobby")
Process.sleep(300)
{:ok, _resposta, visitante} = Hello.Cliente.chamar(visitante, "vitrine:lobby", "pergunta", %{"texto" => "preço? Bota"})
ouvir = fn ouvir, visitante ->
case Hello.Cliente.proxima(visitante, 2000) do
{:ok, %{event: "resposta", payload: payload}, _visitante} -> payload["texto"]
{:ok, _outra, visitante} -> ouvir.(ouvir, visitante)
{:tempo_esgotado, _visitante} -> :ninguem_respondeu
end
end
send(celula, {:visitante_ouviu, ouvir.(ouvir, visitante)})
HelloWeb.Endpoint.broadcast("vitrine:lobby", "venda", %{peca: "Bota"})
end)
{:ok, robo} = Hello.Cliente.conectar(porta, "/socket/websocket?vsn=2.0.0")
{:ok, relatorio, _robo} = Hello.Robo.atender(robo, precos, 2)
receive do
{:visitante_ouviu, texto} -> {texto, relatorio}
after
3000 -> {:visitante_nao_ouviu_nada, relatorio}
endCom o robô pronto, o resultado é {"Bota custa R$ 120,00", %{respostas: 1, vendas: ["Bota"]}}. Antes disso, a célula para com function Hello.Robo.atender/3 is undefined or private.
Trilha.solucao(:brecho)Trilha.conferir(:brecho)Uma dica, se a conferência acusar uma resposta trocada: o robô envia a resposta com chamar/4, e o broadcast dessa mesma resposta pode chegar antes do phx_reply. É o caso que esperar_resposta/3 resolve.
No seu projeto
- No seu projeto, os sockets estão em
lib/hello_web/endpoint.ex(socket "/socket", HelloWeb.UserSocket, ..., quemix phx.gen.socket Userpede para acrescentar), e o caminho do WebSocket é sempre o do socket mais/websocket. Um projeto novo também trazsocket "/live", Phoenix.LiveView.Socket: o LiveView fala este mesmo protocolo, com tópicoslv:..., como você viu no notebook 16. - Para testar um channel à mão, sem escrever cliente, use uma ferramenta de WebSocket de linha de comando, como o websocat:
websocat "ws://localhost:4000/socket/websocket?vsn=2.0.0"e, em cada linha, uma mensagem como["1","1","room:lobby","phx_join",{}]. No zsh, ponha a URL entre aspas, por causa do?. - No navegador, use o cliente oficial, o
phoenix.js, que vem em todo projeto novo; a documentação está em https://hexdocs.pm/phoenix/js/. Ele faz o que você escreveu aqui e mais: reconecta com espera crescente, reentra nos tópicos, guarda as mensagens enviadas sem conexão e bate o heartbeat. - Em outra plataforma, procure antes na lista de clientes do guia de Channels: Swift, Java, Kotlin, C#, GDScript e, em Elixir, slipstream e phoenix_gen_socket_client. Em Elixir, para o WebSocket em si, o guia cita a mint_web_socket, que trata também TLS, quadros divididos e compressão, que o
Hello.WSdeixou de fora. - Nos testes do servidor, você não precisa de cliente nenhum:
Phoenix.ChannelTestconversa com o channel direto, processo a processo. O módulo de testes da trilha chega lá.
Revisão
Sem olhar para cima, responda em voz alta ou por escrito:
- Quais são as três partes do caminho
/mobile/websocket?vsn=2.0.0, e o que cada uma escolhe? - O que o cliente vê quando o
connect/3do socket devolve:error? Por que não é uma mensagem em JSON? - Quais são as cinco posições de uma mensagem, e quem escolhe as duas primeiras?
- Como um cliente separa uma resposta de um broadcast?
- O que chega depois de um
phx_leave? E depois que o channel cai? - Para que serve o
heartbeat, e o que conta para o:timeoutdo servidor? - O que o servidor faz com uma mensagem que traz o
join_refde uma entrada antiga? - Por que
esperar_resposta/3precisa de uma caixa?
Se alguma resposta não veio, volte à seção correspondente e reavalie as células.
Com este notebook termina o módulo de tempo real. O 18 mostrou os Channels pelo lado do servidor: sockets, tópicos, eventos e tokens. O 19 mostrou a Presence, que sabe quem está em cada tópico. Este mostrou o outro lado do fio: cinco posições numa lista, três eventos que toda aplicação entende e uma regra para separar respostas de novidades. O LiveView, do módulo anterior, é um cliente e um servidor desse mesmo protocolo.
