Develop

acceptance/cmd_test.go

@@ -37,10 +37,10 @@ func Start(cfg *static.Config) (*Cmd, error) {
 	feature.Enable(cfg.Features)
 
 	registerDynamicTypes()
-	app := runtime.New(cfg)
-	generator.SetConfig(cfg.DataGen)
 
 	watcher := server.NewConfigWatcher(cfg)
+	app := runtime.New(cfg, watcher)
+	generator.SetConfig(cfg.DataGen)
 
 	certStore, err := cert.NewStore(cfg)
 	if err != nil {

api/handler_events_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"mokapi/config/dynamic/dynamictest"
 	"mokapi/config/static"
 	"mokapi/providers/asyncapi3/kafka/store"
 	"mokapi/providers/openapi"
@@ -202,7 +203,7 @@ func TestHandler_Events(t *testing.T) {
 		tc := tc
 		t.Run(tc.name, func(t *testing.T) {
 			cfg := &static.Config{}
-			app := runtime.New(cfg)
+			app := runtime.New(cfg, &dynamictest.Reader{})
 
 			h := New(app, static.Api{})
 			tc.fn(t, h, app.Events)
@@ -330,7 +331,7 @@ func TestHandler_KafkaEvents(t *testing.T) {
 		tc := tc
 		t.Run(tc.name, func(t *testing.T) {
 			cfg := &static.Config{}
-			app := runtime.New(cfg)
+			app := runtime.New(cfg, &dynamictest.Reader{})
 
 			h := New(app, static.Api{})
 			tc.fn(t, h, app.Events)

api/handler_fileserver_test.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"mokapi/api"
 	"mokapi/config/dynamic"
+	"mokapi/config/dynamic/dynamictest"
 	"mokapi/config/static"
 	"mokapi/providers/openapi/openapitest"
 	"mokapi/runtime"
@@ -131,7 +132,7 @@ func TestHandler_FileServer(t *testing.T) {
 			t.Parallel()
 
 			cfg := &static.Config{}
-			h := api.New(runtime.New(cfg), tc.config)
+			h := api.New(runtime.New(cfg, &dynamictest.Reader{}), tc.config)
 			tc.fn(t, h)
 		})
 	}
@@ -146,7 +147,7 @@ func TestOpenGraphInDashboard(t *testing.T) {
 			name: "http service",
 			test: func(t *testing.T) {
 				cfg := &static.Config{}
-				app := runtime.New(cfg)
+				app := runtime.New(cfg, &dynamictest.Reader{})
 				app.AddHttp(&dynamic.Config{Info: dynamic.ConfigInfo{Url: mustParse("https://foo.bar")}, Data: openapitest.NewConfig("3.0", openapitest.WithInfo("Swagger Petstore", "1.0", "This is a sample server Petstore server."))})
 				h := api.New(app, static.Api{Path: "/mokapi", Dashboard: true})
 				try.Handler(t,
@@ -166,7 +167,7 @@ func TestOpenGraphInDashboard(t *testing.T) {
 			name: "http service path without summary and description",
 			test: func(t *testing.T) {
 				cfg := &static.Config{}
-				app := runtime.New(cfg)
+				app := runtime.New(cfg, &dynamictest.Reader{})
 				app.AddHttp(&dynamic.Config{Info: dynamic.ConfigInfo{Url: mustParse("https://foo.bar")}, Data: openapitest.NewConfig("3.0",
 					openapitest.WithInfo("Swagger Petstore", "1.0", "This is a sample server Petstore server."),
 					openapitest.WithPath("/pet/{petId}"),
@@ -190,7 +191,7 @@ func TestOpenGraphInDashboard(t *testing.T) {
 			name: "http service path with summary and description",
 			test: func(t *testing.T) {
 				cfg := &static.Config{}
-				app := runtime.New(cfg)
+				app := runtime.New(cfg, &dynamictest.Reader{})
 				app.AddHttp(&dynamic.Config{Info: dynamic.ConfigInfo{Url: mustParse("https://foo.bar")}, Data: openapitest.NewConfig("3.0",
 					openapitest.WithInfo("Swagger Petstore", "1.0", "This is a sample server Petstore server."),
 					openapitest.WithPath("/pet/{petId}",
@@ -216,7 +217,7 @@ func TestOpenGraphInDashboard(t *testing.T) {
 			name: "http service path with no summary but description",
 			test: func(t *testing.T) {
 				cfg := &static.Config{}
-				app := runtime.New(cfg)
+				app := runtime.New(cfg, &dynamictest.Reader{})
 				app.AddHttp(&dynamic.Config{Info: dynamic.ConfigInfo{Url: mustParse("https://foo.bar")}, Data: openapitest.NewConfig("3.0",
 					openapitest.WithInfo("Swagger Petstore", "1.0", "This is a sample server Petstore server."),
 					openapitest.WithPath("/pet/{petId}",
@@ -241,7 +242,7 @@ func TestOpenGraphInDashboard(t *testing.T) {
 			name: "http service endpoint no summary and no description",
 			test: func(t *testing.T) {
 				cfg := &static.Config{}
-				app := runtime.New(cfg)
+				app := runtime.New(cfg, &dynamictest.Reader{})
 				app.AddHttp(&dynamic.Config{Info: dynamic.ConfigInfo{Url: mustParse("https://foo.bar")}, Data: openapitest.NewConfig("3.0",
 					openapitest.WithInfo("Swagger Petstore", "1.0", "This is a sample server Petstore server."),
 					openapitest.WithPath("/pet/{petId}",
@@ -266,7 +267,7 @@ func TestOpenGraphInDashboard(t *testing.T) {
 			name: "http service endpoint get right path",
 			test: func(t *testing.T) {
 				cfg := &static.Config{}
-				app := runtime.New(cfg)
+				app := runtime.New(cfg, &dynamictest.Reader{})
 				app.AddHttp(&dynamic.Config{Info: dynamic.ConfigInfo{Url: mustParse("https://foo.bar")}, Data: openapitest.NewConfig("3.0",
 					openapitest.WithInfo("Swagger Petstore", "1.0", "This is a sample server Petstore server."),
 					openapitest.WithPath("/pet/{petId}",

api/handler_http_test.go

@@ -66,7 +66,7 @@ func TestHandler_Http(t *testing.T) {
 		{
 			name: "get http service info",
 			app: func() *runtime.App {
-				app := runtime.New(&static.Config{})
+				app := runtime.New(&static.Config{}, &dynamictest.Reader{})
 				cfg := &dynamic.Config{
 					Info: dynamictest.NewConfigInfo(), Data: openapitest.NewConfig("3.0.0",
 						openapitest.WithInfo("foo", "1.0", "bar"),
@@ -328,7 +328,7 @@ func TestHandler_Http(t *testing.T) {
 
 func TestHandler_Http_NotFound(t *testing.T) {
 	cfg := &static.Config{}
-	h := New(runtime.New(cfg), static.Api{})
+	h := New(runtime.New(cfg, &dynamictest.Reader{}), static.Api{})
 
 	try.Handler(t,
 		http.MethodGet,

api/handler_kafka.go

@@ -17,6 +17,7 @@ import (
 	"time"
 
 	"github.com/pkg/errors"
+	log "github.com/sirupsen/logrus"
 )
 
 type kafkaSummary struct {
@@ -564,18 +565,38 @@ func newTopic(t *store.Topic, ch *asyncapi3.Channel, cfg *asyncapi3.Config) topi
 		}
 
 		if msg.Payload != nil && msg.Payload.Value != nil {
-			m.Payload = &schemaInfo{Schema: msg.Payload.Value.Schema, Format: msg.Payload.Value.Format}
+			format := ""
+			if msf, ok := msg.Payload.Value.(*asyncapi3.MultiSchemaFormat); ok {
+				format = msf.Format
+			}
+			s, err := msg.Payload.GetSchema()
+			if err != nil {
+				log.Errorf("failed to get schema for message in topic '%s': %v", t.Name, err)
+			}
+			m.Payload = &schemaInfo{Schema: s, Format: format}
 		}
 		if msg.Headers != nil && msg.Headers.Value != nil {
-			m.Header = &schemaInfo{Schema: msg.Headers.Value.Schema, Format: msg.Headers.Value.Format}
+			format := ""
+			if msf, ok := msg.Headers.Value.(*asyncapi3.MultiSchemaFormat); ok {
+				format = msf.Format
+			}
+			s, err := msg.Headers.GetSchema()
+			if err != nil {
+				log.Errorf("failed to get schema for headers in topic '%s': %v", t.Name, err)
+			}
+			m.Header = &schemaInfo{Schema: s, Format: format}
 		}
 
 		if m.ContentType == "" {
 			m.ContentType = cfg.DefaultContentType
 		}
 
 		if msg.Bindings.Kafka.Key != nil {
-			m.Key = &schemaInfo{Schema: msg.Bindings.Kafka.Key.Value.Schema}
+			s, err := msg.Bindings.Kafka.Key.GetSchema()
+			if err != nil {
+				log.Errorf("failed to get schema for key in topic '%s': %v", t.Name, err)
+			}
+			m.Key = &schemaInfo{Schema: s}
 		}
 		if result.Messages == nil {
 			result.Messages = map[string]messageConfig{}

api/handler_kafka_test.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"mokapi/api"
 	"mokapi/config/dynamic"
+	"mokapi/config/dynamic/asyncApi"
 	"mokapi/config/dynamic/dynamictest"
 	"mokapi/config/static"
 	"mokapi/engine/enginetest"
@@ -57,7 +58,7 @@ func TestHandler_Kafka(t *testing.T) {
 		{
 			name: "get kafka services",
 			app: func() *runtime.App {
-				app := runtime.New(&static.Config{})
+				app := runtime.New(&static.Config{}, &dynamictest.Reader{})
 				_, _ = app.Kafka.Add(&dynamic.Config{
 					Info: dynamic.ConfigInfo{Url: try.MustUrl("kafka.yaml")},
 					Data: asyncapi3test.NewConfig(
@@ -91,7 +92,7 @@ func TestHandler_Kafka(t *testing.T) {
 		{
 			name: "get specific",
 			app: func() *runtime.App {
-				app := runtime.New(&static.Config{})
+				app := runtime.New(&static.Config{}, &dynamictest.Reader{})
 				cfg := &dynamic.Config{
 					Info: dynamictest.NewConfigInfo(),
 					Data: asyncapi3test.NewConfig(
@@ -221,7 +222,7 @@ func TestHandler_Kafka(t *testing.T) {
 		{
 			name: "get specific with group",
 			app: func() *runtime.App {
-				app := runtime.New(&static.Config{})
+				app := runtime.New(&static.Config{}, &dynamictest.Reader{})
 				app.Kafka.Set("foo", getKafkaInfoWithGroup(asyncapi3test.NewConfig(
 					asyncapi3test.WithInfo("foo", "bar", "1.0"),
 					asyncapi3test.WithServer("foo", "kafka", "foo.bar"),
@@ -246,7 +247,7 @@ func TestHandler_Kafka(t *testing.T) {
 		{
 			name: "get specific with group no generation",
 			app: func() *runtime.App {
-				app := runtime.New(&static.Config{})
+				app := runtime.New(&static.Config{}, &dynamictest.Reader{})
 				app.Kafka.Set("foo", getKafkaInfoWithGroup(asyncapi3test.NewConfig(
 					asyncapi3test.WithInfo("foo", "bar", "1.0"),
 					asyncapi3test.WithServer("foo", "kafka", "foo.bar"),
@@ -273,7 +274,7 @@ func TestHandler_Kafka(t *testing.T) {
 					return t1
 				}
 
-				app := runtime.New(&static.Config{})
+				app := runtime.New(&static.Config{}, &dynamictest.Reader{})
 				app.Kafka.Set("foo", getKafkaInfoWithGroup(asyncapi3test.NewConfig(
 					asyncapi3test.WithInfo("foo", "bar", "1.0"),
 					asyncapi3test.WithServer("foo", "kafka", "foo.bar"),
@@ -321,7 +322,7 @@ func TestHandler_Kafka(t *testing.T) {
 		{
 			name: "get specific with topic and openapi schema",
 			app: func() *runtime.App {
-				app := runtime.New(&static.Config{})
+				app := runtime.New(&static.Config{}, &dynamictest.Reader{})
 				app.Kafka.Set("foo", getKafkaInfo(asyncapi3test.NewConfig(
 					asyncapi3test.WithInfo("foo", "bar", "1.0"),
 					asyncapi3test.WithChannel("foo",
@@ -369,7 +370,7 @@ func TestHandler_KafkaAPI(t *testing.T) {
 		{
 			name: "get kafka topics but empty",
 			app: func() *runtime.App {
-				app := runtime.New(&static.Config{})
+				app := runtime.New(&static.Config{}, &dynamictest.Reader{})
 				_, _ = app.Kafka.Add(&dynamic.Config{
 					Info: dynamic.ConfigInfo{Url: try.MustUrl("kafka.yaml")},
 					Data: asyncapi3test.NewConfig(
@@ -394,7 +395,7 @@ func TestHandler_KafkaAPI(t *testing.T) {
 		{
 			name: "get kafka topics with one topic",
 			app: func() *runtime.App {
-				app := runtime.New(&static.Config{})
+				app := runtime.New(&static.Config{}, &dynamictest.Reader{})
 				_, _ = app.Kafka.Add(&dynamic.Config{
 					Info: dynamic.ConfigInfo{Url: try.MustUrl("kafka.yaml")},
 					Data: asyncapi3test.NewConfig(
@@ -424,7 +425,7 @@ func TestHandler_KafkaAPI(t *testing.T) {
 		{
 			name: "get specific kafka topic",
 			app: func() *runtime.App {
-				app := runtime.New(&static.Config{})
+				app := runtime.New(&static.Config{}, &dynamictest.Reader{})
 				_, _ = app.Kafka.Add(&dynamic.Config{
 					Info: dynamic.ConfigInfo{Url: try.MustUrl("kafka.yaml")},
 					Data: asyncapi3test.NewConfig(
@@ -454,7 +455,7 @@ func TestHandler_KafkaAPI(t *testing.T) {
 		{
 			name: "get specific kafka topic but not found",
 			app: func() *runtime.App {
-				app := runtime.New(&static.Config{})
+				app := runtime.New(&static.Config{}, &dynamictest.Reader{})
 				_, _ = app.Kafka.Add(&dynamic.Config{
 					Info: dynamic.ConfigInfo{Url: try.MustUrl("kafka.yaml")},
 					Data: asyncapi3test.NewConfig(
@@ -482,7 +483,7 @@ func TestHandler_KafkaAPI(t *testing.T) {
 		{
 			name: "produce kafka message into topic",
 			app: func() *runtime.App {
-				app := runtime.New(&static.Config{})
+				app := runtime.New(&static.Config{}, &dynamictest.Reader{})
 
 				msg := asyncapi3test.NewMessage(
 					asyncapi3test.WithContentType("application/json"),
@@ -532,7 +533,7 @@ func TestHandler_KafkaAPI(t *testing.T) {
 		{
 			name: "produce kafka message into topic using binary",
 			app: func() *runtime.App {
-				app := runtime.New(&static.Config{})
+				app := runtime.New(&static.Config{}, &dynamictest.Reader{})
 				_, _ = app.Kafka.Add(&dynamic.Config{
 					Info: dynamic.ConfigInfo{Url: try.MustUrl("kafka.yaml")},
 					Data: asyncapi3test.NewConfig(
@@ -571,7 +572,7 @@ func TestHandler_KafkaAPI(t *testing.T) {
 		{
 			name: "produce invalid kafka message into topic",
 			app: func() *runtime.App {
-				app := runtime.New(&static.Config{})
+				app := runtime.New(&static.Config{}, &dynamictest.Reader{})
 
 				msg := asyncapi3test.NewMessage(
 					asyncapi3test.WithContentType("application/json"),
@@ -627,7 +628,7 @@ func TestHandler_KafkaAPI(t *testing.T) {
 		{
 			name: "get kafka partitions",
 			app: func() *runtime.App {
-				app := runtime.New(&static.Config{})
+				app := runtime.New(&static.Config{}, &dynamictest.Reader{})
 				_, _ = app.Kafka.Add(&dynamic.Config{
 					Info: dynamic.ConfigInfo{Url: try.MustUrl("kafka.yaml")},
 					Data: asyncapi3test.NewConfig(
@@ -656,7 +657,7 @@ func TestHandler_KafkaAPI(t *testing.T) {
 		{
 			name: "get specific kafka partition",
 			app: func() *runtime.App {
-				app := runtime.New(&static.Config{})
+				app := runtime.New(&static.Config{}, &dynamictest.Reader{})
 				_, _ = app.Kafka.Add(&dynamic.Config{
 					Info: dynamic.ConfigInfo{Url: try.MustUrl("kafka.yaml")},
 					Data: asyncapi3test.NewConfig(
@@ -685,7 +686,7 @@ func TestHandler_KafkaAPI(t *testing.T) {
 		{
 			name: "produce kafka message into specific partition",
 			app: func() *runtime.App {
-				app := runtime.New(&static.Config{})
+				app := runtime.New(&static.Config{}, &dynamictest.Reader{})
 
 				msg := asyncapi3test.NewMessage(
 					asyncapi3test.WithContentType("application/json"),
@@ -739,7 +740,7 @@ func TestHandler_KafkaAPI(t *testing.T) {
 		{
 			name: "produce kafka message into specific partition using XML",
 			app: func() *runtime.App {
-				app := runtime.New(&static.Config{})
+				app := runtime.New(&static.Config{}, &dynamictest.Reader{})
 
 				msg := asyncapi3test.NewMessage(
 					asyncapi3test.WithContentType("application/xml"),
@@ -801,7 +802,7 @@ func TestHandler_KafkaAPI(t *testing.T) {
 		{
 			name: "produce kafka message into specific partition using plain XML string",
 			app: func() *runtime.App {
-				app := runtime.New(&static.Config{})
+				app := runtime.New(&static.Config{}, &dynamictest.Reader{})
 
 				msg := asyncapi3test.NewMessage(
 					asyncapi3test.WithContentType("application/xml"),
@@ -853,7 +854,7 @@ func TestHandler_KafkaAPI(t *testing.T) {
 		{
 			name: "produce invalid kafka message into specific partition using plain XML string",
 			app: func() *runtime.App {
-				app := runtime.New(&static.Config{})
+				app := runtime.New(&static.Config{}, &dynamictest.Reader{})
 
 				msg := asyncapi3test.NewMessage(
 					asyncapi3test.WithContentType("application/xml"),
@@ -912,7 +913,7 @@ func TestHandler_KafkaAPI(t *testing.T) {
 		{
 			name: "get records",
 			app: func() *runtime.App {
-				app := runtime.New(&static.Config{})
+				app := runtime.New(&static.Config{}, &dynamictest.Reader{})
 
 				msg := asyncapi3test.NewMessage(
 					asyncapi3test.WithContentType("application/json"),
@@ -970,7 +971,7 @@ func TestHandler_KafkaAPI(t *testing.T) {
 		{
 			name: "get specific record",
 			app: func() *runtime.App {
-				app := runtime.New(&static.Config{})
+				app := runtime.New(&static.Config{}, &dynamictest.Reader{})
 
 				msg := asyncapi3test.NewMessage(
 					asyncapi3test.WithContentType("application/json"),
@@ -1035,7 +1036,7 @@ func TestHandler_KafkaAPI(t *testing.T) {
 }
 
 func TestHandler_Kafka_NotFound(t *testing.T) {
-	h := api.New(runtime.New(&static.Config{}), static.Api{})
+	h := api.New(runtime.New(&static.Config{}, &dynamictest.Reader{}), static.Api{})
 
 	try.Handler(t,
 		http.MethodGet,
@@ -1096,6 +1097,13 @@ func TestHandler_Kafka_Metrics(t *testing.T) {
 	}
 }
 
+func Test_IsAsyncApiConfig(t *testing.T) {
+	v2 := &asyncApi.Config{Info: asyncApi.Info{Name: "foo"}}
+	_, ok := runtime.IsAsyncApiConfig(&dynamic.Config{Data: v2})
+	require.True(t, ok)
+
+}
+
 func getKafkaInfo(config *asyncapi3.Config) *runtime.KafkaInfo {
 	return &runtime.KafkaInfo{
 		Config: config,