struct to schema correctly

3 files changed, 152 insertions, 20 deletions

diff --git a/weed/mq/schema/schema_builder.go b/weed/mq/schema/schema_builder.go
index cadf4f927..224c79309 100644
--- a/weed/mq/schema/schema_builder.go
+++ b/weed/mq/schema/schema_builder.go
@@ -31,7 +31,7 @@ func (rtb *RecordTypeBuilder) Build() *schema_pb.RecordType {
 	return rtb.recordType
 }
 
-func (rtb *RecordTypeBuilder) setField(name string, scalarType *schema_pb.Type) *RecordTypeBuilder {
+func (rtb *RecordTypeBuilder) SetField(name string, scalarType *schema_pb.Type) *RecordTypeBuilder {
 	rtb.recordType.Fields = append(rtb.recordType.Fields, &schema_pb.Field{
 		Name: name,
 		Type: scalarType,
@@ -40,25 +40,25 @@ func (rtb *RecordTypeBuilder) setField(name string, scalarType *schema_pb.Type)
 }
 
 func (rtb *RecordTypeBuilder) SetBoolField(name string) *RecordTypeBuilder {
-	return rtb.setField(name, TypeBoolean)
+	return rtb.SetField(name, TypeBoolean)
 }
 func (rtb *RecordTypeBuilder) SetIntegerField(name string) *RecordTypeBuilder {
-	return rtb.setField(name, TypeInteger)
+	return rtb.SetField(name, TypeInteger)
 }
 func (rtb *RecordTypeBuilder) SetLongField(name string) *RecordTypeBuilder {
-	return rtb.setField(name, TypeLong)
+	return rtb.SetField(name, TypeLong)
 }
 func (rtb *RecordTypeBuilder) SetFloatField(name string) *RecordTypeBuilder {
-	return rtb.setField(name, TypeFloat)
+	return rtb.SetField(name, TypeFloat)
 }
 func (rtb *RecordTypeBuilder) SetDoubleField(name string) *RecordTypeBuilder {
-	return rtb.setField(name, TypeDouble)
+	return rtb.SetField(name, TypeDouble)
 }
 func (rtb *RecordTypeBuilder) SetBytesField(name string) *RecordTypeBuilder {
-	return rtb.setField(name, TypeBytes)
+	return rtb.SetField(name, TypeBytes)
 }
 func (rtb *RecordTypeBuilder) SetStringField(name string) *RecordTypeBuilder {
-	return rtb.setField(name, TypeString)
+	return rtb.SetField(name, TypeString)
 }
 
 func (rtb *RecordTypeBuilder) SetRecordField(name string, recordTypeBuilder *RecordTypeBuilder) *RecordTypeBuilder {
@@ -76,3 +76,7 @@ func (rtb *RecordTypeBuilder) SetListField(name string, elementType *schema_pb.T
 	})
 	return rtb
 }
+
+func List(elementType *schema_pb.Type) *schema_pb.Type {
+	return &schema_pb.Type{Kind: &schema_pb.Type_ListType{ListType: &schema_pb.ListType{ElementType: elementType}}}
+}
diff --git a/weed/mq/schema/struct_to_schema.go b/weed/mq/schema/struct_to_schema.go
index bdd3e2465..80f8b2e57 100644
--- a/weed/mq/schema/struct_to_schema.go
+++ b/weed/mq/schema/struct_to_schema.go
@@ -5,20 +5,13 @@ import (
 	"reflect"
 )
 
-func StructToSchema(instance any) *RecordTypeBuilder {
-	rtb := NewRecordTypeBuilder()
+func StructToSchema(instance any) *schema_pb.RecordType {
 	myType := reflect.TypeOf(instance)
-	for i := 0; i < myType.NumField(); i++ {
-		field := myType.Field(i)
-		fieldType := field.Type
-		fieldName := field.Name
-		schemaField := reflectTypeToSchemaType(fieldType)
-		if schemaField == nil {
-			continue
-		}
-		rtb.setField(fieldName, schemaField)
+	if myType.Kind() != reflect.Struct {
+		return nil
 	}
-	return rtb
+	st := reflectTypeToSchemaType(myType)
+	return st.GetRecordType()
 }
 
 func reflectTypeToSchemaType(t reflect.Type) *schema_pb.Type {
@@ -50,6 +43,26 @@ func reflectTypeToSchemaType(t reflect.Type) *schema_pb.Type {
 				}
 			}
 		}
+	case reflect.Struct:
+		recordType := &schema_pb.RecordType{}
+		for i := 0; i < t.NumField(); i++ {
+			field := t.Field(i)
+			fieldType := field.Type
+			fieldName := field.Name
+			schemaField := reflectTypeToSchemaType(fieldType)
+			if schemaField == nil {
+				return nil
+			}
+			recordType.Fields = append(recordType.Fields, &schema_pb.Field{
+				Name: fieldName,
+				Type: schemaField,
+			})
+		}
+		return &schema_pb.Type{
+			Kind: &schema_pb.Type_RecordType{
+				RecordType: recordType,
+			},
+		}
 	}
 	return nil
 }
diff --git a/weed/mq/schema/struct_to_schema_test.go b/weed/mq/schema/struct_to_schema_test.go
new file mode 100644
index 000000000..6a3d5687f
--- /dev/null
+++ b/weed/mq/schema/struct_to_schema_test.go
@@ -0,0 +1,115 @@
+package schema
+
+import (
+	"github.com/seaweedfs/seaweedfs/weed/pb/schema_pb"
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+func TestStructToSchema(t *testing.T) {
+	type args struct {
+		instance any
+	}
+	tests := []struct {
+		name string
+		args args
+		want *schema_pb.RecordType
+	}{
+		{
+			name: "scalar type",
+			args: args{
+				instance: 1,
+			},
+			want: nil,
+		},
+		{
+			name: "simple struct type",
+			args: args{
+				instance: struct {
+					Field1 int
+					Field2 string
+				}{},
+			},
+			want: NewRecordTypeBuilder().
+				SetField("Field1", TypeInteger).
+				SetField("Field2", TypeString).
+				Build(),
+		},
+		{
+			name: "simple list",
+			args: args{
+				instance: struct {
+					Field1 []int
+					Field2 string
+				}{},
+			},
+			want: NewRecordTypeBuilder().
+				SetField("Field1", List(TypeInteger)).
+				SetField("Field2", TypeString).
+				Build(),
+		},
+		{
+			name: "simple []byte",
+			args: args{
+				instance: struct {
+					Field2 []byte
+				}{},
+			},
+			want: NewRecordTypeBuilder().
+				SetField("Field2", TypeBytes).
+				Build(),
+		},
+		{
+			name: "nested simpe structs",
+			args: args{
+				instance: struct {
+					Field1 int
+					Field2 struct {
+						Field3 string
+						Field4 int
+					}
+				}{},
+			},
+			want: NewRecordTypeBuilder().
+				SetField("Field1", TypeInteger).
+				SetRecordField("Field2", NewRecordTypeBuilder().
+					SetField("Field3", TypeString).
+					SetField("Field4", TypeInteger),
+				).
+				Build(),
+		},
+		{
+			name: "nested struct type",
+			args: args{
+				instance: struct {
+					Field1 int
+					Field2 struct {
+						Field3 string
+						Field4 []int
+						Field5 struct {
+							Field6 string
+							Field7 []byte
+						}
+					}
+				}{},
+			},
+			want: NewRecordTypeBuilder().
+				SetField("Field1", TypeInteger).
+				SetRecordField("Field2", NewRecordTypeBuilder().
+					SetField("Field3", TypeString).
+					SetField("Field4", List(TypeInteger)).
+					SetRecordField("Field5", NewRecordTypeBuilder().
+						SetField("Field6", TypeString).
+						SetField("Field7", TypeBytes),
+					),
+				).
+				Build(),
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equalf(t, tt.want, StructToSchema(tt.args.instance), "StructToSchema(%v)", tt.args.instance)
+		})
+	}
+}