Merge pull request #642 from marle3003/develop

Develop

Author: marle3003

config/dynamic/asyncApi/convert.go

@@ -132,9 +132,15 @@ func convertChannel(name string, c *Channel, config *asyncapi3.Config) (*asyncap
 }
 
 func addOperation(msgName, action string, op *Operation, ch *asyncapi3.ChannelRef, msg *asyncapi3.MessageRef, config *asyncapi3.Config) {
-	name := fmt.Sprintf("%s_%s", action, msgName)
+	var opName string
+	if msgName == "publish" || msgName == "subscribe" {
+		opName = fmt.Sprintf("%s_%s_%s", ch.Value.GetName(), action, msgName)
+	} else {
+		opName = fmt.Sprintf("%s_%s", action, msgName)
+	}
+
 	if len(op.OperationId) > 0 {
-		name = op.OperationId
+		opName = op.OperationId
 	}
 
 	result := &asyncapi3.Operation{
@@ -150,7 +156,7 @@ func addOperation(msgName, action string, op *Operation, ch *asyncapi3.ChannelRe
 		config.Operations = map[string]*asyncapi3.OperationRef{}
 	}
 
-	config.Operations[name] = &asyncapi3.OperationRef{Value: result}
+	config.Operations[opName] = &asyncapi3.OperationRef{Value: result}
 }
 
 func addMessage(target *asyncapi3.Channel, msg *asyncapi3.MessageRef, opId, ref, opName string) string {

config/dynamic/asyncApi/convert_test.go

@@ -74,3 +74,47 @@ func TestConfig_Convert(t *testing.T) {
 	require.Equal(t, channel, op.Channel.Value)
 	require.Equal(t, "Inform about environmental lighting conditions of a particular streetlight.", op.Summary)
 }
+
+func TestConfig_ConvertNoOperationId(t *testing.T) {
+	s := `
+asyncapi: '2.6.0'
+info:
+  title: Streetlights API
+channels:
+  foo:
+    publish:
+      message:
+        payload:
+          schema:
+            type: string
+  bar:
+    publish:
+      message:
+        payload:
+          schema:
+            type: string
+`
+	var old *Config
+	err := yaml.Unmarshal([]byte(s), &old)
+	require.NoError(t, err)
+	cfg, err := old.Convert()
+	require.NoError(t, err)
+
+	require.Len(t, cfg.Channels, 2)
+	require.Contains(t, cfg.Channels, "foo")
+	require.Contains(t, cfg.Channels, "bar")
+
+	require.Len(t, cfg.Channels["foo"].Value.Messages, 1)
+	require.Contains(t, cfg.Channels["foo"].Value.Messages, "publish")
+	require.Len(t, cfg.Channels["bar"].Value.Messages, 1)
+	require.Contains(t, cfg.Channels["bar"].Value.Messages, "publish")
+
+	require.Len(t, cfg.Operations, 2)
+	require.Contains(t, cfg.Operations, "foo_send_publish")
+	require.Contains(t, cfg.Operations, "bar_send_publish")
+
+	require.Len(t, cfg.Operations["foo_send_publish"].Value.Messages, 1)
+	require.Equal(t, cfg.Channels["foo"].Value.Messages["publish"], cfg.Operations["foo_send_publish"].Value.Messages[0])
+	require.Len(t, cfg.Operations["bar_send_publish"].Value.Messages, 1)
+	require.Equal(t, cfg.Channels["bar"].Value.Messages["publish"], cfg.Operations["bar_send_publish"].Value.Messages[0])
+}

engine/kafka.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"fmt"
 	log "github.com/sirupsen/logrus"
+	"maps"
 	"math/rand"
 	"mokapi/engine/common"
 	"mokapi/kafka"
@@ -16,6 +17,7 @@ import (
 	"mokapi/schema/encoding"
 	"mokapi/schema/json/generator"
 	"mokapi/schema/json/schema"
+	"slices"
 	"strings"
 	"time"
 )
@@ -165,24 +167,38 @@ func (c *KafkaClient) getPartition(t *store.Topic, partition int) (*store.Partit
 }
 
 func (c *KafkaClient) createRecordBatch(key, value interface{}, headers map[string]interface{}, topic *asyncapi3.Channel, config *asyncapi3.Config) (rb kafka.RecordBatch, err error) {
-	n := len(topic.Messages)
-	if n == 0 {
-		err = fmt.Errorf("message configuration missing")
-		return
+	contentType := config.DefaultContentType
+	var payload *asyncapi3.SchemaRef
+	keySchema := &asyncapi3.SchemaRef{
+		Value: &asyncapi3.MultiSchemaFormat{
+			Schema: &schema.Schema{Type: schema.Types{"string"}, Pattern: "[a-z]{9}"},
+		},
 	}
+	var headerSchema *schema.Schema
 
-	msg, err := selectMessage(value, topic.Name, config)
-	if err != nil {
-		return
-	}
+	if len(topic.Messages) > 0 {
+		var msg *asyncapi3.Message
+		msg, err = selectMessage(value, topic, config)
+		if err != nil {
+			return
+		}
+		payload = msg.Payload
+
+		if msg.Bindings.Kafka.Key != nil {
+			keySchema = msg.Bindings.Kafka.Key
+		}
+
+		if msg.ContentType != "" {
+			contentType = msg.ContentType
+		}
 
-	keySchema := msg.Bindings.Kafka.Key
-	if keySchema == nil {
-		// use default key schema
-		keySchema = &asyncapi3.SchemaRef{
-			Value: &asyncapi3.MultiSchemaFormat{
-				Schema: &schema.Schema{Type: schema.Types{"string"}, Pattern: "[a-z]{9}"},
-			},
+		if msg.Headers != nil {
+			var ok bool
+			headerSchema, ok = msg.Headers.Value.Schema.(*schema.Schema)
+			if !ok {
+				err = fmt.Errorf("currently only json schema supported")
+				return
+			}
 		}
 	}
 
@@ -194,16 +210,13 @@ func (c *KafkaClient) createRecordBatch(key, value interface{}, headers map[stri
 	}
 
 	if value == nil {
-		value, err = createValue(msg.Payload)
+		value, err = createValue(payload)
 		if err != nil {
 			return rb, fmt.Errorf("unable to generate kafka value: %v", err)
 		}
 	}
 
-	contentType := config.DefaultContentType
-	if msg.ContentType != "" {
-		contentType = msg.ContentType
-	} else if contentType == "" {
+	if contentType == "" {
 		// set default: https://github.com/asyncapi/spec/issues/319
 		contentType = "application/json"
 	}
@@ -212,7 +225,7 @@ func (c *KafkaClient) createRecordBatch(key, value interface{}, headers map[stri
 	if b, ok := value.([]byte); ok {
 		v = b
 	} else {
-		v, err = marshal(value, msg.Payload, contentType)
+		v, err = marshal(value, payload, contentType)
 		if err != nil {
 			return
 		}
@@ -229,16 +242,7 @@ func (c *KafkaClient) createRecordBatch(key, value interface{}, headers map[stri
 	}
 
 	var recordHeaders []kafka.RecordHeader
-	var hs *schema.Schema
-	if msg.Headers != nil {
-		var ok bool
-		hs, ok = msg.Headers.Value.Schema.(*schema.Schema)
-		if !ok {
-			err = fmt.Errorf("currently only json schema supported")
-			return
-		}
-	}
-	recordHeaders, err = getHeaders(headers, hs)
+	recordHeaders, err = getHeaders(headers, headerSchema)
 	if err != nil {
 		return
 	}
@@ -352,22 +356,24 @@ func marshalKey(key interface{}, r *asyncapi3.SchemaRef) ([]byte, error) {
 	}
 }
 
-func selectMessage(value any, topic string, cfg *asyncapi3.Config) (*asyncapi3.Message, error) {
+func selectMessage(value any, topic *asyncapi3.Channel, cfg *asyncapi3.Config) (*asyncapi3.Message, error) {
 	noOperationDefined := true
 
 	// first try to get send operation
-	for name, op := range cfg.Operations {
+	for _, op := range cfg.Operations {
 		if op.Value == nil || op.Value.Channel.Value == nil {
 			continue
 		}
-		if op.Value.Channel.Value.Name == topic && op.Value.Action == "send" {
+		if op.Value.Channel.Value == topic && op.Value.Action == "send" {
 			noOperationDefined = false
+			var messages []*asyncapi3.MessageRef
 			if len(op.Value.Messages) == 0 {
-				log.Warnf("no message defined for operation %s", name)
+				messages = slices.Collect(maps.Values(op.Value.Channel.Value.Messages))
+			} else {
+				messages = op.Value.Messages
 			}
-			for _, msg := range op.Value.Messages {
+			for _, msg := range messages {
 				if msg.Value == nil {
-					log.Errorf("no message defined for operation %s", name)
 					continue
 				}
 				if valueMatchMessagePayload(value, msg.Value) {
@@ -378,18 +384,20 @@ func selectMessage(value any, topic string, cfg *asyncapi3.Config) (*asyncapi3.M
 	}
 
 	// second, try to get receive operation
-	for name, op := range cfg.Operations {
+	for _, op := range cfg.Operations {
 		if op.Value == nil || op.Value.Channel.Value == nil {
 			continue
 		}
-		if op.Value.Channel.Value.Name == topic && op.Value.Action == "receive" {
+		if op.Value.Channel.Value == topic && op.Value.Action == "receive" {
 			noOperationDefined = false
+			var messages []*asyncapi3.MessageRef
 			if len(op.Value.Messages) == 0 {
-				log.Errorf("no message defined for operation %s", name)
+				messages = slices.Collect(maps.Values(op.Value.Channel.Value.Messages))
+			} else {
+				messages = op.Value.Messages
 			}
-			for _, msg := range op.Value.Messages {
+			for _, msg := range messages {
 				if msg.Value == nil {
-					log.Warnf("no message defined for operation %s", name)
 					continue
 				}
 				if valueMatchMessagePayload(value, msg.Value) {
@@ -404,13 +412,13 @@ func selectMessage(value any, topic string, cfg *asyncapi3.Config) (*asyncapi3.M
 	}
 
 	if value != nil {
-		return nil, fmt.Errorf("no matching 'send' or 'receive' operation found for value: %v", value)
+		return nil, fmt.Errorf("no message configuration matches the message value for topic '%s' and value: %v", topic.GetName(), value)
 	}
-	return nil, fmt.Errorf("no matching 'send' or 'receive' operation defined")
+	return nil, fmt.Errorf("no message ")
 }
 
 func valueMatchMessagePayload(value any, msg *asyncapi3.Message) bool {
-	if value == nil {
+	if value == nil || msg.Payload == nil {
 		return true
 	}
 

engine/kafka_test.go

@@ -3,7 +3,6 @@ package engine_test
 import (
 	"bytes"
 	"fmt"
-	"github.com/brianvoe/gofakeit/v6"
 	"github.com/sirupsen/logrus"
 	"github.com/sirupsen/logrus/hooks/test"
 	"github.com/stretchr/testify/require"
@@ -16,6 +15,7 @@ import (
 	"mokapi/providers/asyncapi3"
 	"mokapi/providers/asyncapi3/asyncapi3test"
 	"mokapi/runtime"
+	"mokapi/schema/json/generator"
 	"mokapi/schema/json/schema/schematest"
 	"net/url"
 	"testing"
@@ -499,7 +499,7 @@ func TestKafkaClient(t *testing.T) {
 						produce({ topic: 'foo', messages: [{ data: 12 }] })
 					}
 				`))
-				require.EqualError(t, err, "producing kafka message to 'foo' failed: no matching 'send' or 'receive' operation found for value: 12 at mokapi/js/kafka.(*Module).Produce-fm (native)")
+				require.EqualError(t, err, "producing kafka message to 'foo' failed: no message configuration matches the message value for topic 'foo' and value: 12 at mokapi/js/kafka.(*Module).Produce-fm (native)")
 
 				b, errCode := app.Kafka.Get("foo").Store.Topic("foo").Partition(0).Read(0, 1000)
 				require.Equal(t, kafka.None, errCode)
@@ -508,7 +508,7 @@ func TestKafkaClient(t *testing.T) {
 
 				// logs
 				require.Len(t, hook.Entries, 2)
-				require.Equal(t, "js error: producing kafka message to 'foo' failed: no matching 'send' or 'receive' operation found for value: 12 in test.js", hook.LastEntry().Message)
+				require.Equal(t, "js error: producing kafka message to 'foo' failed: no message configuration matches the message value for topic 'foo' and value: 12 in test.js", hook.LastEntry().Message)
 			},
 		},
 		{
@@ -566,11 +566,91 @@ func TestKafkaClient(t *testing.T) {
 				require.Contains(t, msg, "kafka topic 'retry' not found. Retry in 1s")
 			},
 		},
+		{
+			name: "channel does not define a message",
+			cfg: func() *asyncapi3.Config {
+				ch := asyncapi3test.NewChannel()
+
+				return asyncapi3test.NewConfig(
+					asyncapi3test.WithInfo("foo", "", ""),
+					asyncapi3test.AddChannel("foo", ch),
+					asyncapi3test.WithOperation("sendAction",
+						asyncapi3test.WithOperationAction("send"),
+						asyncapi3test.WithOperationChannel(ch),
+					),
+				)
+			},
+			test: func(t *testing.T, e *engine.Engine, app *runtime.App) {
+				err := e.AddScript(newScript("test.js", `
+					import { produce } from 'mokapi/kafka'
+					export default function() {
+						produce({ topic: 'foo', messages: [{ data: 'foo' }] })
+					}
+				`))
+
+				require.NoError(t, err)
+				b, errCode := app.Kafka.Get("foo").Store.Topic("foo").Partition(0).Read(0, 1000)
+				require.Equal(t, kafka.None, errCode)
+				require.NotNil(t, b)
+				require.Equal(t, "gbrmarxhk", kafka.BytesToString(b.Records[0].Key))
+				require.Equal(t, `"foo"`, kafka.BytesToString(b.Records[0].Value))
+
+				err = e.AddScript(newScript("test.js", `
+					import { produce } from 'mokapi/kafka'
+					export default function() {
+						produce({ topic: 'foo', messages: [{  }] })
+					}
+				`))
+
+				require.NoError(t, err)
+				b, errCode = app.Kafka.Get("foo").Store.Topic("foo").Partition(0).Read(1, 1000)
+				require.Equal(t, kafka.None, errCode)
+				require.NotNil(t, b)
+				require.Equal(t, "ijbptapwy", kafka.BytesToString(b.Records[0].Key))
+				require.Equal(t, `""`, kafka.BytesToString(b.Records[0].Value))
+			},
+		},
+		{
+			name: "no payload is defined and produce with value and nil",
+			cfg: func() *asyncapi3.Config {
+				msg := asyncapi3test.NewMessage()
+				return createCfg("foo", msg)
+			},
+			test: func(t *testing.T, e *engine.Engine, app *runtime.App) {
+				err := e.AddScript(newScript("test.js", `
+					import { produce } from 'mokapi/kafka'
+					export default function() {
+						produce({ topic: 'foo', messages: [{ data: 'foo' }] })
+					}
+				`))
+
+				require.NoError(t, err)
+				b, errCode := app.Kafka.Get("foo").Store.Topic("foo").Partition(0).Read(0, 1000)
+				require.Equal(t, kafka.None, errCode)
+				require.NotNil(t, b)
+				require.Equal(t, "gbrmarxhk", kafka.BytesToString(b.Records[0].Key))
+				require.Equal(t, `"foo"`, kafka.BytesToString(b.Records[0].Value))
+
+				err = e.AddScript(newScript("test.js", `
+					import { produce } from 'mokapi/kafka'
+					export default function() {
+						produce({ topic: 'foo', messages: [{  }] })
+					}
+				`))
+
+				require.NoError(t, err)
+				b, errCode = app.Kafka.Get("foo").Store.Topic("foo").Partition(0).Read(1, 1000)
+				require.Equal(t, kafka.None, errCode)
+				require.NotNil(t, b)
+				require.Equal(t, "ijbptapwy", kafka.BytesToString(b.Records[0].Key))
+				require.Equal(t, `""`, kafka.BytesToString(b.Records[0].Value))
+			},
+		},
 	}
 
 	for _, tc := range testcases {
 		t.Run(tc.name, func(t *testing.T) {
-			gofakeit.Seed(11)
+			generator.Seed(11)
 
 			app := runtime.New(&static.Config{})
 			e := enginetest.NewEngine(