[feat][schema] Introduce BooleanSchema

pulsar/impl_message.go

@@ -357,8 +357,14 @@ func (msg *message) GetSchemaValue(v interface{}) error {
 		if err != nil {
 			return err
 		}
+		if err := schema.Validate(msg.payLoad); err != nil {
+			return err
+		}
 		return schema.Decode(msg.payLoad, v)
 	}
+	if err := msg.schema.Validate(msg.payLoad); err != nil {
+		return err
+	}
 	return msg.schema.Decode(msg.payLoad, v)
 }
 

pulsar/schema.go

@@ -101,6 +101,8 @@ func NewSchema(schemaType SchemaType, schemaData []byte, properties map[string]s
 		s = NewProtoSchema(schemaDef, properties)
 	case AVRO:
 		s = NewAvroSchema(schemaDef, properties)
+	case BOOLEAN:
+		s = NewBooleanSchema(properties)
 	case INT8:
 		s = NewInt8Schema(properties)
 	case INT16:
@@ -178,7 +180,7 @@ func (js *JSONSchema) Decode(data []byte, v interface{}) error {
 }
 
 func (js *JSONSchema) Validate(message []byte) error {
-	return js.Decode(message, nil)
+	return nil
 }
 
 func (js *JSONSchema) GetSchemaInfo() *SchemaInfo {
@@ -227,7 +229,7 @@ func (ps *ProtoSchema) Decode(data []byte, v interface{}) error {
 }
 
 func (ps *ProtoSchema) Validate(message []byte) error {
-	return ps.Decode(message, nil)
+	return nil
 }
 
 func (ps *ProtoSchema) GetSchemaInfo() *SchemaInfo {
@@ -305,7 +307,7 @@ func (ps *ProtoNativeSchema) Decode(data []byte, v interface{}) error {
 }
 
 func (ps *ProtoNativeSchema) Validate(message []byte) error {
-	return ps.Decode(message, nil)
+	return nil
 }
 
 func (ps *ProtoNativeSchema) GetSchemaInfo() *SchemaInfo {
@@ -377,13 +379,50 @@ func (as *AvroSchema) Decode(data []byte, v interface{}) error {
 }
 
 func (as *AvroSchema) Validate(message []byte) error {
-	return as.Decode(message, nil)
+	return nil
 }
 
 func (as *AvroSchema) GetSchemaInfo() *SchemaInfo {
 	return &as.SchemaInfo
 }
 
+var _ Schema = (*BooleanSchema)(nil)
+
+type BooleanSchema struct {
+	SchemaInfo
+}
+
+func NewBooleanSchema(properties map[string]string) *BooleanSchema {
+	boolSchema := new(BooleanSchema)
+	boolSchema.SchemaInfo.Properties = properties
+	boolSchema.SchemaInfo.Name = "Boolean"
+	boolSchema.SchemaInfo.Type = BOOLEAN
+	boolSchema.SchemaInfo.Schema = ""
+	return boolSchema
+}
+
+func (bs *BooleanSchema) Encode(v interface{}) ([]byte, error) {
+	var buf bytes.Buffer
+	err := WriteElements(&buf, v.(bool))
+	return buf.Bytes(), err
+}
+
+func (bs *BooleanSchema) Decode(data []byte, v interface{}) error {
+	buf := bytes.NewReader(data)
+	return ReadElements(buf, v)
+}
+
+func (bs *BooleanSchema) Validate(message []byte) error {
+	if len(message) != 1 {
+		return newError(InvalidMessage, "size of data received by BooleanSchema is not 1")
+	}
+	return nil
+}
+
+func (bs *BooleanSchema) GetSchemaInfo() *SchemaInfo {
+	return &bs.SchemaInfo
+}
+
 type StringSchema struct {
 	SchemaInfo
 }
@@ -409,7 +448,7 @@ func (ss *StringSchema) Decode(data []byte, v interface{}) error {
 }
 
 func (ss *StringSchema) Validate(message []byte) error {
-	return ss.Decode(message, nil)
+	return nil
 }
 
 func (ss *StringSchema) GetSchemaInfo() *SchemaInfo {
@@ -439,7 +478,7 @@ func (bs *BytesSchema) Decode(data []byte, v interface{}) error {
 }
 
 func (bs *BytesSchema) Validate(message []byte) error {
-	return bs.Decode(message, nil)
+	return nil
 }
 
 func (bs *BytesSchema) GetSchemaInfo() *SchemaInfo {

pulsar/schema_test.go

@@ -492,3 +492,37 @@ func TestDoubleSchema(t *testing.T) {
 	assert.Equal(t, res, float64(1))
 	defer consumer.Close()
 }
+
+func TestBooleanSchema(t *testing.T) {
+	client := createClient()
+	defer client.Close()
+
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:  "booleanTopic",
+		Schema: NewBooleanSchema(nil),
+	})
+	assert.Nil(t, err)
+	ctx := context.Background()
+	if _, err := producer.Send(ctx, &ProducerMessage{
+		Value: true,
+	}); err != nil {
+		log.Fatal(err)
+	}
+	defer producer.Close()
+
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:                       "booleanTopic",
+		SubscriptionName:            "sub-2",
+		Schema:                      NewBooleanSchema(nil),
+		SubscriptionInitialPosition: SubscriptionPositionEarliest,
+	})
+	assert.Nil(t, err)
+
+	var res bool
+	msg, err := consumer.Receive(ctx)
+	assert.Nil(t, err)
+	err = msg.GetSchemaValue(&res)
+	assert.Nil(t, err)
+	assert.Equal(t, res, true)
+	defer consumer.Close()
+}