Any

codec/codec.go

@@ -1,17 +1,26 @@
 package codec
 
 import (
-	"github.com/pentops/j5/lib/j5schema"
+	"github.com/pentops/j5/lib/j5reflect"
 	"google.golang.org/protobuf/reflect/protoreflect"
+	"google.golang.org/protobuf/reflect/protoregistry"
 )
 
+// MessageTypeResolver is a subset of protoregistry.MessageTypeResolver
+type MessageTypeResolver interface {
+	FindMessageByName(message protoreflect.FullName) (protoreflect.MessageType, error)
+}
+
 type Codec struct {
-	schemaSet *j5schema.SchemaCache
+	refl     *j5reflect.Reflector
+	resolver MessageTypeResolver
 }
 
-func NewCodec() *Codec {
+func NewCodec(resolver protoregistry.MessageTypeResolver) *Codec {
+	refl := j5reflect.New()
 	return &Codec{
-		schemaSet: j5schema.NewSchemaCache(),
+		refl:     refl,
+		resolver: resolver,
 	}
 }
 

codec/codec_test.go

@@ -12,7 +12,9 @@ import (
 	"google.golang.org/protobuf/encoding/prototext"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/reflect/protoreflect"
+	"google.golang.org/protobuf/reflect/protoregistry"
 	"google.golang.org/protobuf/types/dynamicpb"
+	"google.golang.org/protobuf/types/known/anypb"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"github.com/pentops/flowtest/prototest"
@@ -21,17 +23,18 @@ import (
 	"github.com/pentops/j5/j5types/decimal_j5t"
 )
 
-/*
-	func mustAny(t testing.TB, msg proto.Message) *anypb.Any {
-		a, err := anypb.New(msg)
-		if err != nil {
-			t.Fatal(err)
-		}
-		return a
+func mustAny(t testing.TB, msg proto.Message) *anypb.Any {
+	a, err := anypb.New(msg)
+	if err != nil {
+		t.Fatal(err)
 	}
-*/
+	return a
+}
+
 func TestUnmarshal(t *testing.T) {
 
+	codec := NewCodec(protoregistry.GlobalTypes)
+
 	testTime := time.Date(2020, 1, 1, 0, 0, 0, 0, time.UTC)
 
 	for _, tc := range []struct {
@@ -339,13 +342,12 @@ func TestUnmarshal(t *testing.T) {
 			wantProto: &schema_testpb.FullSchema{
 				KeyString: "keyVal",
 			},
-		},
-		/*{
+		}, {
 			name: "any",
 			json: `{
 				"any": {
-					"!type": "test.v1.Bar",
-					"bar": {
+					"!type": "test.schema.v1.Bar",
+					"value": {
 						"barId": "barId"
 					}
 				}
@@ -355,37 +357,29 @@ func TestUnmarshal(t *testing.T) {
 					BarId: "barId",
 				}),
 			},
-		}*/} {
+		}} {
 		t.Run(tc.name, func(t *testing.T) {
 
 			allInputs := append(tc.altInputJSON, tc.json)
 
-			codec := NewCodec()
 			for _, input := range allInputs {
-				/*
-					schema, err := codec.schemaSet.SchemaObject(tc.wantProto.ProtoReflect().Descriptor())
-					if err != nil {
-						t.Fatal(err)
-					}
-					t.Logf("SCHEMA: %s", protojson.Format(schema))
-				*/
 				logIndent(t, "input", input)
 
 				msg := tc.wantProto.ProtoReflect().New().Interface()
 				if err := codec.JSONToProto([]byte(input), msg.ProtoReflect()); err != nil {
 					t.Fatalf("JSONToProto: %s", err)
 				}
 
-				t.Logf("got decoded proto: %s \n%v\n", msg.ProtoReflect().Descriptor().FullName(), prototext.Format(msg))
+				t.Logf("GOT proto: %s \n%v\n", msg.ProtoReflect().Descriptor().FullName(), prototext.Format(msg))
 
 				if !proto.Equal(tc.wantProto, msg) {
 					a := prototext.Format(tc.wantProto)
-					t.Fatalf("expected proto %s\n%v\n", tc.wantProto.ProtoReflect().Descriptor().FullName(), string(a))
+					t.Fatalf("FATAL: Expected proto %s\n%v\n", tc.wantProto.ProtoReflect().Descriptor().FullName(), string(a))
 				}
 
 				encoded, err := codec.ProtoToJSON(msg.ProtoReflect())
 				if err != nil {
-					t.Fatalf("ProtoToJSON: %s", err)
+					t.Fatalf("FATAL: ProtoToJSON: %s", err)
 				}
 
 				logIndent(t, "output", string(encoded))
@@ -419,7 +413,7 @@ func TestScalars(t *testing.T) {
 
 	runTest := func(t testing.TB, tc testCase) {
 
-		codec := NewCodec()
+		codec := NewCodec(protoregistry.GlobalTypes)
 
 		msgIn := dynamicpb.NewMessage(tc.desc)
 

codec/decoder.go

@@ -7,7 +7,10 @@ import (
 	"strings"
 
 	"github.com/pentops/j5/lib/j5reflect"
+	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/reflect/protoreflect"
+	"google.golang.org/protobuf/reflect/protoregistry"
+	"google.golang.org/protobuf/types/known/anypb"
 )
 
 func (c *Codec) decode(jsonData []byte, msg protoreflect.Message) error {
@@ -18,7 +21,7 @@ func (c *Codec) decode(jsonData []byte, msg protoreflect.Message) error {
 		codec: c,
 	}
 
-	root, err := j5reflect.NewWithCache(c.schemaSet).NewRoot(msg)
+	root, err := c.refl.NewRoot(msg)
 	if err != nil {
 		return err
 	}
@@ -33,6 +36,7 @@ func (c *Codec) decode(jsonData []byte, msg protoreflect.Message) error {
 	}
 }
 
+// decoder is an instance for decoding a single message, not reusable.
 type decoder struct {
 	jd    *json.Decoder
 	codec *Codec
@@ -180,16 +184,7 @@ func (dec *decoder) decodeValue(field j5reflect.Field) error {
 		return dec.decodeEnum(ft)
 
 	case j5reflect.ScalarField:
-		tok, err := dec.Token()
-		if err != nil {
-			return err
-		}
-
-		if _, ok := tok.(json.Delim); ok {
-			return unexpectedTokenError(tok, "scalar")
-		}
-
-		return ft.SetGoValue(tok)
+		return dec.decodeScalar(ft)
 
 	case j5reflect.AnyField:
 		return dec.decodeAny(ft)
@@ -199,6 +194,19 @@ func (dec *decoder) decodeValue(field j5reflect.Field) error {
 	}
 }
 
+func (dec *decoder) decodeScalar(field j5reflect.ScalarField) error {
+	tok, err := dec.Token()
+	if err != nil {
+		return err
+	}
+
+	if _, ok := tok.(json.Delim); ok {
+		return unexpectedTokenError(tok, "scalar")
+	}
+
+	return field.SetGoValue(tok)
+}
+
 func (dec *decoder) decodeEnum(field j5reflect.EnumField) error {
 	token, err := dec.Token()
 	if err != nil {
@@ -257,11 +265,37 @@ func (dec *decoder) decodeAny(field j5reflect.AnyField) error {
 		return newFieldError("value", "no value found in Any")
 	}
 
-	// This code assumes the schema has been pre-loaded
+	// takes the PROTO name, which should match the encoder.
+	innerDesc, err := dec.codec.resolver.FindMessageByName(protoreflect.FullName(*constrainType))
+	if err != nil {
+		if err == protoregistry.NotFound {
+			return newFieldError(*constrainType, fmt.Sprintf("no type %q in registry", *constrainType))
+		}
+		return newFieldError(*constrainType, err.Error())
+	}
+	msg := innerDesc.New()
+
+	if err := dec.codec.decode(valueBytes, msg); err != nil {
+		return newFieldError(*constrainType, err.Error())
+	}
+
+	protoBytes, err := proto.Marshal(msg.Interface())
+	if err != nil {
+		return newFieldError(*constrainType, err.Error())
+	}
+
+	anyVal := &anypb.Any{
+		Value:   protoBytes,
+		TypeUrl: anyPrefix + *constrainType,
+	}
+
+	field.SetProtoAny(anyVal)
 
 	return nil
 }
 
+const anyPrefix = "type.googleapis.com/"
+
 func (dec *decoder) decodeOneof(oneof j5reflect.Oneof) error {
 
 	foundKeys := []string{}

codec/encoder.go

@@ -11,10 +11,11 @@ import (
 
 func (c *Codec) encode(msg protoreflect.Message) ([]byte, error) {
 	enc := &encoder{
-		b: &bytes.Buffer{},
+		codec: c,
+		b:     &bytes.Buffer{},
 	}
 
-	root, err := j5reflect.NewWithCache(c.schemaSet).NewRoot(msg)
+	root, err := c.refl.NewRoot(msg)
 	if err != nil {
 		return nil, err
 	}
@@ -35,7 +36,8 @@ func (c *Codec) encode(msg protoreflect.Message) ([]byte, error) {
 }
 
 type encoder struct {
-	b *bytes.Buffer
+	b     *bytes.Buffer
+	codec *Codec
 }
 
 func (enc *encoder) add(b []byte) {

codec/structure_encode.go

@@ -3,6 +3,7 @@ package codec
 import (
 	"encoding/base64"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/pentops/j5/j5types/date_j5t"
@@ -73,6 +74,45 @@ func (enc *encoder) encodeObject(object j5reflect.Object) error {
 	return enc.encodeObjectBody(object)
 }
 
+func (enc *encoder) encodeAny(anyField j5reflect.AnyField) error {
+	protoAny := anyField.GetProtoAny()
+	msg, err := protoAny.UnmarshalNew()
+	if err != nil {
+		return err
+	}
+
+	innerBytes, err := enc.codec.encode(msg.ProtoReflect())
+	if err != nil {
+		return err
+	}
+
+	enc.openObject()
+	defer enc.closeObject()
+
+	err = enc.fieldLabel("!type")
+	if err != nil {
+		return err
+	}
+
+	typeName := strings.TrimPrefix(protoAny.TypeUrl, anyPrefix)
+	err = enc.addString(typeName)
+	if err != nil {
+		return err
+	}
+
+	enc.fieldSep()
+
+	err = enc.fieldLabel("value")
+	if err != nil {
+		return err
+	}
+
+	enc.add(innerBytes)
+
+	return nil
+
+}
+
 func (enc *encoder) encodeValue(field j5reflect.Field) error {
 
 	switch ft := field.(type) {
@@ -94,6 +134,9 @@ func (enc *encoder) encodeValue(field j5reflect.Field) error {
 	case j5reflect.ScalarField:
 		return enc.encodeScalarField(ft)
 
+	case j5reflect.AnyField:
+		return enc.encodeAny(ft)
+
 	default:
 		return fmt.Errorf("encode value of type %q, unsupported", field.FullTypeName())
 	}