Custom transport mode

client.go

@@ -7,6 +7,7 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
+	"io"
 	"net"
 	"net/http"
 	"net/url"
@@ -129,6 +130,62 @@ func NewMergeClient(ctx context.Context, addr string, namespace string, outs []i
 
 }
 
+// NewCustomClient is like NewMergeClient in single-request (http) mode, except it allows for a custom doRequest function
+func NewCustomClient(namespace string, outs []interface{}, doRequest func(ctx context.Context, body []byte) (io.ReadCloser, error), opts ...Option) (ClientCloser, error) {
+	config := defaultConfig()
+	for _, o := range opts {
+		o(&config)
+	}
+
+	c := client{
+		namespace:     namespace,
+		paramEncoders: config.paramEncoders,
+		errors:        config.errors,
+	}
+
+	stop := make(chan struct{})
+	c.exiting = stop
+
+	c.doRequest = func(ctx context.Context, cr clientRequest) (clientResponse, error) {
+		b, err := json.Marshal(&cr.req)
+		if err != nil {
+			return clientResponse{}, xerrors.Errorf("marshalling request: %w", err)
+		}
+
+		rawResp, err := doRequest(ctx, b)
+		if err != nil {
+			return clientResponse{}, xerrors.Errorf("doRequest failed: %w", err)
+		}
+
+		defer rawResp.Close()
+
+		var resp clientResponse
+		if cr.req.ID != nil { // non-notification
+			if err := json.NewDecoder(rawResp).Decode(&resp); err != nil {
+				return clientResponse{}, xerrors.Errorf("unmarshaling response: %w", err)
+			}
+
+			if resp.ID, err = normalizeID(resp.ID); err != nil {
+				return clientResponse{}, xerrors.Errorf("failed to response ID: %w", err)
+			}
+
+			if resp.ID != cr.req.ID {
+				return clientResponse{}, xerrors.New("request and response id didn't match")
+			}
+		}
+
+		return resp, nil
+	}
+
+	if err := c.provide(outs); err != nil {
+		return nil, err
+	}
+
+	return func() {
+		close(stop)
+	}, nil
+}
+
 func httpClient(ctx context.Context, addr string, namespace string, outs []interface{}, requestHeader http.Header, config Config) (ClientCloser, error) {
 	c := client{
 		namespace:     namespace,

server.go

@@ -104,6 +104,10 @@ func (s *RPCServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	s.handleReader(ctx, r.Body, w, rpcError)
 }
 
+func (s *RPCServer) HandleRequest(ctx context.Context, r io.Reader, w io.Writer) {
+	s.handleReader(ctx, r, w, rpcError)
+}
+
 func rpcError(wf func(func(io.Writer)), req *request, code ErrorCode, err error) {
 	log.Errorf("RPC Error: %s", err)
 	wf(func(w io.Writer) {