Marshal/unmarshal schema.Type (#78)

* marshal/unmarshal schema.Type

* add test for marshal/unmarshal

opencdc/operation.go

@@ -32,32 +32,32 @@ const (
 // Operation defines what triggered the creation of a record.
 type Operation int
 
-func (i Operation) MarshalText() ([]byte, error) {
-	return []byte(i.String()), nil
+func (o Operation) MarshalText() ([]byte, error) {
+	return []byte(o.String()), nil
 }
 
-func (i *Operation) UnmarshalText(b []byte) error {
+func (o *Operation) UnmarshalText(b []byte) error {
 	if len(b) == 0 {
 		return nil // empty string, do nothing
 	}
 
 	switch string(b) {
 	case OperationCreate.String():
-		*i = OperationCreate
+		*o = OperationCreate
 	case OperationUpdate.String():
-		*i = OperationUpdate
+		*o = OperationUpdate
 	case OperationDelete.String():
-		*i = OperationDelete
+		*o = OperationDelete
 	case OperationSnapshot.String():
-		*i = OperationSnapshot
+		*o = OperationSnapshot
 	default:
 		// it's not a known operation, but we also allow Operation(int)
 		valIntRaw := strings.TrimSuffix(strings.TrimPrefix(string(b), "Operation("), ")")
 		valInt, err := strconv.Atoi(valIntRaw)
 		if err != nil {
 			return fmt.Errorf("operation %q: %w", b, ErrUnknownOperation)
 		}
-		*i = Operation(valInt)
+		*o = Operation(valInt)
 	}
 
 	return nil

schema/schema.go

@@ -18,6 +18,8 @@ package schema
 
 import (
 	"fmt"
+	"strconv"
+	"strings"
 	"time"
 
 	"github.com/conduitio/conduit-commons/rabin"
@@ -125,3 +127,30 @@ var KnownSerdeFactories = map[Type]SerdeFactory{
 		SerdeForType: func(v any) (Serde, error) { return avro.SerdeForType(v) },
 	},
 }
+
+// MarshalText returns the textual representation of the schema type.
+func (t Type) MarshalText() ([]byte, error) {
+	return []byte(t.String()), nil
+}
+
+// UnmarshalText parses the textual representation of the schema type.
+func (t *Type) UnmarshalText(b []byte) error {
+	if len(b) == 0 {
+		return nil // empty string, do nothing
+	}
+
+	switch string(b) {
+	case TypeAvro.String():
+		*t = TypeAvro
+	default:
+		// it's not a known type, but we also allow Type(int)
+		valIntRaw := strings.TrimSuffix(strings.TrimPrefix(string(b), "Type("), ")")
+		valInt, err := strconv.Atoi(valIntRaw)
+		if err != nil {
+			return fmt.Errorf("schema type %q: %w", b, ErrUnsupportedType)
+		}
+		*t = Type(valInt)
+	}
+
+	return nil
+}

schema/schema_test.go

@@ -0,0 +1,48 @@
+// Copyright © 2024 Meroxa, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package schema
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/matryer/is"
+)
+
+func TestType(t *testing.T) {
+	testCases := []struct {
+		typ  Type
+		text []byte
+	}{
+		{typ: TypeAvro, text: []byte("avro")},
+		{typ: Type(2), text: []byte("Type(2)")},
+	}
+
+	for _, tc := range testCases {
+		t.Run(fmt.Sprintf("MarshalText_%s", tc.typ.String()), func(t *testing.T) {
+			is := is.New(t)
+			text, err := tc.typ.MarshalText()
+			is.NoErr(err)
+			is.Equal(text, tc.text)
+		})
+		t.Run(fmt.Sprintf("UnmarshalText_%s", tc.typ.String()), func(t *testing.T) {
+			is := is.New(t)
+			var typ Type
+			err := typ.UnmarshalText(tc.text)
+			is.NoErr(err)
+			is.Equal(typ, tc.typ)
+		})
+	}
+}