Function structDecodeFuncOf

protocol/decode.go:389–445

(typ reflect.Type, version int16, flexible bool)

}

func structDecodeFuncOf(typ reflect.Type, version int16, flexible bool) decodeFunc {
	type field struct {
		decode decodeFunc
		index  index
		tagID  int
	}

	var fields []field
	taggedFields := map[int]*field{}

	forEachStructField(typ, func(typ reflect.Type, index index, tag string) {
		forEachStructTag(tag, func(tag structTag) bool {
			if tag.MinVersion <= version && version <= tag.MaxVersion {
				f := field{
					decode: decodeFuncOf(typ, version, flexible, tag),
					index:  index,
					tagID:  tag.TagID,
				}

				if tag.TagID < -1 {
					// Normal required field
					fields = append(fields, f)
				} else {
					// Optional tagged field (flexible messages only)
					taggedFields[tag.TagID] = &f
				}
				return false
			}
			return true
		})
	})

	return func(d *decoder, v value) {
		for i := range fields {
			f := &fields[i]
			f.decode(d, v.fieldByIndex(f.index))
		}

		if flexible {
			// See https://cwiki.apache.org/confluence/display/KAFKA/KIP-482%3A+The+Kafka+Protocol+should+Support+Optional+Tagged+Fields
			// for details of tag buffers in "flexible" messages.
			n := int(d.readUnsignedVarInt())

			for i := 0; i < n; i++ {
				tagID := int(d.readUnsignedVarInt())
				size := int(d.readUnsignedVarInt())

				f, ok := taggedFields[tagID]
				if ok {
					f.decode(d, v.fieldByIndex(f.index))
				} else {
					d.read(size)
				}
			}
		}
	}
}