[TT-13201] Improve code organization

apidef/oas/oas.go

@@ -17,6 +17,9 @@ const (
 	// ExtensionTykAPIGateway is the OAS schema key for the Tyk extension.
 	ExtensionTykAPIGateway = "x-tyk-api-gateway"
 
+	// ExtensionTykStreaming is the OAS schema key for the Tyk Streams extension.
+	ExtensionTykStreaming = "x-tyk-streaming"
+
 	// Main holds the default version value (empty).
 	Main = ""
 
@@ -107,6 +110,51 @@ func (s *OAS) ExtractTo(api *apidef.APIDefinition) {
 	api.VersionData.Versions[Main] = vInfo
 }
 
+func (s *OAS) SetTykStreamingExtension(xTykStreaming *XTykStreaming) {
+	if s.Extensions == nil {
+		s.Extensions = make(map[string]interface{})
+	}
+
+	s.Extensions[ExtensionTykStreaming] = xTykStreaming
+}
+
+func (s *OAS) GetTykStreamingExtension() *XTykStreaming {
+	if s.Extensions == nil {
+		return nil
+	}
+
+	if ext := s.Extensions[ExtensionTykStreaming]; ext != nil {
+		rawTykStreaming, ok := ext.(json.RawMessage)
+		if ok {
+			var xTykStreaming XTykStreaming
+			_ = json.Unmarshal(rawTykStreaming, &xTykStreaming)
+			s.Extensions[ExtensionTykStreaming] = &xTykStreaming
+			return &xTykStreaming
+		}
+
+		mapTykAPIGateway, ok := ext.(map[string]interface{})
+		if ok {
+			var xTykStreaming XTykStreaming
+			dbByte, _ := json.Marshal(mapTykAPIGateway)
+			_ = json.Unmarshal(dbByte, &xTykStreaming)
+			s.Extensions[ExtensionTykStreaming] = &xTykStreaming
+			return &xTykStreaming
+		}
+
+		return ext.(*XTykStreaming)
+	}
+
+	return nil
+}
+
+func (s *OAS) RemoveTykStreamingExtension() {
+	if s.Extensions == nil {
+		return
+	}
+
+	delete(s.Extensions, ExtensionTykStreaming)
+}
+
 // SetTykExtension populates our OAS schema extension inside *OAS.
 func (s *OAS) SetTykExtension(xTykAPIGateway *XTykAPIGateway) {
 	if s.Extensions == nil {

apidef/oas/root.go

@@ -8,6 +8,18 @@ import (
 	"github.com/TykTechnologies/tyk/apidef"
 )
 
+// XTykStreaming represents the structure for Tyk streaming configurations.
+type XTykStreaming struct {
+	// Info contains the main metadata for the API definition.
+	Info Info `bson:"info" json:"info"` // required
+	// Server contains the configurations related to the server.
+	Server Server `bson:"server" json:"server"` // required
+	// Streams contains the configurations related to Tyk Streams
+	Streams map[string]interface{} `bson:"streams" json:"streams"` // required
+	// Middleware contains the configurations related to the Tyk middleware.
+	Middleware *Middleware `bson:"middleware,omitempty" json:"middleware,omitempty"`
+}
+
 // XTykAPIGateway contains custom Tyk API extensions for the OpenAPI definition.
 // The values for the extensions are stored inside the OpenAPI document, under
 // the key `x-tyk-api-gateway`.

apidef/streams/validator.go

@@ -4,6 +4,7 @@ import (
 	"embed"
 	"errors"
 	"fmt"
+	"github.com/TykTechnologies/tyk/apidef/oas"
 	"path/filepath"
 	"sort"
 	"strings"
@@ -22,11 +23,6 @@ import (
 //go:embed schema/*
 var schemaDir embed.FS
 
-const (
-	// ExtensionTykStreaming is the OAS schema key for the Tyk Streams extension.
-	ExtensionTykStreaming = "x-tyk-streaming"
-)
-
 const (
 	keyStreams                  = "streams"
 	keyDefinitions              = "definitions"
@@ -47,9 +43,9 @@ var (
 
 func loadSchemas() error {
 	loadOAS := func() error {
-		xTykStreamingSchema, err := schemaDir.ReadFile(fmt.Sprintf("schema/%s.json", ExtensionTykStreaming))
+		xTykStreamingSchema, err := schemaDir.ReadFile(fmt.Sprintf("schema/%s.json", oas.ExtensionTykStreaming))
 		if err != nil {
-			return fmt.Errorf("%s loading failed: %w", ExtensionTykStreaming, err)
+			return fmt.Errorf("%s loading failed: %w", oas.ExtensionTykStreaming, err)
 		}
 
 		xTykStreamingSchemaWithoutDefs := jsonparser.Delete(xTykStreamingSchema, keyDefinitions)
@@ -65,7 +61,7 @@ func loadSchemas() error {
 				continue
 			}
 
-			if strings.HasSuffix(fileName, fmt.Sprintf("%s.json", ExtensionTykStreaming)) {
+			if strings.HasSuffix(fileName, fmt.Sprintf("%s.json", oas.ExtensionTykStreaming)) {
 				continue
 			}
 
@@ -75,7 +71,7 @@ func loadSchemas() error {
 				return err
 			}
 
-			data, err = jsonparser.Set(data, xTykStreamingSchemaWithoutDefs, keyProperties, ExtensionTykStreaming)
+			data, err = jsonparser.Set(data, xTykStreamingSchemaWithoutDefs, keyProperties, oas.ExtensionTykStreaming)
 			if err != nil {
 				return err
 			}
@@ -119,7 +115,7 @@ func loadSchemas() error {
 }
 
 func validateBentoConfiguration(document []byte, bentoValidatorKind bento.ValidatorKind) error {
-	streams, _, _, err := jsonparser.Get(document, ExtensionTykStreaming, keyStreams)
+	streams, _, _, err := jsonparser.Get(document, oas.ExtensionTykStreaming, keyStreams)
 	if errors.Is(err, jsonparser.KeyPathNotFoundError) {
 		// no streams found
 		return nil
@@ -191,7 +187,7 @@ func ValidateOASTemplateWithBentoValidator(documentBody []byte, oasVersion strin
 		return err
 	}
 
-	oasSchema = jsonparser.Delete(oasSchema, keyProperties, ExtensionTykStreaming, keyRequired)
+	oasSchema = jsonparser.Delete(oasSchema, keyProperties, oas.ExtensionTykStreaming, keyRequired)
 
 	definitions, _, _, err := jsonparser.Get(oasSchema, keyDefinitions)
 	if err != nil {

apidef/streams/validator_test.go

@@ -30,7 +30,7 @@ func TestValidateTykStreamsOASObject(t *testing.T) {
 		},
 	}
 
-	validXTykAPIStreaming := XTykStreaming{
+	validXTykAPIStreaming := oas.XTykStreaming{
 		Info: oas.Info{
 			Name: "test-streams",
 			State: oas.State{
@@ -45,7 +45,7 @@ func TestValidateTykStreamsOASObject(t *testing.T) {
 		Streams: map[string]interface{}{},
 	}
 
-	validOASObject.Extensions[ExtensionTykStreaming] = &validXTykAPIStreaming
+	validOASObject.Extensions[oas.ExtensionTykStreaming] = &validXTykAPIStreaming
 
 	validOAS3Definition, _ := validOASObject.MarshalJSON()
 
@@ -59,7 +59,7 @@ func TestValidateTykStreamsOASObject(t *testing.T) {
 	invalidXTykAPIGateway := validXTykAPIStreaming
 	invalidXTykAPIGateway.Info = oas.Info{}
 	invalidXTykAPIGateway.Server.GatewayTags = &oas.GatewayTags{Enabled: true, Tags: []string{}}
-	invalidOASObject.Extensions[ExtensionTykStreaming] = &invalidXTykAPIGateway
+	invalidOASObject.Extensions[oas.ExtensionTykStreaming] = &invalidXTykAPIGateway
 	invalidOAS3Definition, _ := invalidOASObject.MarshalJSON()
 
 	t.Run("invalid OAS object", func(t *testing.T) {
@@ -136,7 +136,7 @@ func Test_loadOASSchema(t *testing.T) {
 		assert.NotNil(t, oasJSONSchemas)
 		for oasVersion := range oasJSONSchemas {
 			var xTykStreaming, xTykStreams []byte
-			xTykStreaming, _, _, err = jsonparser.Get(oasJSONSchemas[oasVersion], keyProperties, ExtensionTykStreaming)
+			xTykStreaming, _, _, err = jsonparser.Get(oasJSONSchemas[oasVersion], keyProperties, oas.ExtensionTykStreaming)
 			assert.NoError(t, err)
 			assert.NotNil(t, xTykStreaming)