Add a helper for serving a payload encoder over HTTP. (#717)

* Add a helper for serving a payload encoder over HTTP.

This will be used to replace the data converter plugin infrastructure with a more flexible approach.

* Add RemoteEncoderDataConverter.

This can be used by tctl for example.

converter/encoding_data_converter.go

@@ -25,8 +25,15 @@ package converter
 import (
 	"bytes"
 	"compress/zlib"
+	"encoding/json"
+	"fmt"
+	"io"
 	"io/ioutil"
+	"net/http"
+	"path"
+	"strings"
 
+	"github.com/gogo/protobuf/jsonpb"
 	"github.com/gogo/protobuf/proto"
 	commonpb "go.temporal.io/api/common/v1"
 )
@@ -244,3 +251,127 @@ func partiallyClonePayload(p *commonpb.Payload) *commonpb.Payload {
 	}
 	return ret
 }
+
+const remotePayloadEncoderEncodePath = "/encode"
+const remotePayloadEncoderDecodePath = "/decode"
+
+type encoderHTTPHandler struct {
+	encoder PayloadEncoder
+}
+
+// ServeHTTP implements the http.Handler interface.
+func (e *encoderHTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	if r.Method != "POST" {
+		http.NotFound(w, r)
+		return
+	}
+
+	path := r.URL.Path
+
+	if !strings.HasSuffix(path, remotePayloadEncoderEncodePath) &&
+		!strings.HasSuffix(path, remotePayloadEncoderDecodePath) {
+		http.NotFound(w, r)
+		return
+	}
+
+	var p commonpb.Payload
+
+	if r.Body == nil {
+		http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
+		return
+	}
+
+	err := jsonpb.Unmarshal(r.Body, &p)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	switch {
+	case strings.HasSuffix(path, remotePayloadEncoderEncodePath):
+		err = e.encoder.Encode(&p)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+	case strings.HasSuffix(path, remotePayloadEncoderDecodePath):
+		err = e.encoder.Decode(&p)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+	default:
+		http.NotFound(w, r)
+		return
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	err = json.NewEncoder(w).Encode(p)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+}
+
+// NewPayloadEncoderHTTPHandler creates a http.Handler for a PayloadEncoder.
+// This can be used to provide a remote data converter.
+func NewPayloadEncoderHTTPHandler(e PayloadEncoder) http.Handler {
+	return &encoderHTTPHandler{encoder: e}
+}
+
+// RemotePayloadEncoderOptions are options for NewRemotePayloadEncoder.
+// Client is optional.
+type RemotePayloadEncoderOptions struct {
+	Endpoint string
+	Client   http.Client
+}
+
+type remotePayloadEncoder struct {
+	options RemotePayloadEncoderOptions
+}
+
+// NewRemotePayloadEncoder creates a PayloadEncoder that uses a remote endpoint to encode/decode.
+func NewRemotePayloadEncoder(options RemotePayloadEncoderOptions) PayloadEncoder {
+	return &remotePayloadEncoder{options}
+}
+
+func (rdc *remotePayloadEncoder) sendHTTP(endpoint string, p *commonpb.Payload) error {
+	payload, err := json.Marshal(p)
+	if err != nil {
+		return fmt.Errorf("unable to marshal payload: %w", err)
+	}
+
+	req, err := http.NewRequest(http.MethodPost, endpoint, bytes.NewReader(payload))
+	if err != nil {
+		return fmt.Errorf("unable to build request: %w", err)
+	}
+
+	req.Header.Set("Content-Type", "application/json")
+
+	response, err := rdc.options.Client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = response.Body.Close() }()
+
+	if response.StatusCode == 200 {
+		err = jsonpb.Unmarshal(response.Body, p)
+		if err != nil {
+			return fmt.Errorf("unable to unmarshal payload: %w", err)
+		}
+		return nil
+	}
+
+	message, _ := io.ReadAll(response.Body)
+	return fmt.Errorf("%s: %s", http.StatusText(response.StatusCode), message)
+}
+
+// Encode sends a payload to remote payload encoder and returns the encoded payload.
+func (rdc *remotePayloadEncoder) Encode(p *commonpb.Payload) error {
+	return rdc.sendHTTP(path.Join(rdc.options.Endpoint, remotePayloadEncoderEncodePath), p)
+}
+
+// Decode sends a payload to a remote payload encoder and returns the decoded payload.
+func (rdc *remotePayloadEncoder) Decode(p *commonpb.Payload) error {
+	return rdc.sendHTTP(path.Join(rdc.options.Endpoint, remotePayloadEncoderDecodePath), p)
+}

converter/encoding_data_converter_test.go

@@ -23,7 +23,11 @@
 package converter_test
 
 import (
+	"bytes"
+	"encoding/json"
 	"fmt"
+	"net/http"
+	"net/http/httptest"
 	"reflect"
 	"strings"
 	"testing"
@@ -170,3 +174,61 @@ func (c *captureToPayloadDataConverter) ToPayload(value interface{}) (*commonpb.
 	c.lastToPayloadResult = p
 	return p, err
 }
+
+func TestPayloadEncoderHTTPHandler(t *testing.T) {
+	defaultConv := converter.GetDefaultDataConverter()
+	encoder := converter.NewZlibEncoder(converter.ZlibEncoderOptions{AlwaysEncode: true})
+	handler := converter.NewPayloadEncoderHTTPHandler(encoder)
+
+	req, err := http.NewRequest("GET", "/encode", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	rr := httptest.NewRecorder()
+	handler.ServeHTTP(rr, req)
+
+	require.Equal(t, http.StatusNotFound, rr.Code)
+
+	req, err = http.NewRequest("POST", "/missing", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	rr = httptest.NewRecorder()
+	handler.ServeHTTP(rr, req)
+
+	req, err = http.NewRequest("POST", "/encode", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	rr = httptest.NewRecorder()
+	handler.ServeHTTP(rr, req)
+
+	require.Equal(t, http.StatusBadRequest, rr.Code)
+
+	payload, _ := defaultConv.ToPayload("test")
+	payloadJSON, _ := json.Marshal(payload)
+
+	fmt.Printf("%s", payloadJSON)
+
+	req, err = http.NewRequest("POST", "/encode", bytes.NewReader(payloadJSON))
+	if err != nil {
+		t.Fatal(err)
+	}
+	rr = httptest.NewRecorder()
+	handler.ServeHTTP(rr, req)
+
+	require.Equal(t, http.StatusOK, rr.Code)
+	encodedPayloadJSON := strings.TrimSpace(rr.Body.String())
+	require.NotEqual(t, payloadJSON, encodedPayloadJSON)
+
+	req, err = http.NewRequest("POST", "/decode", strings.NewReader(encodedPayloadJSON))
+	if err != nil {
+		t.Fatal(err)
+	}
+	rr = httptest.NewRecorder()
+	handler.ServeHTTP(rr, req)
+
+	require.Equal(t, http.StatusOK, rr.Code)
+	decodedPayloadJSON := strings.TrimSpace(rr.Body.String())
+	require.Equal(t, string(payloadJSON), decodedPayloadJSON)
+}