Merge pull request #934 from poanetwork/924
Reconnect, rerequest, and resubscribe when websocket disconnectspull/939/merge
Luke Imhoff
6 years ago
committed by
GitHub
commit
c23c10520d
@ -0,0 +1,17 @@ |
||||
defmodule EthereumJSONRPC.WebSocket.WebSocketClient.Options do |
||||
@moduledoc """ |
||||
`t:EthereumJSONRPC.WebSocket.options/0` for `EthereumJSONRPC.WebSocket.WebSocketClient` `t:EthereumJSONRPC.Subscription.t/0` `transport_options`. |
||||
""" |
||||
|
||||
alias EthereumJSONRPC.Subscription |
||||
|
||||
@enforce_keys ~w(web_socket)a |
||||
defstruct ~w(web_socket event params)a |
||||
|
||||
@typedoc """ |
||||
* `web_socket` - the `t:pid/0` of the `EthereumJSONRPC.WebSocket.WebSocketClient`. |
||||
* `event` - the event that should be resubscribed to after disconnect. |
||||
* `params` - the parameters that should be used to customized `event` when resubscribing after disconnect. |
||||
""" |
||||
@type t :: %__MODULE__{web_socket: pid(), event: Subscription.event(), params: Subscription.params()} |
||||
end |
||||
@ -0,0 +1,223 @@ |
||||
defmodule EthereumJSONRPC.WebSocket.WebSocketClientTest do |
||||
use ExUnit.Case |
||||
|
||||
alias EthereumJSONRPC.Subscription |
||||
alias EthereumJSONRPC.WebSocket.{Registration, WebSocketClient} |
||||
|
||||
import EthereumJSONRPC, only: [unique_request_id: 0] |
||||
|
||||
describe "ondisconnect/2" do |
||||
setup :example_state |
||||
|
||||
test "reconnects", %{state: state} do |
||||
assert {:reconnect, _} = WebSocketClient.ondisconnect({:closed, :remote}, state) |
||||
end |
||||
|
||||
test "treats in-progress unsubscribes as successful", %{state: state} do |
||||
subscription_id = 1 |
||||
|
||||
state = put_subscription(state, subscription_id) |
||||
|
||||
%Registration{from: {_, ref}} = |
||||
registration = registration(%{type: :unsubscribe, subscription_id: subscription_id}) |
||||
|
||||
state = put_registration(state, registration) |
||||
|
||||
assert {_, disconnected_state} = WebSocketClient.ondisconnect({:closed, :remote}, state) |
||||
|
||||
assert Enum.empty?(disconnected_state.request_id_to_registration) |
||||
assert Enum.empty?(disconnected_state.subscription_id_to_subscription_reference) |
||||
assert Enum.empty?(disconnected_state.subscription_reference_to_subscription) |
||||
assert Enum.empty?(disconnected_state.subscription_reference_to_subscription_id) |
||||
|
||||
assert_receive {^ref, :ok} |
||||
end |
||||
|
||||
test "keeps :json_rpc requests for re-requesting on reconnect", %{state: state} do |
||||
state = put_registration(state, %{type: :json_rpc, method: "eth_getBlockByNumber", params: [1, true]}) |
||||
|
||||
assert {_, disconnected_state} = WebSocketClient.ondisconnect({:closed, :remote}, state) |
||||
|
||||
assert Enum.count(disconnected_state.request_id_to_registration) == 1 |
||||
end |
||||
|
||||
test "keeps :subscribe requests for re-requesting on reconnect", %{state: state} do |
||||
state = put_registration(state, %{type: :subscribe}) |
||||
|
||||
assert {_, disconnected_state} = WebSocketClient.ondisconnect({:closed, :remote}, state) |
||||
|
||||
assert Enum.count(disconnected_state.request_id_to_registration) == 1 |
||||
end |
||||
end |
||||
|
||||
describe "websocket_handle/3" do |
||||
setup :example_state |
||||
|
||||
test "Jason.decode errors are broadcast to all subscribers", %{state: %WebSocketClient{url: url} = state} do |
||||
subscription_id = 1 |
||||
subscription_reference = make_ref() |
||||
subscription = subscription(%{url: url, reference: subscription_reference}) |
||||
state = put_subscription(state, subscription_id, subscription) |
||||
|
||||
assert {:ok, ^state} = WebSocketClient.websocket_handle({:text, ""}, nil, state) |
||||
assert_receive {^subscription, {:error, %Jason.DecodeError{}}} |
||||
end |
||||
end |
||||
|
||||
describe "websocket_terminate/3" do |
||||
setup :example_state |
||||
|
||||
test "broadcasts close to all subscribers", %{state: %WebSocketClient{url: url} = state} do |
||||
subscription_id = 1 |
||||
subscription_reference = make_ref() |
||||
subscription = subscription(%{url: url, reference: subscription_reference}) |
||||
state = put_subscription(state, subscription_id, subscription) |
||||
|
||||
assert {:ok, ^state} = WebSocketClient.websocket_handle({:text, ""}, nil, state) |
||||
assert_receive {^subscription, {:error, %Jason.DecodeError{}}} |
||||
end |
||||
end |
||||
|
||||
describe "reconnect" do |
||||
setup do |
||||
dispatch = :cowboy_router.compile([{:_, [{"/websocket", EthereumJSONRPC.WebSocket.Cowboy.WebSocketHandler, []}]}]) |
||||
{:ok, _} = :cowboy.start_http(EthereumJSONRPC.WebSocket.Cowboy, 100, [], env: [dispatch: dispatch]) |
||||
|
||||
on_exit(fn -> |
||||
:ranch.stop_listener(EthereumJSONRPC.WebSocket.Cowboy) |
||||
end) |
||||
|
||||
port = :ranch.get_port(EthereumJSONRPC.WebSocket.Cowboy) |
||||
|
||||
pid = start_supervised!({WebSocketClient, ["ws://localhost:#{port}/websocket", []]}) |
||||
|
||||
%{pid: pid, port: port} |
||||
end |
||||
|
||||
test "resubscribes", %{pid: pid, port: port} do |
||||
assert {:ok, subscription} = WebSocketClient.subscribe(pid, "newHeads", []) |
||||
|
||||
assert_receive {^subscription, {:ok, %{}}}, 500 |
||||
|
||||
assert :ok = :ranch.stop_listener(EthereumJSONRPC.WebSocket.Cowboy) |
||||
|
||||
refute_receive {^subscription, {:ok, %{}}}, 100 |
||||
|
||||
cowboy(port) |
||||
|
||||
assert_receive {^subscription, {:ok, %{}}}, 500 |
||||
end |
||||
|
||||
test "rerequests", %{pid: pid, port: port} do |
||||
first_params = [1] |
||||
|
||||
# json_rpc requests work before connection is closed |
||||
assert {:ok, ^first_params} = |
||||
WebSocketClient.json_rpc( |
||||
pid, |
||||
EthereumJSONRPC.request(%{id: :erlang.unique_integer(), method: "echo", params: first_params}) |
||||
) |
||||
|
||||
assert :ok = :ranch.stop_listener(EthereumJSONRPC.WebSocket.Cowboy) |
||||
|
||||
spawn_link(fn -> |
||||
Process.sleep(500) |
||||
cowboy(port) |
||||
end) |
||||
|
||||
second_params = [2] |
||||
|
||||
assert {:ok, ^second_params} = |
||||
WebSocketClient.json_rpc( |
||||
pid, |
||||
EthereumJSONRPC.request(%{id: :erlang.unique_integer(), method: "echo", params: second_params}) |
||||
) |
||||
end |
||||
end |
||||
|
||||
defp cowboy(0) do |
||||
dispatch = :cowboy_router.compile([{:_, [{"/websocket", EthereumJSONRPC.WebSocket.Cowboy.WebSocketHandler, []}]}]) |
||||
{:ok, _} = :cowboy.start_http(EthereumJSONRPC.WebSocket.Cowboy, 100, [], env: [dispatch: dispatch]) |
||||
:ranch.get_port(EthereumJSONRPC.WebSocket.Cowboy) |
||||
end |
||||
|
||||
defp cowboy(port) do |
||||
dispatch = :cowboy_router.compile([{:_, [{"/websocket", EthereumJSONRPC.WebSocket.Cowboy.WebSocketHandler, []}]}]) |
||||
{:ok, _} = :cowboy.start_http(EthereumJSONRPC.WebSocket.Cowboy, 100, [port: port], env: [dispatch: dispatch]) |
||||
port |
||||
end |
||||
|
||||
defp example_state(_) do |
||||
%{state: %WebSocketClient{url: "ws://example.com"}} |
||||
end |
||||
|
||||
defp put_registration(%WebSocketClient{} = state, %Registration{request: %{id: request_id}} = registration) do |
||||
%WebSocketClient{state | request_id_to_registration: %{request_id => registration}} |
||||
end |
||||
|
||||
defp put_registration(%WebSocketClient{} = state, map) when is_map(map) do |
||||
put_registration(state, registration(map)) |
||||
end |
||||
|
||||
defp put_subscription(%WebSocketClient{url: url} = state, subscription_id) when is_integer(subscription_id) do |
||||
subscription_reference = make_ref() |
||||
put_subscription(state, subscription_id, subscription(%{url: url, reference: subscription_reference})) |
||||
end |
||||
|
||||
defp put_subscription( |
||||
%WebSocketClient{url: url} = state, |
||||
subscription_id, |
||||
%Subscription{ |
||||
reference: subscription_reference, |
||||
transport_options: %EthereumJSONRPC.WebSocket{url: url} |
||||
} = subscription |
||||
) do |
||||
%WebSocketClient{ |
||||
state |
||||
| subscription_id_to_subscription_reference: %{subscription_id => subscription_reference}, |
||||
subscription_reference_to_subscription: %{subscription_reference => subscription}, |
||||
subscription_reference_to_subscription_id: %{subscription_reference => subscription_id} |
||||
} |
||||
end |
||||
|
||||
defp registration(%{type: :subscribe = type}) do |
||||
%Registration{ |
||||
type: type, |
||||
from: {self(), make_ref()}, |
||||
request: %{id: unique_request_id(), method: "eth_subscribe", params: ["newHeads"]} |
||||
} |
||||
end |
||||
|
||||
defp registration(%{type: :unsubscribe = type, subscription_id: subscription_id}) do |
||||
%Registration{ |
||||
type: type, |
||||
from: {self(), make_ref()}, |
||||
request: %{id: unique_request_id(), method: "eth_unsubscribe", params: [subscription_id]} |
||||
} |
||||
end |
||||
|
||||
defp registration(%{type: type, method: method, params: params}) do |
||||
%Registration{ |
||||
type: type, |
||||
from: {self(), make_ref()}, |
||||
request: %{id: unique_request_id(), method: method, params: params} |
||||
} |
||||
end |
||||
|
||||
defp subscription(%{reference: reference, url: url}) do |
||||
%Subscription{ |
||||
reference: reference, |
||||
subscriber_pid: self(), |
||||
transport: EthereumJSONRPC.WebSocket, |
||||
transport_options: %EthereumJSONRPC.WebSocket{ |
||||
url: url, |
||||
web_socket: WebSocketClient, |
||||
web_socket_options: %WebSocketClient.Options{ |
||||
web_socket: self(), |
||||
event: "newHeads", |
||||
params: [] |
||||
} |
||||
} |
||||
} |
||||
end |
||||
end |
||||
@ -0,0 +1,71 @@ |
||||
# See https://github.com/ninenines/cowboy/blob/1.1.x/examples/websocket/src/ws_handler.erl |
||||
defmodule EthereumJSONRPC.WebSocket.Cowboy.WebSocketHandler do |
||||
@behaviour :cowboy_websocket_handler |
||||
|
||||
defstruct subscription_id_set: MapSet.new(), |
||||
new_heads_timer_reference: nil |
||||
|
||||
def init({:tcp, :http}, _request, _opts) do |
||||
{:upgrade, :protocol, :cowboy_websocket} |
||||
end |
||||
|
||||
@impl :cowboy_websocket_handler |
||||
def websocket_init(_transport_name, request, _opts) do |
||||
{:ok, request, %__MODULE__{}} |
||||
end |
||||
|
||||
@impl :cowboy_websocket_handler |
||||
def websocket_handle( |
||||
{:text, text}, |
||||
request, |
||||
%__MODULE__{subscription_id_set: subscription_id_set, new_heads_timer_reference: new_heads_timer_reference} = |
||||
state |
||||
) do |
||||
json = Jason.decode!(text) |
||||
|
||||
case json do |
||||
%{"id" => id, "method" => "eth_subscribe", "params" => ["newHeads"]} -> |
||||
subscription_id = :erlang.unique_integer() |
||||
response = %{id: id, result: subscription_id} |
||||
frame = {:text, Jason.encode!(response)} |
||||
|
||||
new_heads_timer_reference = |
||||
case new_heads_timer_reference do |
||||
nil -> |
||||
{:ok, timer_reference} = :timer.send_interval(10, :new_head) |
||||
timer_reference |
||||
|
||||
_ -> |
||||
new_heads_timer_reference |
||||
end |
||||
|
||||
{:reply, frame, request, |
||||
%__MODULE__{ |
||||
state |
||||
| new_heads_timer_reference: new_heads_timer_reference, |
||||
subscription_id_set: MapSet.put(subscription_id_set, subscription_id) |
||||
}} |
||||
|
||||
%{"id" => id, "method" => "echo", "params" => params} -> |
||||
response = %{id: id, result: params} |
||||
frame = {:text, Jason.encode!(response)} |
||||
{:reply, frame, request, state} |
||||
end |
||||
end |
||||
|
||||
@impl :cowboy_websocket_handler |
||||
def websocket_info(:new_head, request, %__MODULE__{subscription_id_set: subscription_id_set} = state) do |
||||
frames = |
||||
Enum.map(subscription_id_set, fn subscription_id -> |
||||
response = %{method: "eth_subscription", params: %{result: %{}, subscription: subscription_id}} |
||||
{:text, Jason.encode!(response)} |
||||
end) |
||||
|
||||
{:reply, frames, request, state} |
||||
end |
||||
|
||||
@impl :cowboy_websocket_handler |
||||
def websocket_terminate(_reason, _request, _state) do |
||||
:ok |
||||
end |
||||
end |
||||
Loading…
Reference in new issue