Add recursive structs supported

.gitignore

@@ -0,0 +1,2 @@
+# Goland
+.idea

codec_record.go

@@ -54,6 +54,13 @@ func createEncoderOfRecord(cfg *frozenConfig, schema Schema, typ reflect2.Type)
 }
 
 func decoderOfStruct(cfg *frozenConfig, schema Schema, typ reflect2.Type) ValDecoder {
+	processing := cfg.getProcessingDecoderFromCache(schema.Fingerprint(), typ.RType())
+	if processing != nil {
+		return processing
+	}
+	dec := &structDecoder{typ: typ, fields: nil}
+	cfg.addProcessingDecoderToCache(schema.Fingerprint(), typ.RType(), dec)
+
 	rec := schema.(*RecordSchema)
 	structDesc := describeStruct(cfg.getTagKey(), typ)
 
@@ -77,14 +84,14 @@ func decoderOfStruct(cfg *frozenConfig, schema Schema, typ reflect2.Type) ValDec
 			continue
 		}
 
-		dec := decoderOfType(cfg, field.Type(), sf.Field[len(sf.Field)-1].Type())
 		fields = append(fields, &structFieldDecoder{
 			field:   sf.Field,
-			decoder: dec,
+			decoder: decoderOfType(cfg, field.Type(), sf.Field[len(sf.Field)-1].Type()),
 		})
 	}
 
-	return &structDecoder{typ: typ, fields: fields}
+	dec.fields = fields
+	return dec
 }
 
 type structFieldDecoder struct {
@@ -134,6 +141,13 @@ func (d *structDecoder) Decode(ptr unsafe.Pointer, r *Reader) {
 }
 
 func encoderOfStruct(cfg *frozenConfig, schema Schema, typ reflect2.Type) ValEncoder {
+	processing := cfg.getProcessingEncoderFromCache(schema.Fingerprint(), typ.RType())
+	if processing != nil {
+		return processing
+	}
+	enc := &structEncoder{typ: typ, fields: nil}
+	cfg.addProcessingEncoderToCache(schema.Fingerprint(), typ.RType(), enc)
+
 	rec := schema.(*RecordSchema)
 	structDesc := describeStruct(cfg.getTagKey(), typ)
 
@@ -181,7 +195,9 @@ func encoderOfStruct(cfg *frozenConfig, schema Schema, typ reflect2.Type) ValEnc
 			encoder:    defaultEncoder,
 		})
 	}
-	return &structEncoder{typ: typ, fields: fields}
+
+	enc.fields = fields
+	return enc
 }
 
 type structFieldEncoder struct {

config.go

@@ -114,6 +114,9 @@ type frozenConfig struct {
 	decoderCache sync.Map // map[cacheKey]ValDecoder
 	encoderCache sync.Map // map[cacheKey]ValEncoder
 
+	processingDecoderCache sync.Map // map[cacheKey]ValDecoder
+	processingEncoderCache sync.Map // map[cacheKey]ValEncoder
+
 	readerPool *sync.Pool
 	writerPool *sync.Pool
 
@@ -243,6 +246,42 @@ func (c *frozenConfig) getEncoderFromCache(fingerprint [32]byte, rtype uintptr)
 	return nil
 }
 
+func (c *frozenConfig) addProcessingDecoderToCache(fingerprint [32]byte, rtype uintptr, dec ValDecoder) {
+	key := cacheKey{fingerprint: fingerprint, rtype: rtype}
+	c.processingDecoderCache.Store(key, dec)
+}
+
+func (c *frozenConfig) getProcessingDecoderFromCache(fingerprint [32]byte, rtype uintptr) ValDecoder {
+	key := cacheKey{fingerprint: fingerprint, rtype: rtype}
+	if !c.config.DisableCaching {
+		if dec, ok := c.decoderCache.Load(key); ok {
+			return dec.(ValDecoder)
+		}
+	}
+	if dec, ok := c.processingDecoderCache.Load(key); ok {
+		return dec.(ValDecoder)
+	}
+	return nil
+}
+
+func (c *frozenConfig) addProcessingEncoderToCache(fingerprint [32]byte, rtype uintptr, enc ValEncoder) {
+	key := cacheKey{fingerprint: fingerprint, rtype: rtype}
+	c.processingEncoderCache.Store(key, enc)
+}
+
+func (c *frozenConfig) getProcessingEncoderFromCache(fingerprint [32]byte, rtype uintptr) ValEncoder {
+	key := cacheKey{fingerprint: fingerprint, rtype: rtype}
+	if !c.config.DisableCaching {
+		if enc, ok := c.encoderCache.Load(key); ok {
+			return enc.(ValEncoder)
+		}
+	}
+	if enc, ok := c.processingEncoderCache.Load(key); ok {
+		return enc.(ValEncoder)
+	}
+	return nil
+}
+
 func (c *frozenConfig) getTagKey() string {
 	tagKey := c.config.TagKey
 	if tagKey == "" {

decoder_record_test.go

@@ -393,3 +393,28 @@ func TestDecoder_RefStruct(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, want, got)
 }
+
+func TestDecoder_RecursiveStructs(t *testing.T) {
+	defer ConfigTeardown()
+
+	data := []byte{0x36, 0x6, 0x66, 0x6f, 0x6f, 0x2, 0x38, 0x6, 0x62, 0x61, 0x72, 0x0}
+	schema := `{
+	"type": "record",
+	"namespace": "tns",
+	"name": "test",
+	"fields" : [
+		{"name": "a", "type": "long"},
+	    {"name": "b", "type": "string"},
+	    {"name": "r", "type": ["null", "tns.test"]}
+	]
+}`
+
+	dec, err := avro.NewDecoder(schema, bytes.NewReader(data))
+	require.NoError(t, err)
+
+	var got TestRecursiveRecord
+	err = dec.Decode(&got)
+
+	require.NoError(t, err)
+	assert.Equal(t, TestRecursiveRecord{A: 27, B: "foo", R: &TestRecursiveRecord{A: 28, B: "bar"}}, got)
+}

encoder_record_test.go

@@ -622,3 +622,26 @@ func TestEncoder_RefStruct(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, []byte{0x36, 0x06, 0x66, 0x6f, 0x6f, 0x36, 0x06, 0x66, 0x6f, 0x6f}, buf.Bytes())
 }
+
+func TestEncoder_RecursiveStructs(t *testing.T) {
+	defer ConfigTeardown()
+	schema := `{
+	"type": "record",
+	"namespace": "tns",
+	"name": "test",
+	"fields" : [
+		{"name": "a", "type": "long"},
+	    {"name": "b", "type": "string"},
+	    {"name": "r", "type": ["null", "tns.test"]}
+	]
+}`
+	obj := TestRecursiveRecord{A: 27, B: "foo", R: &TestRecursiveRecord{A: 28, B: "bar"}}
+	buf := &bytes.Buffer{}
+	enc, err := avro.NewEncoder(schema, buf)
+	require.NoError(t, err)
+
+	err = enc.Encode(obj)
+
+	require.NoError(t, err)
+	assert.Equal(t, []byte{0x36, 0x6, 0x66, 0x6f, 0x6f, 0x2, 0x38, 0x6, 0x62, 0x61, 0x72, 0x0}, buf.Bytes())
+}

types_test.go

@@ -55,3 +55,9 @@ type TestUnexportedRecord struct {
 	A int64  `avro:"a"`
 	b string `avro:"b"`
 }
+
+type TestRecursiveRecord struct {
+	A int64                `avro:"a"`
+	B string               `avro:"b"`
+	R *TestRecursiveRecord `avro:"r"`
+}