Skip to content

Commit

Permalink
Add example for use with GenServer (#20)
Browse files Browse the repository at this point in the history
  • Loading branch information
hkrutzer authored Mar 31, 2022
1 parent 6281c05 commit b6eccfc
Showing 1 changed file with 184 additions and 0 deletions.
184 changes: 184 additions & 0 deletions examples/genserver.exs
Original file line number Diff line number Diff line change
@@ -0,0 +1,184 @@
Mix.install([:mint_web_socket, :castore])

# Also see https://github.com/phoenixframework/phoenix/blob/4da71906da970a162c88e165cdd2fdfaf9083ac3/test/support/websocket_client.exs

defmodule Ws do
use GenServer

require Logger
require Mint.HTTP

defstruct [:conn, :websocket, :request_ref, :caller, :status, :resp_headers, :closing?]

def connect(url) do
with {:ok, socket} <- GenServer.start_link(__MODULE__, []),
{:ok, :connected} <- GenServer.call(socket, {:connect, url}) do
{:ok, socket}
end
end

def send_message(pid, text) do
GenServer.call(pid, {:send_text, text})
end

@impl GenServer
def init([]) do
{:ok, %__MODULE__{}}
end

@impl GenServer
def handle_call({:send_text, text}, _from, state) do
{:ok, state} = send_frame(state, {:text, text})
{:reply, :ok, state}
end

@impl GenServer
def handle_call({:connect, url}, from, state) do
uri = URI.parse(url)

http_scheme =
case uri.scheme do
"ws" -> :http
"wss" -> :https
end

ws_scheme =
case uri.scheme do
"ws" -> :ws
"wss" -> :wss
end

path =
case uri.query do
nil -> uri.path
query -> uri.path <> "?" <> query
end

with {:ok, conn} <- Mint.HTTP.connect(http_scheme, uri.host, uri.port),
{:ok, conn, ref} <- Mint.WebSocket.upgrade(ws_scheme, conn, path, []) do
state = %{state | conn: conn, request_ref: ref, caller: from}
{:noreply, state}
else
{:error, reason} ->
{:reply, {:error, reason}, state}

{:error, conn, reason} ->
{:reply, {:error, reason}, put_in(state.conn, conn)}
end
end

@impl GenServer
def handle_info(message, state) do
case Mint.WebSocket.stream(state.conn, message) do
{:ok, conn, responses} ->
state = put_in(state.conn, conn) |> handle_responses(responses)
if state.closing?, do: do_close(state), else: {:noreply, state}

{:error, conn, reason, _responses} ->
state = put_in(state.conn, conn) |> reply({:error, reason})
{:noreply, state}

:unknown ->
{:noreply, state}
end
end

defp handle_responses(state, responses)

defp handle_responses(%{request_ref: ref} = state, [{:status, ref, status} | rest]) do
put_in(state.status, status)
|> handle_responses(rest)
end

defp handle_responses(%{request_ref: ref} = state, [{:headers, ref, resp_headers} | rest]) do
put_in(state.resp_headers, resp_headers)
|> handle_responses(rest)
end

defp handle_responses(%{request_ref: ref} = state, [{:done, ref} | rest]) do
case Mint.WebSocket.new(state.conn, ref, state.status, state.resp_headers) do
{:ok, conn, websocket} ->
%{state | conn: conn, websocket: websocket, status: nil, resp_headers: nil}
|> reply({:ok, :connected})
|> handle_responses(rest)

{:error, conn, reason} ->
put_in(state.conn, conn)
|> reply({:error, reason})
end
end

defp handle_responses(%{request_ref: ref, websocket: websocket} = state, [
{:data, ref, data} | rest
])
when websocket != nil do
case Mint.WebSocket.decode(websocket, data) do
{:ok, websocket, frames} ->
put_in(state.websocket, websocket)
|> handle_frames(frames)
|> handle_responses(rest)

{:error, websocket, reason} ->
put_in(state.websocket, websocket)
|> reply({:error, reason})
end
end

defp handle_responses(state, [_response | rest]) do
handle_responses(state, rest)
end

defp handle_responses(state, []), do: state

defp send_frame(state, frame) do
with {:ok, websocket, data} <- Mint.WebSocket.encode(state.websocket, frame),
state = put_in(state.websocket, websocket),
{:ok, conn} <- Mint.WebSocket.stream_request_body(state.conn, state.request_ref, data) do
{:ok, put_in(state.conn, conn)}
else
{:error, %Mint.WebSocket{} = websocket, reason} ->
{:error, put_in(state.websocket, websocket), reason}

{:error, conn, reason} ->
{:error, put_in(state.conn, conn), reason}
end
end

def handle_frames(state, frames) do
Enum.reduce(frames, state, fn
# reply to pings with pongs
{:ping, data}, state ->
{:ok, state} = send_frame(state, {:pong, data})
state

{:close, _code, reason}, state ->
Logger.debug("Closing connection: #{inspect(reason)}")
%{state | closing?: true}

{:text, text}, state ->
Logger.debug("Received: #{inspect(text)}, sending back the reverse")
{:ok, state} = send_frame(state, {:text, String.reverse(text)})
state

frame, state ->
Logger.debug("Unexpected frame received: #{inspect(frame)}")
state
end)
end

defp do_close(state) do
# Streaming a close frame may fail if the server has already closed
# for writing.
_ = send_frame(state, :close)
Mint.HTTP.close(state.conn)
{:stop, :normal, state}
end

defp reply(state, response) do
if state.caller, do: GenServer.reply(state.caller, response)
put_in(state.caller, nil)
end
end

{:ok, pid} = Ws.connect("ws://localhost:1234/")
Ws.send_message(pid, "Hello from WS client")

0 comments on commit b6eccfc

Please sign in to comment.