add support for recursive schemas & structs

.gitignore

@@ -0,0 +1 @@
+.idea

README.md

@@ -142,10 +142,6 @@ For security reasons, the configuration `Config.MaxByteSliceSize` restricts the
 by the `Reader`. The default maximum size is `1MiB` and is configurable. This is required to stop untrusted input from consuming all memory and
 crashing the application. Should this not be need, setting a negative number will disable the behaviour.
 
-### Recursive Structs
-
-At this moment recursive structs are not supported. It is planned for the future.
-
 ## Benchmark
 
 Benchmark source code can be found at: [https://github.com/nrwiersma/avro-benchmarks](https://github.com/nrwiersma/avro-benchmarks)

codec.go

@@ -71,48 +71,88 @@ func (c *frozenConfig) DecoderOf(schema Schema, typ reflect2.Type) ValDecoder {
 	}
 
 	ptrType := typ.(*reflect2.UnsafePtrType)
-	decoder = decoderOfType(c, schema, ptrType.Elem())
+	decoder = decoderOfType(newDecoderContext(c), schema, ptrType.Elem())
 	c.addDecoderToCache(schema.CacheFingerprint(), rtype, decoder)
 	return decoder
 }
 
-func decoderOfType(cfg *frozenConfig, schema Schema, typ reflect2.Type) ValDecoder {
-	if dec := createDecoderOfMarshaler(cfg, schema, typ); dec != nil {
+type deferDecoder struct {
+	decoder ValDecoder
+}
+
+func (d *deferDecoder) Decode(ptr unsafe.Pointer, r *Reader) {
+	d.decoder.Decode(ptr, r)
+}
+
+type deferEncoder struct {
+	encoder ValEncoder
+}
+
+func (d *deferEncoder) Encode(ptr unsafe.Pointer, w *Writer) {
+	d.encoder.Encode(ptr, w)
+}
+
+type decoderContext struct {
+	cfg      *frozenConfig
+	decoders map[cacheKey]ValDecoder
+}
+
+func newDecoderContext(cfg *frozenConfig) *decoderContext {
+	return &decoderContext{
+		cfg:      cfg,
+		decoders: make(map[cacheKey]ValDecoder),
+	}
+}
+
+type encoderContext struct {
+	cfg      *frozenConfig
+	encoders map[cacheKey]ValEncoder
+}
+
+func newEncoderContext(cfg *frozenConfig) *encoderContext {
+	return &encoderContext{
+		cfg:      cfg,
+		encoders: make(map[cacheKey]ValEncoder),
+	}
+}
+
+func decoderOfType(d *decoderContext, schema Schema, typ reflect2.Type) ValDecoder {
+	if dec := createDecoderOfMarshaler(schema, typ); dec != nil {
 		return dec
 	}
 
-	// Handle eface case when it isnt a union
+	// Handle eface (empty interface) case when it isn't a union
 	if typ.Kind() == reflect.Interface && schema.Type() != Union {
 		if _, ok := typ.(*reflect2.UnsafeIFaceType); !ok {
-			return newEfaceDecoder(cfg, schema)
+			return newEfaceDecoder(d, schema)
 		}
 	}
 
 	switch schema.Type() {
 	case String, Bytes, Int, Long, Float, Double, Boolean:
 		return createDecoderOfNative(schema.(*PrimitiveSchema), typ)
-
 	case Record:
-		return createDecoderOfRecord(cfg, schema, typ)
-
+		key := cacheKey{fingerprint: schema.Fingerprint(), rtype: typ.RType()}
+		defDec := &deferDecoder{}
+		d.decoders[key] = defDec
+		defDec.decoder = createDecoderOfRecord(d, schema.(*RecordSchema), typ)
+		return defDec.decoder
 	case Ref:
-		return decoderOfType(cfg, schema.(*RefSchema).Schema(), typ)
-
+		key := cacheKey{fingerprint: schema.(*RefSchema).Schema().Fingerprint(), rtype: typ.RType()}
+		if dec, f := d.decoders[key]; f {
+			return dec
+		}
+		return decoderOfType(d, schema.(*RefSchema).Schema(), typ)
 	case Enum:
-		return createDecoderOfEnum(schema, typ)
-
+		return createDecoderOfEnum(schema.(*EnumSchema), typ)
 	case Array:
-		return createDecoderOfArray(cfg, schema, typ)
-
+		return createDecoderOfArray(d, schema.(*ArraySchema), typ)
 	case Map:
-		return createDecoderOfMap(cfg, schema, typ)
-
+		return createDecoderOfMap(d, schema.(*MapSchema), typ)
 	case Union:
-		return createDecoderOfUnion(cfg, schema, typ)
-
+		return createDecoderOfUnion(d, schema.(*UnionSchema), typ)
 	case Fixed:
-		return createDecoderOfFixed(schema, typ)
-
+		return createDecoderOfFixed(schema.(*FixedSchema), typ)
 	default:
 		// It is impossible to get here with a valid schema
 		return &errorDecoder{err: fmt.Errorf("avro: schema type %s is unsupported", schema.Type())}
@@ -130,7 +170,7 @@ func (c *frozenConfig) EncoderOf(schema Schema, typ reflect2.Type) ValEncoder {
 		return encoder
 	}
 
-	encoder = encoderOfType(c, schema, typ)
+	encoder = encoderOfType(newEncoderContext(c), schema, typ)
 	if typ.LikePtr() {
 		encoder = &onePtrEncoder{encoder}
 	}
@@ -146,8 +186,8 @@ func (e *onePtrEncoder) Encode(ptr unsafe.Pointer, w *Writer) {
 	e.enc.Encode(noescape(unsafe.Pointer(&ptr)), w)
 }
 
-func encoderOfType(cfg *frozenConfig, schema Schema, typ reflect2.Type) ValEncoder {
-	if enc := createEncoderOfMarshaler(cfg, schema, typ); enc != nil {
+func encoderOfType(e *encoderContext, schema Schema, typ reflect2.Type) ValEncoder {
+	if enc := createEncoderOfMarshaler(schema, typ); enc != nil {
 		return enc
 	}
 
@@ -158,28 +198,28 @@ func encoderOfType(cfg *frozenConfig, schema Schema, typ reflect2.Type) ValEncod
 	switch schema.Type() {
 	case String, Bytes, Int, Long, Float, Double, Boolean, Null:
 		return createEncoderOfNative(schema, typ)
-
 	case Record:
-		return createEncoderOfRecord(cfg, schema, typ)
-
+		key := cacheKey{fingerprint: schema.Fingerprint(), rtype: typ.RType()}
+		defEnc := &deferEncoder{}
+		e.encoders[key] = defEnc
+		defEnc.encoder = createEncoderOfRecord(e, schema.(*RecordSchema), typ)
+		return defEnc.encoder
 	case Ref:
-		return encoderOfType(cfg, schema.(*RefSchema).Schema(), typ)
-
+		key := cacheKey{fingerprint: schema.(*RefSchema).Schema().Fingerprint(), rtype: typ.RType()}
+		if enc, f := e.encoders[key]; f {
+			return enc
+		}
+		return encoderOfType(e, schema.(*RefSchema).Schema(), typ)
 	case Enum:
-		return createEncoderOfEnum(schema, typ)
-
+		return createEncoderOfEnum(schema.(*EnumSchema), typ)
 	case Array:
-		return createEncoderOfArray(cfg, schema, typ)
-
+		return createEncoderOfArray(e, schema.(*ArraySchema), typ)
 	case Map:
-		return createEncoderOfMap(cfg, schema, typ)
-
+		return createEncoderOfMap(e, schema.(*MapSchema), typ)
 	case Union:
-		return createEncoderOfUnion(cfg, schema, typ)
-
+		return createEncoderOfUnion(e, schema.(*UnionSchema), typ)
 	case Fixed:
-		return createEncoderOfFixed(schema, typ)
-
+		return createEncoderOfFixed(schema.(*FixedSchema), typ)
 	default:
 		// It is impossible to get here with a valid schema
 		return &errorEncoder{err: fmt.Errorf("avro: schema type %s is unsupported", schema.Type())}

codec_array.go

@@ -10,26 +10,25 @@ import (
 	"github.com/modern-go/reflect2"
 )
 
-func createDecoderOfArray(cfg *frozenConfig, schema Schema, typ reflect2.Type) ValDecoder {
+func createDecoderOfArray(d *decoderContext, schema *ArraySchema, typ reflect2.Type) ValDecoder {
 	if typ.Kind() == reflect.Slice {
-		return decoderOfArray(cfg, schema, typ)
+		return decoderOfArray(d, schema, typ)
 	}
 
 	return &errorDecoder{err: fmt.Errorf("avro: %s is unsupported for Avro %s", typ.String(), schema.Type())}
 }
 
-func createEncoderOfArray(cfg *frozenConfig, schema Schema, typ reflect2.Type) ValEncoder {
+func createEncoderOfArray(e *encoderContext, schema *ArraySchema, typ reflect2.Type) ValEncoder {
 	if typ.Kind() == reflect.Slice {
-		return encoderOfArray(cfg, schema, typ)
+		return encoderOfArray(e, schema, typ)
 	}
 
 	return &errorEncoder{err: fmt.Errorf("avro: %s is unsupported for Avro %s", typ.String(), schema.Type())}
 }
 
-func decoderOfArray(cfg *frozenConfig, schema Schema, typ reflect2.Type) ValDecoder {
-	arr := schema.(*ArraySchema)
+func decoderOfArray(d *decoderContext, arr *ArraySchema, typ reflect2.Type) ValDecoder {
 	sliceType := typ.(*reflect2.UnsafeSliceType)
-	decoder := decoderOfType(cfg, arr.Items(), sliceType.Elem())
+	decoder := decoderOfType(d, arr.Items(), sliceType.Elem())
 
 	return &arrayDecoder{typ: sliceType, decoder: decoder}
 }
@@ -74,13 +73,12 @@ func (d *arrayDecoder) Decode(ptr unsafe.Pointer, r *Reader) {
 	}
 }
 
-func encoderOfArray(cfg *frozenConfig, schema Schema, typ reflect2.Type) ValEncoder {
-	arr := schema.(*ArraySchema)
+func encoderOfArray(e *encoderContext, arr *ArraySchema, typ reflect2.Type) ValEncoder {
 	sliceType := typ.(*reflect2.UnsafeSliceType)
-	encoder := encoderOfType(cfg, arr.Items(), sliceType.Elem())
+	encoder := encoderOfType(e, arr.Items(), sliceType.Elem())
 
 	return &arrayEncoder{
-		blockLength: cfg.getBlockLength(),
+		blockLength: e.cfg.getBlockLength(),
 		typ:         sliceType,
 		encoder:     encoder,
 	}

codec_default.go

@@ -7,13 +7,14 @@ import (
 	"github.com/modern-go/reflect2"
 )
 
-func createDefaultDecoder(cfg *frozenConfig, field *Field, typ reflect2.Type) ValDecoder {
+func createDefaultDecoder(d *decoderContext, field *Field, typ reflect2.Type) ValDecoder {
+	var cfg = d.cfg
 	fn := func(def any) ([]byte, error) {
 		defaultType := reflect2.TypeOf(def)
 		if defaultType == nil {
 			defaultType = reflect2.TypeOf((*null)(nil))
 		}
-		defaultEncoder := encoderOfType(cfg, field.Type(), defaultType)
+		defaultEncoder := encoderOfType(newEncoderContext(cfg), field.Type(), defaultType)
 		if defaultType.LikePtr() {
 			defaultEncoder = &onePtrEncoder{defaultEncoder}
 		}
@@ -37,7 +38,7 @@ func createDefaultDecoder(cfg *frozenConfig, field *Field, typ reflect2.Type) Va
 	}
 	return &defaultDecoder{
 		data:    b,
-		decoder: decoderOfType(cfg, field.Type(), typ),
+		decoder: decoderOfType(d, field.Type(), typ),
 	}
 }
 

codec_dynamic.go

@@ -13,9 +13,9 @@ type efaceDecoder struct {
 	dec    ValDecoder
 }
 
-func newEfaceDecoder(cfg *frozenConfig, schema Schema) *efaceDecoder {
+func newEfaceDecoder(d *decoderContext, schema Schema) *efaceDecoder {
 	typ, _ := genericReceiver(schema)
-	dec := decoderOfType(cfg, schema, typ)
+	dec := decoderOfType(d, schema, typ)
 
 	return &efaceDecoder{
 		schema: schema,

codec_enum.go

@@ -10,27 +10,27 @@ import (
 	"github.com/modern-go/reflect2"
 )
 
-func createDecoderOfEnum(schema Schema, typ reflect2.Type) ValDecoder {
+func createDecoderOfEnum(schema *EnumSchema, typ reflect2.Type) ValDecoder {
 	switch {
 	case typ.Kind() == reflect.String:
-		return &enumCodec{enum: schema.(*EnumSchema)}
+		return &enumCodec{enum: schema}
 	case typ.Implements(textUnmarshalerType):
-		return &enumTextMarshalerCodec{typ: typ, enum: schema.(*EnumSchema)}
+		return &enumTextMarshalerCodec{typ: typ, enum: schema}
 	case reflect2.PtrTo(typ).Implements(textUnmarshalerType):
-		return &enumTextMarshalerCodec{typ: typ, enum: schema.(*EnumSchema), ptr: true}
+		return &enumTextMarshalerCodec{typ: typ, enum: schema, ptr: true}
 	}
 
 	return &errorDecoder{err: fmt.Errorf("avro: %s is unsupported for Avro %s", typ.String(), schema.Type())}
 }
 
-func createEncoderOfEnum(schema Schema, typ reflect2.Type) ValEncoder {
+func createEncoderOfEnum(schema *EnumSchema, typ reflect2.Type) ValEncoder {
 	switch {
 	case typ.Kind() == reflect.String:
-		return &enumCodec{enum: schema.(*EnumSchema)}
+		return &enumCodec{enum: schema}
 	case typ.Implements(textMarshalerType):
-		return &enumTextMarshalerCodec{typ: typ, enum: schema.(*EnumSchema)}
+		return &enumTextMarshalerCodec{typ: typ, enum: schema}
 	case reflect2.PtrTo(typ).Implements(textMarshalerType):
-		return &enumTextMarshalerCodec{typ: typ, enum: schema.(*EnumSchema), ptr: true}
+		return &enumTextMarshalerCodec{typ: typ, enum: schema, ptr: true}
 	}
 
 	return &errorEncoder{err: fmt.Errorf("avro: %s is unsupported for Avro %s", typ.String(), schema.Type())}

codec_fixed.go

@@ -10,8 +10,7 @@ import (
 	"github.com/modern-go/reflect2"
 )
 
-func createDecoderOfFixed(schema Schema, typ reflect2.Type) ValDecoder {
-	fixed := schema.(*FixedSchema)
+func createDecoderOfFixed(fixed *FixedSchema, typ reflect2.Type) ValDecoder {
 	switch typ.Kind() {
 	case reflect.Array:
 		arrayType := typ.(reflect2.ArrayType)
@@ -21,7 +20,6 @@ func createDecoderOfFixed(schema Schema, typ reflect2.Type) ValDecoder {
 		return &fixedCodec{arrayType: typ.(*reflect2.UnsafeArrayType)}
 
 	case reflect.Uint64:
-		fixed := schema.(*FixedSchema)
 		if fixed.Size() != 8 {
 			break
 		}
@@ -44,23 +42,20 @@ func createDecoderOfFixed(schema Schema, typ reflect2.Type) ValDecoder {
 	}
 
 	return &errorDecoder{
-		err: fmt.Errorf("avro: %s is unsupported for Avro %s, size=%d", typ.String(), schema.Type(), fixed.Size()),
+		err: fmt.Errorf("avro: %s is unsupported for Avro %s, size=%d", typ.String(), fixed.Type(), fixed.Size()),
 	}
 }
 
-func createEncoderOfFixed(schema Schema, typ reflect2.Type) ValEncoder {
-	fixed := schema.(*FixedSchema)
+func createEncoderOfFixed(fixed *FixedSchema, typ reflect2.Type) ValEncoder {
 	switch typ.Kind() {
 	case reflect.Array:
 		arrayType := typ.(reflect2.ArrayType)
-		fixed := schema.(*FixedSchema)
 		if arrayType.Elem().Kind() != reflect.Uint8 || arrayType.Len() != fixed.Size() {
 			break
 		}
 		return &fixedCodec{arrayType: typ.(*reflect2.UnsafeArrayType)}
 
 	case reflect.Uint64:
-		fixed := schema.(*FixedSchema)
 		if fixed.Size() != 8 {
 			break
 		}
@@ -92,7 +87,7 @@ func createEncoderOfFixed(schema Schema, typ reflect2.Type) ValEncoder {
 	}
 
 	return &errorEncoder{
-		err: fmt.Errorf("avro: %s is unsupported for Avro %s, size=%d", typ.String(), schema.Type(), fixed.Size()),
+		err: fmt.Errorf("avro: %s is unsupported for Avro %s, size=%d", typ.String(), fixed.Type(), fixed.Size()),
 	}
 }
 

codec_generic_internal_test.go

@@ -228,7 +228,7 @@ func TestGenericDecode(t *testing.T) {
 
 			typ, err := genericReceiver(schema)
 			require.NoError(t, err)
-			dec := decoderOfType(DefaultConfig.(*frozenConfig), schema, typ)
+			dec := decoderOfType(newDecoderContext(DefaultConfig.(*frozenConfig)), schema, typ)
 
 			got := genericDecode(typ, dec, r)