Support client call handlers

Author: magik6k

client.go

@@ -226,13 +226,22 @@ func websocketClient(ctx context.Context, addr string, namespace string, outs []
 	exiting := make(chan struct{})
 	c.exiting = exiting
 
+	var hnd reqestHandler
+	if len(config.reverseHandlers) > 0 {
+		h := makeHandler(defaultServerConfig())
+		for _, reverseHandler := range config.reverseHandlers {
+			h.register(reverseHandler.ns, reverseHandler.hnd)
+		}
+		hnd = h
+	}
+
 	go (&wsConn{
 		conn:             conn,
 		connFactory:      connFactory,
 		reconnectBackoff: config.reconnectBackoff,
 		pingInterval:     config.pingInterval,
 		timeout:          config.timeout,
-		handler:          nil,
+		handler:          hnd,
 		requests:         requests,
 		stop:             stop,
 		exiting:          exiting,

handler.go

@@ -20,7 +20,8 @@ import (
 	"github.com/filecoin-project/go-jsonrpc/metrics"
 )
 
-type rpcHandler struct {
+// methodHandler is a handler for a single method
+type methodHandler struct {
 	paramReceivers []reflect.Type
 	nParams        int
 
@@ -94,9 +95,34 @@ type response struct {
 	Error   *respError  `json:"error,omitempty"`
 }
 
+type handler struct {
+	methods map[string]methodHandler
+	errors  *Errors
+
+	maxRequestSize int64
+
+	// aliasedMethods contains a map of alias:original method names.
+	// These are used as fallbacks if a method is not found by the given method name.
+	aliasedMethods map[string]string
+
+	paramDecoders map[reflect.Type]ParamDecoder
+}
+
+func makeHandler(sc ServerConfig) *handler {
+	return &handler{
+		methods: make(map[string]methodHandler),
+		errors:  sc.errors,
+
+		aliasedMethods: map[string]string{},
+		paramDecoders:  sc.paramDecoders,
+
+		maxRequestSize: sc.maxRequestSize,
+	}
+}
+
 // Register
 
-func (s *RPCServer) register(namespace string, r interface{}) {
+func (s *handler) register(namespace string, r interface{}) {
 	val := reflect.ValueOf(r)
 	// TODO: expect ptr
 
@@ -117,7 +143,7 @@ func (s *RPCServer) register(namespace string, r interface{}) {
 
 		valOut, errOut, _ := processFuncOut(funcType)
 
-		s.methods[namespace+"."+method.Name] = rpcHandler{
+		s.methods[namespace+"."+method.Name] = methodHandler{
 			paramReceivers: recvs,
 			nParams:        ins,
 
@@ -137,7 +163,7 @@ func (s *RPCServer) register(namespace string, r interface{}) {
 type rpcErrFunc func(w func(func(io.Writer)), req *request, code ErrorCode, err error)
 type chanOut func(reflect.Value, interface{}) error
 
-func (s *RPCServer) handleReader(ctx context.Context, r io.Reader, w io.Writer, rpcError rpcErrFunc) {
+func (s *handler) handleReader(ctx context.Context, r io.Reader, w io.Writer, rpcError rpcErrFunc) {
 	wf := func(cb func(io.Writer)) {
 		cb(w)
 	}
@@ -194,7 +220,7 @@ func doCall(methodName string, f reflect.Value, params []reflect.Value) (out []r
 	return out, nil
 }
 
-func (s *RPCServer) getSpan(ctx context.Context, req request) (context.Context, *trace.Span) {
+func (s *handler) getSpan(ctx context.Context, req request) (context.Context, *trace.Span) {
 	if req.Meta == nil {
 		return ctx, nil
 	}
@@ -221,7 +247,7 @@ func (s *RPCServer) getSpan(ctx context.Context, req request) (context.Context,
 	return ctx, span
 }
 
-func (s *RPCServer) createError(err error) *respError {
+func (s *handler) createError(err error) *respError {
 	var code ErrorCode = 1
 	if s.errors != nil {
 		c, ok := s.errors.byType[reflect.TypeOf(err)]
@@ -245,7 +271,7 @@ func (s *RPCServer) createError(err error) *respError {
 	return out
 }
 
-func (s *RPCServer) handle(ctx context.Context, req request, w func(func(io.Writer)), rpcError rpcErrFunc, done func(keepCtx bool), chOut chanOut) {
+func (s *handler) handle(ctx context.Context, req request, w func(func(io.Writer)), rpcError rpcErrFunc, done func(keepCtx bool), chOut chanOut) {
 	// Not sure if we need to sanitize the incoming req.Method or not.
 	ctx, span := s.getSpan(ctx, req)
 	ctx, _ = tag.New(ctx, tag.Insert(metrics.RPCMethod, req.Method))

options.go

@@ -10,6 +10,11 @@ import (
 
 type ParamEncoder func(reflect.Value) (reflect.Value, error)
 
+type clientHandler struct {
+	ns  string
+	hnd interface{}
+}
+
 type Config struct {
 	reconnectBackoff backoff
 	pingInterval     time.Duration
@@ -18,6 +23,8 @@ type Config struct {
 	paramEncoders map[reflect.Type]ParamEncoder
 	errors        *Errors
 
+	reverseHandlers []clientHandler
+
 	httpClient *http.Client
 
 	noReconnect      bool
@@ -81,6 +88,12 @@ func WithErrors(es Errors) func(c *Config) {
 	}
 }
 
+func WithClientHandler(ns string, hnd interface{}) func(c *Config) {
+	return func(c *Config) {
+		c.reverseHandlers = append(c.reverseHandlers, clientHandler{ns, hnd})
+	}
+}
+
 func WithHTTPClient(h *http.Client) func(c *Config) {
 	return func(c *Config) {
 		c.httpClient = h

options_server.go

@@ -8,6 +8,7 @@ import (
 	"golang.org/x/xerrors"
 )
 
+// note: we embed reflect.Type because proxy-structs are not comparable
 type jsonrpcReverseClient struct{ reflect.Type }
 
 type ParamDecoder func(ctx context.Context, json []byte) (reflect.Value, error)
@@ -57,6 +58,8 @@ func WithServerPingInterval(d time.Duration) ServerOption {
 	}
 }
 
+// WithReverseClient will allow extracting reverse client on **WEBSOCKET** calls.
+// RP is a proxy-struct type, much like the one passed to NewClient.
 func WithReverseClient[RP any](namespace string) ServerOption {
 	return func(c *ServerConfig) {
 		c.reverseClientBuilder = func(ctx context.Context, conn *wsConn) (context.Context, error) {
@@ -86,6 +89,11 @@ func WithReverseClient[RP any](namespace string) ServerOption {
 	}
 }
 
+// ExtractReverseClient will extract reverse client from context. Reverse client for the type
+// will only be present if the server was constructed with a matching WithReverseClient option
+// and the connection was a websocket connection.
+// If there is no reverse client, the call will return a zero value and `false`. Otherwise a reverse
+// client and `true` will be returned.
 func ExtractReverseClient[C any](ctx context.Context) (C, bool) {
 	c, ok := ctx.Value(jsonrpcReverseClient{reflect.TypeOf(new(C)).Elem()}).(*C)
 	if !ok {

rpc_test.go

@@ -1142,8 +1142,8 @@ func TestIDHandling(t *testing.T) {
 	}
 }
 
-// 1. make server call on client
-// 2. make client handle
+// 1. make server call on client **
+// 2. make client handle **
 // 3. alias on client
 // 4. alias call on server
 // 6. custom/object param type
@@ -1174,11 +1174,17 @@ type RevCallTestClientProxy struct {
 	CallOnClient func(int) (int, error)
 }
 
-func TestServerCallUser(t *testing.T) {
+type RevCallTestClientHandler struct {
+}
+
+func (h *RevCallTestClientHandler) CallOnClient(a int) (int, error) {
+	return a * 2, nil
+}
+
+func TestReverseCall(t *testing.T) {
 	// setup server
 
 	rpcServer := NewServer(WithReverseClient[RevCallTestClientProxy]("Client"))
-
 	rpcServer.Register("Server", &RevCallTestServerHandler{})
 
 	// httptest stuff
@@ -1192,9 +1198,11 @@ func TestServerCallUser(t *testing.T) {
 	}
 	closer, err := NewMergeClient(context.Background(), "ws://"+testServ.Listener.Addr().String(), "Server", []interface{}{
 		&client,
-	}, nil)
+	}, nil, WithClientHandler("Client", &RevCallTestClientHandler{}))
 	require.NoError(t, err)
 
+	// do the call!
+
 	e := client.Call()
 	require.NoError(t, e)
 

server.go

@@ -5,7 +5,6 @@ import (
 	"encoding/json"
 	"io"
 	"net/http"
-	"reflect"
 	"strings"
 	"time"
 
@@ -20,19 +19,10 @@ const (
 
 // RPCServer provides a jsonrpc 2.0 http server handler
 type RPCServer struct {
-	methods map[string]rpcHandler
-	errors  *Errors
-
-	// aliasedMethods contains a map of alias:original method names.
-	// These are used as fallbacks if a method is not found by the given method name.
-	aliasedMethods map[string]string
-
-	paramDecoders map[reflect.Type]ParamDecoder
-
+	*handler
 	reverseClientBuilder func(context.Context, *wsConn) (context.Context, error)
 
 	pingInterval   time.Duration
-	maxRequestSize int64
 }
 
 // NewServer creates new RPCServer instance
@@ -43,12 +33,8 @@ func NewServer(opts ...ServerOption) *RPCServer {
 	}
 
 	return &RPCServer{
-		methods:              map[string]rpcHandler{},
-		aliasedMethods:       map[string]string{},
-		paramDecoders:        config.paramDecoders,
+		handler:              makeHandler(config),
 		reverseClientBuilder: config.reverseClientBuilder,
-		maxRequestSize:       config.maxRequestSize,
-		errors:               config.errors,
 
 		pingInterval: config.pingInterval,
 	}

websocket.go

@@ -45,14 +45,18 @@ type outChanReg struct {
 	ch   reflect.Value
 }
 
+type reqestHandler interface {
+	handle(ctx context.Context, req request, w func(func(io.Writer)), rpcError rpcErrFunc, done func(keepCtx bool), chOut chanOut)
+}
+
 type wsConn struct {
 	// outside params
 	conn             *websocket.Conn
 	connFactory      func() (*websocket.Conn, error)
 	reconnectBackoff backoff
 	pingInterval     time.Duration
 	timeout          time.Duration
-	handler          *RPCServer
+	handler          reqestHandler
 	requests         <-chan clientRequest
 	pongs            chan struct{}
 	stopPings        func()