Merge pull request #592 from marle3003/develop

Develop

Author: marle3003

media/contenttype_test.go

@@ -68,6 +68,15 @@ func TestParseContentType(t *testing.T) {
 				require.False(t, ct.IsRange())
 			},
 		},
+		{
+			s: `application/xml;boundary="foo"`,
+			validate: func(t *testing.T, ct ContentType) {
+				require.Equal(t, "application", ct.Type)
+				require.Equal(t, "xml", ct.Subtype)
+				require.Equal(t, "foo", ct.Parameters["boundary"])
+				require.False(t, ct.IsRange())
+			},
+		},
 	}
 
 	for _, testcase := range testcases {

media/contentype.go

@@ -3,6 +3,7 @@ package media
 import (
 	"fmt"
 	log "github.com/sirupsen/logrus"
+	"mime"
 	"reflect"
 	"strconv"
 	"strings"
@@ -22,32 +23,19 @@ type ContentType struct {
 
 func ParseContentType(s string) ContentType {
 	c := ContentType{raw: s, Parameters: make(map[string]string), Q: 1.0}
-	a := strings.Split(s, ";")
-	m := strings.Split(a[0], "/")
-	c.Type = strings.ToLower(strings.TrimSpace(m[0]))
-	if len(m) > 1 {
-		c.Subtype = strings.ToLower(strings.TrimSpace(m[1]))
-	}
-	for _, p := range a[1:] {
-		kv := strings.Split(p, "=")
-		switch kv[0] {
-		case "q":
-			if len(kv) > 1 {
-				var err error
-				if c.Q, err = strconv.ParseFloat(kv[1], 64); err != nil {
-					log.Debugf("invalid q parameter in %v", s)
-					c.Q = 1.0
-				}
-			}
-		default:
-			if len(kv) > 1 {
-				c.Parameters[strings.TrimSpace(kv[0])] = strings.TrimSpace(kv[1])
-			} else {
-				c.Parameters[kv[0]] = ""
-			}
+	c.Type, c.Parameters, _ = mime.ParseMediaType(s)
+	if q, ok := c.Parameters["q"]; ok {
+		var err error
+		if c.Q, err = strconv.ParseFloat(q, 64); err != nil {
+			log.Debugf("invalid q parameter in %v", s)
+			c.Q = 1.0
 		}
 	}
-
+	t := strings.Split(c.Type, "/")
+	if len(t) > 1 {
+		c.Type = strings.TrimSpace(t[0])
+		c.Subtype = strings.TrimSpace(t[1])
+	}
 	return c
 }
 

providers/asyncapi3/kafka/store/validation.go

@@ -170,7 +170,7 @@ func (mv *messageValidator) Validate(record *kafka.Record) (*KafkaLog, error) {
 			return r, err
 		} else {
 			b, _ := json.Marshal(k)
-			r.Message.Value = string(b)
+			r.Key.Value = string(b)
 		}
 	} else {
 		r.Key.Binary = kafka.Read(record.Key)

providers/asyncapi3/kafka/store/validation_test.go

@@ -76,8 +76,8 @@ func TestValidation(t *testing.T) {
 						asyncapi3test.WithKey(schematest.New("integer")),
 					),
 					asyncapi3test.WithKafkaChannelBinding(asyncapi3.TopicBindings{
-						Partitions:          1,
 						KeySchemaValidation: false,
+						Partitions:          1,
 					}),
 				),
 			),
@@ -92,6 +92,44 @@ func TestValidation(t *testing.T) {
 				})
 				require.NoError(t, err)
 				require.Len(t, batch, 0)
+
+				e := events.GetEvents(events.NewTraits())
+				require.Len(t, e, 1)
+				require.Equal(t, []byte("foo"), e[0].Data.(*store.KafkaLog).Key.Binary)
+			},
+		},
+		{
+			name: "validating value and key",
+			cfg: asyncapi3test.NewConfig(
+				asyncapi3test.WithChannel("foo",
+					asyncapi3test.WithMessage("foo",
+						asyncapi3test.WithPayload(schematest.New("string")),
+						asyncapi3test.WithKey(schematest.New("string")),
+					),
+					asyncapi3test.WithKafkaChannelBinding(asyncapi3.TopicBindings{
+						KeySchemaValidation:   true,
+						ValueSchemaValidation: true,
+						Partitions:            1,
+					}),
+				),
+			),
+			test: func(t *testing.T, s *store.Store) {
+				p := s.Topic("foo").Partition(0)
+				_, batch, err := p.Write(kafka.RecordBatch{
+					Records: []*kafka.Record{
+						{
+							Key:   kafka.NewBytes([]byte("12")),
+							Value: kafka.NewBytes([]byte("foo")),
+						},
+					},
+				})
+				require.NoError(t, err)
+				require.Len(t, batch, 0)
+
+				e := events.GetEvents(events.NewTraits())
+				require.Len(t, e, 1)
+				require.Equal(t, `"12"`, e[0].Data.(*store.KafkaLog).Key.Value)
+				require.Equal(t, `"foo"`, e[0].Data.(*store.KafkaLog).Message.Value)
 			},
 		},
 		{