tidy up

Author: michal-harish

protocol.go

@@ -0,0 +1,35 @@
+package schema_registry
+
+type newSchemaRequest struct {
+	SchemaType string     `json:"schemaType"`
+	References references `json:"references"`
+	Schema     string     `json:"schema"`
+}
+
+type references []reference
+
+type reference struct {
+	Name    string `json:"name"`
+	Subject string `json:"subject"`
+	Version int    `json:"version"`
+}
+
+type newSchemaResponse struct {
+	Id uint32 `json:"id"`
+}
+
+type getSchemaResponse struct {
+	Id         uint32     `json:"id"`
+	SchemaType string     `json:"schemaType"`
+	Schema     string     `json:"schema"`
+	References references `json:"references"`
+}
+
+type getSchemaSubjectsResponse []subjectVersion
+
+type subjectVersion struct {
+	Subject string `json:"subject"`
+	Version int    `json:"version"`
+}
+
+

resolver_proto.go

@@ -0,0 +1,54 @@
+package schema_registry
+
+import (
+	"context"
+	"google.golang.org/protobuf/reflect/protoreflect"
+	"google.golang.org/protobuf/reflect/protoregistry"
+	"strings"
+)
+
+type versionedResolver struct {
+	ctx      context.Context
+	registry *Client
+	refs     references
+}
+
+func (r *versionedResolver) FindFileByPath(subject string) (protoreflect.FileDescriptor, error) {
+	for _, ref := range r.refs {
+		if ref.Subject == subject {
+			schema, err := r.registry.GetSubjectVersion(r.ctx, subject, ref.Version)
+			if err != nil {
+				return nil, err
+			}
+			return schema.descriptor, nil
+		}
+	}
+	return nil, protoregistry.NotFound
+}
+
+func (r *versionedResolver) FindDescriptorByName(name protoreflect.FullName) (protoreflect.Descriptor, error) {
+	for _, ref := range r.refs {
+		schema, err := r.registry.GetSubjectVersion(r.ctx, ref.Subject, ref.Version)
+		if err != nil {
+
+		}
+		if schema == nil {
+			return nil, protoregistry.NotFound
+		}
+		matchParent := false
+		for p := name.Parent(); p != ""; p = p.Parent() {
+			if strings.HasSuffix(string(p), string(schema.descriptor.FullName())) {
+				matchParent = true
+			}
+		}
+		if !matchParent {
+			continue
+		}
+		if m := schema.descriptor.Messages().ByName(name.Name()); m != nil {
+			return m, nil
+		}
+
+	}
+	return nil, protoregistry.NotFound
+}
+

schema.go

@@ -0,0 +1,75 @@
+package schema_registry
+
+import (
+	"crypto/sha256"
+	"fmt"
+	"github.com/golang/protobuf/jsonpb"
+	"github.com/jhump/protoreflect/desc"
+	"google.golang.org/protobuf/reflect/protodesc"
+	"google.golang.org/protobuf/reflect/protoreflect"
+)
+
+type Schema struct {
+	descriptor protoreflect.FileDescriptor
+	definition *desc.FileDescriptor
+}
+
+func NewSchema(d protoreflect.FileDescriptor) (*Schema, error) {
+	var convert func(protoreflect.FileDescriptor, bool) (*desc.FileDescriptor, references, error)
+	convert = func(in protoreflect.FileDescriptor, dive bool) (*desc.FileDescriptor, references, error) {
+		fdpb := protodesc.ToFileDescriptorProto(in)
+		imports := in.Imports()
+		refs := make(references, 0)
+		var deps []*desc.FileDescriptor
+		if dive {
+			for i := 0; i < imports.Len(); i++ {
+				imp := imports.Get(i)
+				dp, rs, err := convert(imp, false)
+				if err != nil {
+					return nil, nil, err
+				}
+				for _, r := range rs {
+					refs = append(refs, r)
+				}
+				deps = append(deps, dp)
+			}
+		}
+		fd, err := desc.CreateFileDescriptor(fdpb, deps...)
+		return fd, refs, err
+	}
+	file, _, err := convert(d, true)
+	if err != nil {
+		return nil, fmt.Errorf("NewSchema.CreateFileDescriptor: %v", err)
+	}
+	return &Schema{
+		definition: file,
+		descriptor: d,
+	}, nil
+}
+
+func (s Schema) Fingerprint() (*Fingerprint, error) {
+	m := &jsonpb.Marshaler{
+		OrigName:     true,
+		EnumsAsInts:  true,
+		EmitDefaults: false,
+	}
+	pb := protodesc.ToFileDescriptorProto(s.descriptor)
+	d, err := m.MarshalToString(pb)
+	if err != nil {
+		return nil, err
+	}
+	f := Fingerprint(sha256.Sum256([]byte(d)))
+	return &f, nil
+}
+
+type Fingerprint [32]byte
+
+func (f *Fingerprint) Equal(other *Fingerprint) bool {
+	for i, b := range f {
+		if other[i] != b {
+			return false
+		}
+	}
+	return true
+}
+