Merge pull request #742 from twmb/lovro

pkg/sr: touchups

pkg/sr/api.go

@@ -216,7 +216,7 @@ func (cl *Client) SubjectsByID(ctx context.Context, id int) ([]string, error) {
 	return subjects, err
 }
 
-// SchemaVersion is a subject version pair.
+// SubjectVersion is a subject version pair.
 type SubjectVersion struct {
 	Subject string `json:"subject"`
 	Version int    `json:"version"`
@@ -602,7 +602,7 @@ type SetCompatibility struct {
 	OverrideRuleSet  *SchemaRuleSet     `json:"overrideRuleSet,omitempty"`    // Override rule set used for schema registration.
 }
 
-// SetCompatibilitysets the compatibility for each requested subject. The
+// SetCompatibility sets the compatibility for each requested subject. The
 // global compatibility can be set by either using an empty subject or by
 // specifying no subjects. If specifying no subjects, this returns one element.
 func (cl *Client) SetCompatibility(ctx context.Context, compat SetCompatibility, subjects ...string) []CompatibilityResult {

pkg/sr/client.go

@@ -63,7 +63,7 @@ type Client struct {
 }
 
 // NewClient returns a new schema registry client.
-func NewClient(opts ...Opt) (*Client, error) {
+func NewClient(opts ...ClientOpt) (*Client, error) {
 	cl := &Client{
 		urls:   []string{"http://localhost:8081"},
 		httpcl: &http.Client{Timeout: 5 * time.Second},

pkg/sr/config.go → pkg/sr/clientopt.go

@@ -9,30 +9,30 @@ import (
 )
 
 type (
-	// Opt is an option to configure a client.
-	Opt interface{ apply(*Client) }
-	opt struct{ fn func(*Client) }
+	// ClientOpt is an option to configure a client.
+	ClientOpt interface{ apply(*Client) }
+	clientOpt struct{ fn func(*Client) }
 )
 
-func (o opt) apply(cl *Client) { o.fn(cl) }
+func (o clientOpt) apply(cl *Client) { o.fn(cl) }
 
 // HTTPClient sets the http client that the schema registry client uses,
 // overriding the default client that speaks plaintext with a timeout of 5s.
-func HTTPClient(httpcl *http.Client) Opt {
-	return opt{func(cl *Client) { cl.httpcl = httpcl }}
+func HTTPClient(httpcl *http.Client) ClientOpt {
+	return clientOpt{func(cl *Client) { cl.httpcl = httpcl }}
 }
 
 // UserAgent sets the User-Agent to use in requests, overriding the default
 // "franz-go".
-func UserAgent(ua string) Opt {
-	return opt{func(cl *Client) { cl.ua = ua }}
+func UserAgent(ua string) ClientOpt {
+	return clientOpt{func(cl *Client) { cl.ua = ua }}
 }
 
 // URLs sets the URLs that the client speaks to, overriding the default
 // http://localhost:8081. This option automatically prefixes any URL that is
 // missing an http:// or https:// prefix with http://.
-func URLs(urls ...string) Opt {
-	return opt{func(cl *Client) {
+func URLs(urls ...string) ClientOpt {
+	return clientOpt{func(cl *Client) {
 		for i, u := range urls {
 			if strings.HasPrefix(u, "http://") || strings.HasPrefix(u, "https://") {
 				continue
@@ -45,8 +45,8 @@ func URLs(urls ...string) Opt {
 }
 
 // DialTLSConfig sets a tls.Config to use in the default http client.
-func DialTLSConfig(c *tls.Config) Opt {
-	return opt{func(cl *Client) {
+func DialTLSConfig(c *tls.Config) ClientOpt {
+	return clientOpt{func(cl *Client) {
 		cl.httpcl = &http.Client{
 			Timeout: 5 * time.Second,
 			Transport: &http.Transport{
@@ -68,8 +68,8 @@ func DialTLSConfig(c *tls.Config) Opt {
 }
 
 // BasicAuth sets basic authorization to use for every request.
-func BasicAuth(user, pass string) Opt {
-	return opt{func(cl *Client) {
+func BasicAuth(user, pass string) ClientOpt {
+	return clientOpt{func(cl *Client) {
 		cl.basicAuth = &struct {
 			user string
 			pass string
@@ -78,8 +78,8 @@ func BasicAuth(user, pass string) Opt {
 }
 
 // DefaultParams sets default parameters to apply to every request.
-func DefaultParams(ps ...Param) Opt {
-	return opt{func(cl *Client) {
+func DefaultParams(ps ...Param) ClientOpt {
+	return clientOpt{func(cl *Client) {
 		cl.defParams = mergeParams(ps...)
 	}}
 }

pkg/sr/enums.go

@@ -197,7 +197,7 @@ func (k *SchemaRuleKind) UnmarshalText(text []byte) error {
 	return nil
 }
 
-// Mode specifies a schema rule's mode.
+// SchemaRuleMode specifies a schema rule's mode.
 //
 // Migration rules can be specified for an UPGRADE, DOWNGRADE, or both
 // (UPDOWN). Migration rules are used during complex schema evolution.

pkg/sr/serde.go

@@ -21,33 +21,53 @@ var (
 )
 
 type (
-	// SerdeOpt is an option to configure a Serde.
-	SerdeOpt interface{ apply(*tserde) }
-	serdeOpt struct{ fn func(*tserde) }
+	// EncodingOpt is an option to configure the behavior of Serde.Encode and
+	// Serde.Decode.
+	EncodingOpt interface {
+		serdeOrEncodingOpt()
+		apply(*tserde)
+	}
+	encodingOpt struct{ fn func(*tserde) }
+
+	// SerdeOpt is an option to configure Serde.
+	SerdeOpt interface {
+		serdeOrEncodingOpt()
+		apply(*Serde)
+	}
+	serdeOpt struct{ fn func(serde *Serde) }
+
+	// SerdeOrEncodingOpt is either a SerdeOpt or EncodingOpt.
+	SerdeOrEncodingOpt interface {
+		serdeOrEncodingOpt()
+	}
 )
 
-func (o serdeOpt) apply(t *tserde) { o.fn(t) }
+func (o serdeOpt) serdeOrEncodingOpt() { /* satisfy interface */ }
+func (o serdeOpt) apply(s *Serde)      { o.fn(s) }
+
+func (o encodingOpt) serdeOrEncodingOpt() { /* satisfy interface */ }
+func (o encodingOpt) apply(t *tserde)     { o.fn(t) }
 
 // EncodeFn allows Serde to encode a value.
-func EncodeFn(fn func(any) ([]byte, error)) SerdeOpt {
-	return serdeOpt{func(t *tserde) { t.encode = fn }}
+func EncodeFn(fn func(any) ([]byte, error)) EncodingOpt {
+	return encodingOpt{func(t *tserde) { t.encode = fn }}
 }
 
 // AppendEncodeFn allows Serde to encode a value to an existing slice. This
 // can be more efficient than EncodeFn; this function is used if it exists.
-func AppendEncodeFn(fn func([]byte, any) ([]byte, error)) SerdeOpt {
-	return serdeOpt{func(t *tserde) { t.appendEncode = fn }}
+func AppendEncodeFn(fn func([]byte, any) ([]byte, error)) EncodingOpt {
+	return encodingOpt{func(t *tserde) { t.appendEncode = fn }}
 }
 
 // DecodeFn allows Serde to decode into a value.
-func DecodeFn(fn func([]byte, any) error) SerdeOpt {
-	return serdeOpt{func(t *tserde) { t.decode = fn }}
+func DecodeFn(fn func([]byte, any) error) EncodingOpt {
+	return encodingOpt{func(t *tserde) { t.decode = fn }}
 }
 
 // GenerateFn returns a new(Value) that can be decoded into. This function can
 // be used to control the instantiation of a new type for DecodeNew.
-func GenerateFn(fn func() any) SerdeOpt {
-	return serdeOpt{func(t *tserde) { t.gen = fn }}
+func GenerateFn(fn func() any) EncodingOpt {
+	return encodingOpt{func(t *tserde) { t.gen = fn }}
 }
 
 // Index attaches a message index to a value. A single schema ID can be
@@ -62,8 +82,13 @@ func GenerateFn(fn func() any) SerdeOpt {
 // For more information, see where `message-indexes` are described in:
 //
 //	https://docs.confluent.io/platform/current/schema-registry/serdes-develop/index.html#wire-format
-func Index(index ...int) SerdeOpt {
-	return serdeOpt{func(t *tserde) { t.index = index }}
+func Index(index ...int) EncodingOpt {
+	return encodingOpt{func(t *tserde) { t.index = index }}
+}
+
+// Header defines the SerdeHeader used to encode and decode the message header.
+func Header(header SerdeHeader) SerdeOpt {
+	return serdeOpt{func(s *Serde) { s.h = header }}
 }
 
 type tserde struct {
@@ -96,7 +121,7 @@ type Serde struct {
 	types atomic.Value // map[reflect.Type]tserde
 	mu    sync.Mutex
 
-	defaults []SerdeOpt
+	defaults []EncodingOpt
 	h        SerdeHeader
 }
 
@@ -105,6 +130,25 @@ var (
 	noTypes = make(map[reflect.Type]tserde)
 )
 
+// NewSerde returns a new Serde using the supplied default options, which are
+// applied to every registered type. These options are always applied first, so
+// you can override them as necessary when registering.
+//
+// This can be useful if you always want to use the same encoding or decoding
+// functions.
+func NewSerde(opts ...SerdeOrEncodingOpt) *Serde {
+	var s Serde
+	for _, opt := range opts {
+		switch opt := opt.(type) {
+		case SerdeOpt:
+			opt.apply(&s)
+		case EncodingOpt:
+			s.defaults = append(s.defaults, opt)
+		}
+	}
+	return &s
+}
+
 func (s *Serde) loadIDs() map[int]tserde {
 	ids := s.ids.Load()
 	if ids == nil {
@@ -121,16 +165,6 @@ func (s *Serde) loadTypes() map[reflect.Type]tserde {
 	return types.(map[reflect.Type]tserde)
 }
 
-// SetDefaults sets default options to apply to every registered type. These
-// options are always applied first, so you can override them as necessary when
-// registering.
-//
-// This can be useful if you always want to use the same encoding or decoding
-// functions.
-func (s *Serde) SetDefaults(opts ...SerdeOpt) {
-	s.defaults = opts
-}
-
 // DecodeID decodes an ID from b, returning the ID and the remaining bytes,
 // or an error.
 func (s *Serde) DecodeID(b []byte) (id int, out []byte, err error) {
@@ -154,7 +188,7 @@ func (s *Serde) header() SerdeHeader {
 // Register registers a schema ID and the value it corresponds to, as well as
 // the encoding or decoding functions. You need to register functions depending
 // on whether you are only encoding, only decoding, or both.
-func (s *Serde) Register(id int, v any, opts ...SerdeOpt) {
+func (s *Serde) Register(id int, v any, opts ...EncodingOpt) {
 	var t tserde
 	for _, opt := range s.defaults {
 		opt.apply(&t)
@@ -258,20 +292,18 @@ func (s *Serde) Encode(v any) ([]byte, error) {
 	return s.AppendEncode(nil, v)
 }
 
-// AppendEncode appends an encoded value to b according to the schema registry
-// wire format and returns it. If EncodeFn was not used, this returns
-// ErrNotRegistered.
+// AppendEncode encodes a value and prepends the header according to the
+// configured SerdeHeader, appends it to b and returns b. If EncodeFn was not
+// registered, this returns ErrNotRegistered.
 func (s *Serde) AppendEncode(b []byte, v any) ([]byte, error) {
 	t, ok := s.loadTypes()[reflect.TypeOf(v)]
 	if !ok || (t.encode == nil && t.appendEncode == nil) {
 		return b, ErrNotRegistered
 	}
-
 	b, err := s.header().AppendEncode(b, int(t.id), t.index)
 	if err != nil {
 		return nil, err
 	}
-
 	if t.appendEncode != nil {
 		return t.appendEncode(b, v)
 	}
@@ -328,8 +360,10 @@ func (s *Serde) DecodeNew(b []byte) (any, error) {
 	var v any
 	if t.gen != nil {
 		v = t.gen()
-	} else {
+	} else if t.typeof != nil {
 		v = reflect.New(t.typeof).Interface()
+	} else {
+		return nil, ErrNotRegistered
 	}
 	return v, t.decode(b, v)
 }
@@ -339,7 +373,6 @@ func (s *Serde) decodeFind(b []byte) ([]byte, tserde, error) {
 	if err != nil {
 		return nil, tserde{}, err
 	}
-
 	t := s.loadIDs()[id]
 	if len(t.subindex) > 0 {
 		var index []int
@@ -360,6 +393,28 @@ func (s *Serde) decodeFind(b []byte) ([]byte, tserde, error) {
 	return b, t, nil
 }
 
+// Encode encodes a value and prepends the header. If the encoding function
+// fails, this returns an error.
+func Encode(v any, id int, index []int, h SerdeHeader, enc func(any) ([]byte, error)) ([]byte, error) {
+	return AppendEncode(nil, v, id, index, h, func(b []byte, val any) ([]byte, error) {
+		encoded, err := enc(val)
+		if err != nil {
+			return nil, err
+		}
+		return append(b, encoded...), nil
+	})
+}
+
+// AppendEncode encodes a value and prepends the header, appends it to b and
+// returns b. If the encoding function fails, this returns an error.
+func AppendEncode(b []byte, v any, id int, index []int, h SerdeHeader, enc func([]byte, any) ([]byte, error)) ([]byte, error) {
+	b, err := h.AppendEncode(b, id, index)
+	if err != nil {
+		return nil, err
+	}
+	return enc(b, v)
+}
+
 // SerdeHeader encodes and decodes a message header.
 type SerdeHeader interface {
 	// AppendEncode encodes a schema ID and optional index to b, returning the

pkg/sr/serde_test.go

@@ -36,8 +36,7 @@ func TestSerde(t *testing.T) {
 		}
 	)
 
-	var serde Serde
-	serde.SetDefaults(
+	serde := NewSerde(
 		EncodeFn(json.Marshal),
 		DecodeFn(json.Unmarshal),
 	)
@@ -113,6 +112,19 @@ func TestSerde(t *testing.T) {
 			t.Errorf("#%d got MustAppendEncode(%v) != Encode(foo%v)", i, b2, b)
 		}
 
+		bIndented, err := Encode(test.enc, 100, []int{0}, serde.header(), func(v any) ([]byte, error) {
+			return json.MarshalIndent(v, "", "  ")
+		})
+		if err != nil {
+			t.Errorf("#%d Encode[ID=100]: got err? %v, exp err? %v", i, gotErr, test.expErr)
+			continue
+		}
+		if i := bytes.IndexByte(bIndented, '{'); !bytes.Equal(bIndented[:i], []byte{0, 0, 0, 0, 100, 0}) {
+			t.Errorf("#%d got Encode[ID=100](%v) != exp(%v)", i, bIndented[:i], []byte{0, 0, 0, 0, 100, 0})
+		} else if expIndented := extractIndentedJSON(b); !bytes.Equal(bIndented[i:], expIndented) {
+			t.Errorf("#%d got Encode[ID=100](%v) != exp(%v)", i, bIndented[i:], expIndented)
+		}
+
 		v, err := serde.DecodeNew(b)
 		if err != nil {
 			t.Errorf("#%d DecodeNew: got unexpected err %v", i, err)
@@ -126,6 +138,7 @@ func TestSerde(t *testing.T) {
 		}
 		if !reflect.DeepEqual(v, exp) {
 			t.Errorf("#%d round trip: got %v != exp %v", i, v, exp)
+			continue
 		}
 	}
 
@@ -141,6 +154,20 @@ func TestSerde(t *testing.T) {
 	if _, err := serde.DecodeNew([]byte{0, 0, 0, 0, 99}); err != ErrNotRegistered {
 		t.Errorf("got %v != exp ErrNotRegistered", err)
 	}
+	if _, err := serde.DecodeNew([]byte{0, 0, 0, 0, 100, 0}); err != ErrNotRegistered {
+		// schema is registered but type is unknown
+		t.Errorf("got %v != exp ErrNotRegistered", err)
+	}
+}
+
+func extractIndentedJSON(in []byte) []byte {
+	i := bytes.IndexByte(in, '{') // skip header
+	var out bytes.Buffer
+	err := json.Indent(&out, in[i:], "", "  ")
+	if err != nil {
+		panic(err)
+	}
+	return out.Bytes()
 }
 
 func TestConfluentHeader(t *testing.T) {