Decouple ee build tags for streaming middleware

apidef/api_definitions.go

@@ -12,21 +12,17 @@ import (
 
 	"github.com/clbanning/mxj"
 	"github.com/lonelycode/osin"
-
-	"github.com/TykTechnologies/storage/persistent/model"
-
-	"github.com/TykTechnologies/tyk/internal/event"
-
-	"github.com/TykTechnologies/tyk/internal/reflect"
-
 	"golang.org/x/oauth2"
 
-	"github.com/TykTechnologies/graphql-go-tools/pkg/execution/datasource"
-
 	"github.com/TykTechnologies/gojsonschema"
+	"github.com/TykTechnologies/graphql-go-tools/pkg/execution/datasource"
+	"github.com/TykTechnologies/storage/persistent/model"
 
 	"github.com/TykTechnologies/tyk/regexp"
 
+	"github.com/TykTechnologies/tyk/internal/event"
+	"github.com/TykTechnologies/tyk/internal/httputil"
+	"github.com/TykTechnologies/tyk/internal/reflect"
 	"github.com/TykTechnologies/tyk/internal/uuid"
 )
 
@@ -1172,6 +1168,11 @@ type GraphQLPlayground struct {
 	Path string `bson:"path" json:"path"`
 }
 
+// StripListenPath will strip the listen path from the URL, keeping version in tact.
+func (a *APIDefinition) StripListenPath(reqPath string) string {
+	return httputil.StripListenPath(a.Proxy.ListenPath, reqPath)
+}
+
 // EncodeForDB will encode map[string]struct variables for saving in URL format
 func (a *APIDefinition) EncodeForDB() {
 	newVersion := make(map[string]VersionInfo)

ctx/ctx.go

@@ -5,16 +5,14 @@ import (
 	"encoding/json"
 	"net/http"
 
-	"github.com/TykTechnologies/tyk/internal/httputil"
-
+	"github.com/TykTechnologies/tyk/apidef"
 	"github.com/TykTechnologies/tyk/apidef/oas"
-
 	"github.com/TykTechnologies/tyk/config"
-
-	"github.com/TykTechnologies/tyk/apidef"
 	logger "github.com/TykTechnologies/tyk/log"
 	"github.com/TykTechnologies/tyk/storage"
 	"github.com/TykTechnologies/tyk/user"
+
+	"github.com/TykTechnologies/tyk/internal/repository"
 )
 
 type Key uint
@@ -78,7 +76,7 @@ func ctxSetSession(r *http.Request, s *user.SessionState, scheduleUpdate bool, h
 		s.Touch()
 	}
 
-	httputil.SetContext(r, ctx)
+	repository.SetContext(r, ctx)
 }
 
 func GetAuthToken(r *http.Request) string {
@@ -116,7 +114,7 @@ func SetSession(r *http.Request, s *user.SessionState, scheduleUpdate bool, hash
 func SetDefinition(r *http.Request, s *apidef.APIDefinition) {
 	ctx := r.Context()
 	ctx = context.WithValue(ctx, Definition, s)
-	httputil.SetContext(r, ctx)
+	repository.SetContext(r, ctx)
 }
 
 func GetDefinition(r *http.Request) *apidef.APIDefinition {

ee/middleware/streams/middleware.go

@@ -21,10 +21,9 @@ import (
 
 // Middleware implements a streaming middleware.
 type Middleware struct {
-	Spec *APISpec
-	Gw   Gateway
-
-	base BaseMiddleware
+	Spec     model.MergedAPI
+	Gw       Gateway
+	logEntry *logrus.Entry
 
 	createStreamManagerLock sync.Mutex
 	StreamManagerCache      sync.Map // Map of payload hash to Manager
@@ -35,26 +34,29 @@ type Middleware struct {
 	defaultManager *Manager
 }
 
+// Name holds the middleware name as a constant.
+const Name = "StreamingMiddleware"
+
 // Middleware implements model.Middleware.
 var _ model.Middleware = &Middleware{}
 
 // NewMiddleware returns a new instance of Middleware.
-func NewMiddleware(gw Gateway, mw BaseMiddleware, spec *APISpec) *Middleware {
+func NewMiddleware(gw Gateway, logger *logrus.Entry, spec model.MergedAPI) *Middleware {
 	return &Middleware{
-		base: mw,
-		Gw:   gw,
-		Spec: spec,
+		Gw:       gw,
+		Spec:     spec,
+		logEntry: logger.WithField("mw", Name),
 	}
 }
 
 // Logger returns a logger with middleware filled out.
 func (s *Middleware) Logger() *logrus.Entry {
-	return s.base.Logger().WithField("mw", s.Name())
+	return s.logEntry
 }
 
 // Name returns the name for the middleware.
 func (s *Middleware) Name() string {
-	return "StreamingMiddleware"
+	return Name
 }
 
 // EnabledForSpec checks if streaming is enabled on the config.

ee/middleware/streams/model.go

@@ -4,7 +4,6 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/TykTechnologies/tyk/apidef/oas"
 	"github.com/TykTechnologies/tyk/internal/model"
 )
 
@@ -25,28 +24,6 @@ type Gateway interface {
 	model.ReplaceTykVariables
 }
 
-// APISpec is a subset of gateway.APISpec for the values the middleware consumes.
-type APISpec struct {
-	APIID string
-	Name  string
-	IsOAS bool
-	OAS   oas.OAS
-
-	StripListenPath model.StripListenPathFunc
-}
-
-// NewAPISpec creates a new APISpec object based on the required inputs.
-// The resulting object is a subset of `*gateway.APISpec`.
-func NewAPISpec(id string, name string, isOasDef bool, oasDef oas.OAS, stripListenPath model.StripListenPathFunc) *APISpec {
-	return &APISpec{
-		APIID:           id,
-		Name:            name,
-		IsOAS:           isOasDef,
-		OAS:             oasDef,
-		StripListenPath: stripListenPath,
-	}
-}
-
 // StreamsConfig represents a stream configuration.
 type StreamsConfig struct {
 	Info struct {

ee/register/register.go

@@ -0,0 +1,2 @@
+// This package takes care of registering the EE componts to Gateway.
+package register

ee/register/register_ee.go

@@ -0,0 +1,16 @@
+//go:build ee || dev
+
+package register
+
+import (
+	"github.com/TykTechnologies/tyk/ee/middleware/streams"
+
+	"github.com/TykTechnologies/tyk/internal/middleware"
+	"github.com/TykTechnologies/tyk/internal/model"
+)
+
+func init() {
+	middleware.Add("ee:middleware:streaming", func(gw model.Gateway, base model.LoggerProvider, spec model.MergedAPI) model.Middleware {
+		return streams.NewMiddleware(gw, base.Logger(), spec)
+	})
+}

gateway/api_definition.go

@@ -1750,11 +1750,6 @@ func (a *APISpec) Version(r *http.Request) (*apidef.VersionInfo, RequestStatus)
 	return &version, StatusOk
 }
 
-// StripListenPath will strip the listen path from the URL, keeping version in tact.
-func (a *APISpec) StripListenPath(reqPath string) string {
-	return httputil.StripListenPath(a.Proxy.ListenPath, reqPath)
-}
-
 // StripVersionPath will strip the version from the URL. The input URL
 // should already have listen path stripped.
 func (a *APISpec) StripVersionPath(reqPath string) string {

gateway/api_loader.go

@@ -26,6 +26,7 @@ import (
 	"github.com/TykTechnologies/tyk/storage"
 	"github.com/TykTechnologies/tyk/trace"
 
+	"github.com/TykTechnologies/tyk/internal/middleware"
 	"github.com/TykTechnologies/tyk/internal/otel"
 )
 
@@ -427,8 +428,11 @@ func (gw *Gateway) processSpec(spec *APISpec, apisByListen map[string]int,
 	gw.mwAppendEnabled(&chainArray, &RateLimitForAPI{BaseMiddleware: baseMid})
 	gw.mwAppendEnabled(&chainArray, &GraphQLMiddleware{BaseMiddleware: baseMid})
 
-	if streamMw := getStreamingMiddleware(baseMid); streamMw != nil {
-		gw.mwAppendEnabled(&chainArray, streamMw)
+	if streamMw := middleware.Get("ee:middleware:streaming"); len(streamMw) > 0 {
+		for _, createMw := range streamMw {
+			mw := WrapMiddleware(baseMid, createMw(baseMid.Gw, baseMid, spec.MergedAPI()))
+			gw.mwAppendEnabled(&chainArray, mw)
+		}
 	}
 
 	if !spec.UseKeylessAccess {

gateway/model_apispec.go

@@ -3,6 +3,8 @@ package gateway
 import (
 	"net/http"
 	"strings"
+
+	"github.com/TykTechnologies/tyk/internal/model"
 )
 
 // CheckSpecMatchesStatus checks if a URL spec has a specific status.
@@ -73,3 +75,8 @@ func (a *APISpec) getMatchPathAndMethod(r *http.Request, mode URLStatus) (string
 
 	return matchPath, method
 }
+
+// MergedAPI provides a model.MergedAPI from an APISpec.
+func (a *APISpec) MergedAPI() model.MergedAPI {
+	return model.NewMergedAPI(a.APIDefinition, &a.OAS)
+}

gateway/mw_oauth2_auth.go

@@ -16,7 +16,7 @@ import (
 
 	"github.com/TykTechnologies/tyk/apidef"
 	"github.com/TykTechnologies/tyk/header"
-	"github.com/TykTechnologies/tyk/internal/httputil"
+	"github.com/TykTechnologies/tyk/internal/repository"
 	"github.com/TykTechnologies/tyk/storage"
 )
 
@@ -158,7 +158,8 @@ func (OAuthSpec *UpstreamOAuth) ProcessRequest(_ http.ResponseWriter, r *http.Re
 		}
 	}
 
-	httputil.SetUpstreamAuth(r, upstreamOAuthProvider)
+	repository.SetUpstreamAuth(r, upstreamOAuthProvider)
+
 	return nil, http.StatusOK
 }
 

gateway/mw_upstream_basic_auth.go

@@ -6,6 +6,7 @@ import (
 	"github.com/sirupsen/logrus"
 
 	"github.com/TykTechnologies/tyk/internal/httputil"
+	"github.com/TykTechnologies/tyk/internal/repository"
 
 	"github.com/TykTechnologies/tyk/header"
 )
@@ -48,7 +49,7 @@ func (t *UpstreamBasicAuth) ProcessRequest(_ http.ResponseWriter, r *http.Reques
 
 	upstreamBasicAuthProvider.AuthValue = httputil.AuthHeader(basicAuthConfig.Username, basicAuthConfig.Password)
 
-	httputil.SetUpstreamAuth(r, upstreamBasicAuthProvider)
+	repository.SetUpstreamAuth(r, upstreamBasicAuthProvider)
 	return nil, http.StatusOK
 }
 

gateway/reverse_proxy.go

@@ -43,6 +43,7 @@ import (
 	"github.com/TykTechnologies/tyk/internal/graphengine"
 	"github.com/TykTechnologies/tyk/internal/httputil"
 	"github.com/TykTechnologies/tyk/internal/otel"
+	"github.com/TykTechnologies/tyk/internal/repository"
 	"github.com/TykTechnologies/tyk/regexp"
 	"github.com/TykTechnologies/tyk/trace"
 	"github.com/TykTechnologies/tyk/user"
@@ -1853,7 +1854,7 @@ func (p *ReverseProxy) addAuthInfo(outReq, req *http.Request) {
 		return
 	}
 
-	if authProvider := httputil.GetUpstreamAuth(req); authProvider != nil {
+	if authProvider := repository.GetUpstreamAuth(req); authProvider != nil {
 		authProvider.Fill(outReq)
 	}
 }

internal/middleware/registry.go

@@ -0,0 +1,32 @@
+package middleware
+
+import (
+	"sync"
+
+	"github.com/TykTechnologies/tyk/internal/model"
+)
+
+type ProviderFn func(model.Gateway, model.LoggerProvider, model.MergedAPI) model.Middleware
+
+var globals = struct {
+	mu       *sync.RWMutex
+	registry map[string][]ProviderFn
+}{
+	mu:       &sync.RWMutex{},
+	registry: map[string][]ProviderFn{},
+}
+
+func Add(hook string, provider ProviderFn) {
+	globals.mu.Lock()
+	defer globals.mu.Unlock()
+
+	globals.registry[hook] = append(globals.registry[hook], provider)
+}
+
+func Get(hook string) []ProviderFn {
+	globals.mu.RLock()
+	defer globals.mu.RUnlock()
+
+	data, _ := globals.registry[hook]
+	return data
+}

internal/model/merged_apis.go

@@ -23,6 +23,11 @@ type MergedAPI struct {
 	OAS                   *oas.OAS `json:"oas"`
 }
 
+// NewMergedAPI exists due to embedding.
+func NewMergedAPI(classic *apidef.APIDefinition, oas *oas.OAS) MergedAPI {
+	return MergedAPI{classic, oas}
+}
+
 // Set sets the available classic API definitions to the MergedAPIList.
 func (f *MergedAPIList) SetClassic(defs []*apidef.APIDefinition) {
 	for _, def := range defs {

internal/httputil/context.go → internal/repository/context.go

@@ -1,4 +1,4 @@
-package httputil
+package repository
 
 import (
 	"context"