@ -9,15 +9,38 @@ defmodule EthereumJSONRPC.WebSocket.WebSocketClient do
alias EthereumJSONRPC . { Subscription , Transport , WebSocket } alias EthereumJSONRPC . { Subscription , Transport , WebSocket }
alias EthereumJSONRPC.WebSocket.Registration alias EthereumJSONRPC.WebSocket.Registration
alias EthereumJSONRPC.WebSocket.WebSocketClient.Options
@behaviour :websocket_client @behaviour :websocket_client
@behaviour WebSocket @behaviour WebSocket
@enforce_keys ~w( url )a @enforce_keys ~w( url )a
defstruct request_id_to_registration : %{ } , defstruct connected : false ,
subscription_id_to_subscription : %{ } , request_id_to_registration : %{ } ,
subscription_id_to_subscription_reference : %{ } ,
subscription_reference_to_subscription_id : %{ } ,
subscription_reference_to_subscription : %{ } ,
url : nil url : nil
@typedoc """
* ` request_id_to_registration ` - maps id of requests in flight to their
` t :EthereumSJONRPC.WebSocket.Registration . t / 0 ` , so that when the response is received from the server , the caller
in ` from ` of the registration can be ` GenServer . reply / 2 ` ed to .
* ` subscription_id_to_subscription_reference ` - maps id of subscription on the server to the ` t :reference / 0 ` used in
the ` t :EthereumJSONRPC.Subscription . t / 0 ` . Subscriptions use a ` t :reference / 0 ` instead of the server - side id , so
that on reconnect , the id can change , but the subscribe does not need to be notified .
* ` subscription_reference_to_subscription ` - maps ` t :reference / 0 ` in ` t :EthereumJSONRPC.Subscription . t / 0 ` to that
` t :EthereumJSONRPC.Subscription . t / 0 ` , so that the ` subscriber_pid ` can be notified of subscription messages .
* ` subscription_reference_to_subscription_id ` - maps ` t :reference / 0 ` in ` t :EthereumJSONRPC.Subscription . t / 0 to id of
the subscription on the server , so that the subscriber can unsubscribe with the ` t :reference / 0 ` .
"""
@type t :: % __MODULE__ {
request_id_to_registration : %{ EthereumJSONRPC . request_id ( ) = > Registration . t ( ) } ,
subscription_id_to_subscription_reference : %{ Subscription . id ( ) = > reference ( ) } ,
subscription_reference_to_subscription : %{ reference ( ) = > Subscription . t ( ) } ,
subscription_reference_to_subscription_id : %{ reference ( ) = > Subscription . id ( ) }
}
# Supervisor interface # Supervisor interface
@impl WebSocket @impl WebSocket
@ -98,13 +121,15 @@ defmodule EthereumJSONRPC.WebSocket.WebSocketClient do
end end
@impl :websocket_client @impl :websocket_client
def onconnect ( _ , % __MODULE__ { } = state ) do def onconnect ( _ , % __MODULE__ { connected : false } = state ) do
{ :ok , state } { :ok , reconnect ( % __MODULE__ { state | connected : true } ) }
end end
@impl :websocket_client @impl :websocket_client
def ondisconnect ( reason , % __MODULE__ { } = state ) do def ondisconnect ( _reason , % __MODULE__ { request_id_to_registration : request_id_to_registration } = state ) do
{ :close , reason , state } final_state = Enum . reduce ( request_id_to_registration , state , & disconnect_request_id_registration / 2 )
{ :reconnect , % __MODULE__ { final_state | connected : false } }
end end
@impl :websocket_client @impl :websocket_client
@ -121,7 +146,10 @@ defmodule EthereumJSONRPC.WebSocket.WebSocketClient do
@impl :websocket_client @impl :websocket_client
def websocket_info ( { :" $gen_call " , from , request } , _ , % __MODULE__ { } = state ) do def websocket_info ( { :" $gen_call " , from , request } , _ , % __MODULE__ { } = state ) do
handle_call ( request , from , state ) case handle_call ( request , from , state ) do
{ :reply , _ , % __MODULE__ { } } = reply -> reply
{ :noreply , % __MODULE__ { } = new_state } -> { :ok , new_state }
end
end end
@impl :websocket_client @impl :websocket_client
@ -129,24 +157,79 @@ defmodule EthereumJSONRPC.WebSocket.WebSocketClient do
broadcast ( close , state ) broadcast ( close , state )
end end
defp broadcast ( message , % __MODULE__ { subscription_id_to_subscription : id _to_subscription } ) do defp broadcast ( message , % __MODULE__ { subscription_reference_to_subscription : subscription_reference _to_subscription } ) do
id _to_subscription subscription_reference _to_subscription
|> Map . values ( ) |> Map . values ( )
|> Subscription . broadcast ( message ) |> Subscription . broadcast ( message )
end end
defp handle_call ( message , from , % __MODULE__ { } = state ) do # Not re-subscribing after disconnect is the same as a successful unsubscribe
{ updated_state , unique_request } = register ( message , from , state ) defp disconnect_request_id_registration (
{ request_id ,
% Registration {
type : :unsubscribe ,
from : from ,
request : %{ method : " eth_unsubscribe " , params : [ subscription_id ] }
} } ,
% __MODULE__ {
request_id_to_registration : request_id_to_registration ,
subscription_id_to_subscription_reference : subscription_id_to_subscription_reference ,
subscription_reference_to_subscription : subscription_reference_to_subscription ,
subscription_reference_to_subscription_id : subscription_reference_to_subscription_id
} = acc_state
) do
GenServer . reply ( from , :ok )
%{ ^ subscription_id = > subscription_reference } = subscription_id_to_subscription_reference
% __MODULE__ {
acc_state
| request_id_to_registration : Map . delete ( request_id_to_registration , request_id ) ,
subscription_id_to_subscription_reference :
Map . delete ( subscription_id_to_subscription_reference , subscription_id ) ,
subscription_reference_to_subscription :
Map . delete ( subscription_reference_to_subscription , subscription_reference ) ,
subscription_reference_to_subscription_id :
Map . delete ( subscription_reference_to_subscription_id , subscription_reference )
}
end
# Re-run in `onconnect\2`
defp disconnect_request_id_registration ( % Registration { type : type } , state ) when type in ~w( json_rpc subscribe )a do
state
end
defp frame ( request ) do
{ :text , Jason . encode! ( request ) }
end
defp handle_call ( message , from , % __MODULE__ { connected : connected } = state ) do
case register ( message , from , state ) do
{ :ok , unique_request , updated_state } ->
case connected do
true ->
{ :reply , frame ( unique_request ) , updated_state }
false ->
{ :noreply , updated_state }
end
{ :reply , { :text , Jason . encode! ( unique_request ) } , updated_state } { :error , _reason } = error ->
GenServer . reply ( from , error )
{ :noreply , state }
end
end end
defp handle_response ( defp handle_response (
%{ " method " = > " eth_subscription " , " params " = > %{ " result " = > result , " subscription " = > subscription_id } } , %{ " method " = > " eth_subscription " , " params " = > %{ " result " = > result , " subscription " = > subscription_id } } ,
% __MODULE__ { subscription_id_to_subscription : subscription_id_to_subscription } = state % __MODULE__ {
subscription_id_to_subscription_reference : subscription_id_to_subscription_reference ,
subscription_reference_to_subscription : subscription_reference_to_subscription
} = state
) do ) do
case subscription_id_to_subscription do case subscription_id_to_subscription_reference do
%{ ^ subscription_id = > subscription } -> %{ ^ subscription_id = > subscription_reference } ->
%{ ^ subscription_reference = > subscription } = subscription_reference_to_subscription
Subscription . publish ( subscription , { :ok , result } ) Subscription . publish ( subscription , { :ok , result } )
_ -> _ ->
@ -157,7 +240,7 @@ defmodule EthereumJSONRPC.WebSocket.WebSocketClient do
" ) result ( " , " ) result ( " ,
inspect ( result ) , inspect ( result ) ,
" ). Subscription ID not in known subscription IDs ( " , " ). Subscription ID not in known subscription IDs ( " ,
subscription_id_to_subscription subscription_id_to_subscription_reference
|> Map . values ( ) |> Map . values ( )
|> Enum . map ( & inspect / 1 ) , |> Enum . map ( & inspect / 1 ) ,
" ). " " ). "
@ -192,21 +275,30 @@ defmodule EthereumJSONRPC.WebSocket.WebSocketClient do
{ :ok , state } { :ok , state }
end end
defp reconnect ( % __MODULE__ { } = state ) do
state
|> rerequest ( )
|> resubscribe ( )
end
defp register ( defp register (
{ :json_rpc , original_request } , { :json_rpc , original_request } ,
from , from ,
% __MODULE__ { request_id_to_registration : request_id_to_registration } = state % __MODULE__ { request_id_to_registration : request_id_to_registration } = state
) do ) do
unique_id = unique_request_id ( state ) unique_id = unique_request_id ( state )
request = %{ original_request | id : unique_id }
{ % __MODULE__ { { :ok , request ,
% __MODULE__ {
state state
| request_id_to_registration : | request_id_to_registration :
Map . put ( request_id_to_registration , unique_id , % Registration { Map . put ( request_id_to_registration , unique_id , % Registration {
from : from , from : from ,
type : :json_rpc type : :json_rpc ,
request : request
} ) } )
} , %{ original_request | id : unique_id } } } }
end end
defp register ( defp register (
@ -216,36 +308,54 @@ defmodule EthereumJSONRPC.WebSocket.WebSocketClient do
) )
when is_binary ( event ) and is_list ( params ) do when is_binary ( event ) and is_list ( params ) do
unique_id = unique_request_id ( state ) unique_id = unique_request_id ( state )
request = request ( %{ id : unique_id , method : " eth_subscribe " , params : [ event | params ] } )
{ { :ok , request ,
% __MODULE__ { % __MODULE__ {
state state
| request_id_to_registration : | request_id_to_registration :
Map . put ( request_id_to_registration , unique_id , % Registration { from : from , type : :subscribe } ) Map . put ( request_id_to_registration , unique_id , % Registration { from : from , type : :subscribe , request : request } )
} , } }
request ( %{ id : unique_id , method : " eth_subscribe " , params : [ event | params ] } )
}
end end
defp register ( defp register (
{ :unsubscribe , % Subscription { id : subscription_id } } , { :unsubscribe , % Subscription { reference : subscription_reference } } ,
from , from ,
% __MODULE__ { request_id_to_registration : request_id_to_registration } = state % __MODULE__ {
request_id_to_registration : request_id_to_registration ,
subscription_reference_to_subscription_id : subscription_reference_to_subscription_id
} = state
) do ) do
unique_id = unique_request_id ( state ) case subscription_reference_to_subscription_id do
%{ ^ subscription_reference = > subscription_id } ->
unique_id = unique_request_id ( state )
request = request ( %{ id : unique_id , method : " eth_unsubscribe " , params : [ subscription_id ] } )
{
:ok ,
request ,
% __MODULE__ {
state
| request_id_to_registration :
Map . put ( request_id_to_registration , unique_id , % Registration {
from : from ,
type : :unsubscribe ,
request : request
} )
}
}
{ _ ->
% __MODULE__ { { :error , :not_found }
state end
| request_id_to_registration : end
Map . put ( request_id_to_registration , unique_id , % Registration {
from : from , defp rerequest ( % __MODULE__ { request_id_to_registration : request_id_to_registration } = state ) do
type : :unsubscribe , Enum . each ( request_id_to_registration , fn { _ , % Registration { request : request } } ->
subscription_id : subscription_id :websocket_client . cast ( self ( ) , frame ( request ) )
} ) end )
} ,
request ( %{ id : unique_id , method : " eth_unsubscribe " , params : [ subscription_id ] } ) state
}
end end
defp respond_to_registration ( defp respond_to_registration (
@ -265,22 +375,38 @@ defmodule EthereumJSONRPC.WebSocket.WebSocketClient do
end end
defp respond_to_registration ( defp respond_to_registration (
% Registration { type : :subscribe , from : { subscriber_pid , _ } = from } , % Registration { type : :subscribe , from : { subscriber_pid , _ } = from , request : %{ params : [ event | params ] } } ,
%{ " result " = > subscription_id } , %{ " result " = > subscription_id } ,
% __MODULE__ { subscription_id_to_subscription : subscription_id_to_subscription , url : url } = state % __MODULE__ {
subscription_id_to_subscription_reference : subscription_id_to_subscription_reference ,
subscription_reference_to_subscription : subscription_reference_to_subscription ,
subscription_reference_to_subscription_id : subscription_reference_to_subscription_id ,
url : url
} = state
) do ) do
subscription_reference = make_ref ( )
subscription = % Subscription { subscription = % Subscription {
id : subscription_id , reference : subscription_reference ,
subscriber_pid : subscriber_pid , subscriber_pid : subscriber_pid ,
transport : EthereumJSONRPC.WebSocket , transport : EthereumJSONRPC.WebSocket ,
transport_options : [ web_socket : __MODULE__ , web_socket_options : %{ web_socket : self ( ) } , url : url ] transport_options : % EthereumJSONRPC.WebSocket {
web_socket : __MODULE__ ,
web_socket_options : % Options { web_socket : self ( ) , event : event , params : params } ,
url : url
}
} }
GenServer . reply ( from , { :ok , subscription } ) GenServer . reply ( from , { :ok , subscription } )
new_state = % __MODULE__ { new_state = % __MODULE__ {
state state
| subscription_id_to_subscription : Map . put ( subscription_id_to_subscription , subscription_id , subscription ) | subscription_id_to_subscription_reference :
Map . put ( subscription_id_to_subscription_reference , subscription_id , subscription_reference ) ,
subscription_reference_to_subscription :
Map . put ( subscription_reference_to_subscription , subscription_reference , subscription ) ,
subscription_reference_to_subscription_id :
Map . put ( subscription_reference_to_subscription_id , subscription_reference , subscription_id )
} }
{ :ok , new_state } { :ok , new_state }
@ -297,9 +423,17 @@ defmodule EthereumJSONRPC.WebSocket.WebSocketClient do
end end
defp respond_to_registration ( defp respond_to_registration (
% Registration { type : :unsubscribe , from : from , subscription_id : subscription_id } , % Registration {
type : :unsubscribe ,
from : from ,
request : %{ method : " eth_unsubscribe " , params : [ subscription_id ] }
} ,
response , response ,
% __MODULE__ { subscription_id_to_subscription : subscription_id_to_subscription } = state % __MODULE__ {
subscription_id_to_subscription_reference : subscription_id_to_subscription_reference ,
subscription_reference_to_subscription : subscription_reference_to_subscription ,
subscription_reference_to_subscription_id : subscription_reference_to_subscription_id
} = state
) do ) do
reply = reply =
case response do case response do
@ -311,20 +445,77 @@ defmodule EthereumJSONRPC.WebSocket.WebSocketClient do
GenServer . reply ( from , reply ) GenServer . reply ( from , reply )
new_state = % __MODULE__ { new_state =
state case subscription_id_to_subscription_reference do
| subscription_id_to_subscription : Map . delete ( subscription_id_to_subscription , subscription_id ) %{ ^ subscription_id = > subscription_reference } ->
} % __MODULE__ {
state
| subscription_id_to_subscription_reference :
Map . delete ( subscription_id_to_subscription_reference , subscription_id ) ,
subscription_reference_to_subscription :
Map . delete ( subscription_reference_to_subscription , subscription_reference ) ,
subscription_reference_to_subscription_id :
Map . delete ( subscription_reference_to_subscription_id , subscription_reference )
}
_ ->
state
end
{ :ok , new_state } { :ok , new_state }
end end
defp respond_to_registration ( nil , response , % __MODULE__ { } = state ) do defp respond_to_registration (
Logger . error ( fn -> [ " Got response for unregistered request ID: " , inspect ( response ) ] end ) nil ,
response ,
% __MODULE__ { request_id_to_registration : request_id_to_registration } = state
) do
Logger . error ( fn ->
[
" Got response for unregistered request ID: " ,
inspect ( response ) ,
" . Outstanding request registrations: " ,
inspect ( request_id_to_registration )
]
end )
{ :ok , state } { :ok , state }
end end
defp resubscribe (
% __MODULE__ { subscription_reference_to_subscription : subscription_reference_to_subscription } = initial_state
) do
Enum . reduce ( subscription_reference_to_subscription , initial_state , fn { subscription_reference ,
% Subscription {
transport_options : % WebSocket {
web_socket : __MODULE__ ,
web_socket_options : % Options {
event : event ,
params : params
}
}
} } ,
% __MODULE__ {
request_id_to_registration :
acc_request_id_to_registration
} = acc_state ->
request_id = unique_request_id ( acc_state )
request = request ( %{ id : request_id , method : " eth_subscribe " , params : [ event | params ] } )
:websocket_client . cast ( self ( ) , frame ( request ) )
% __MODULE__ {
acc_state
| request_id_to_registration :
Map . put ( acc_request_id_to_registration , request_id , % Registration {
from : { self ( ) , subscription_reference } ,
type : :subscribe ,
request : request
} )
}
end )
end
defp unique_request_id ( % __MODULE__ { request_id_to_registration : request_id_to_registration } = state ) do defp unique_request_id ( % __MODULE__ { request_id_to_registration : request_id_to_registration } = state ) do
unique_request_id = EthereumJSONRPC . unique_request_id ( ) unique_request_id = EthereumJSONRPC . unique_request_id ( )
Luke Imhoff
6 years ago