[TT-12893]: Adding first implementation of streams API

.gitignore

@@ -52,10 +52,12 @@ tyk.test
 tyk-gateway.pid
 
 tyk_linux_*
+.aider*
 /dist/
 
 .terraform**
+.terraform.lock.hcl
 .task/
 *.test
 
-main
+main

apidef/oas/linter_test.go

@@ -78,7 +78,7 @@ func TestXTykGateway_Lint(t *testing.T) {
 			settings.Server.EventHandlers[i].Webhook.CoolDownPeriod = ReadableDuration(time.Second * 20)
 		}
 
-		for idx, _ := range settings.Middleware.Operations {
+		for idx := range settings.Middleware.Operations {
 			settings.Middleware.Operations[idx].CircuitBreaker.Threshold = 0.5
 		}
 

benthos.yaml

@@ -0,0 +1,11 @@
+input:
+  stdin: {}
+
+output:
+  http_server:
+    address: "0.0.0.0:4196"
+    path: /get
+    stream_path: /get/stream
+    ws_path: /get/ws
+    allowed_verbs:
+      - GET

cli/linter/schema.json

@@ -163,9 +163,23 @@
           }
         }
       }
+    },
+    "Streaming": {
+      "type": "object",
+      "properties": {
+        "enabled": {
+          "type": "boolean"
+        },
+        "allow_unsafe": {
+          "type": "array"
+        }
+      }
     }
   },
   "properties": {
+    "streaming": {
+      "$ref": "#/definitions/Streaming"
+    },
     "allow_insecure_configs": {
       "type": "boolean"
     },
@@ -1194,6 +1208,11 @@
     "oas_config": {
       "validate_examples": false,
       "validate_schema_defaults": false
+    },
+    "labs": {
+      "type": ["object", "null"],
+      "additionalProperties": true,
+      "properties": {}
     }
   }
 }

config/config.go

@@ -52,6 +52,10 @@ var (
 		LivenessCheck: LivenessCheckConfig{
 			CheckDuration: time.Second * 10,
 		},
+		Streaming: StreamingConfig{
+			Enabled:     false,
+			AllowUnsafe: []string{},
+		},
 	}
 )
 
@@ -653,6 +657,12 @@ func (pwl *PortsWhiteList) Decode(value string) error {
 	return nil
 }
 
+// StreamingConfig is for configuring tyk streaming
+type StreamingConfig struct {
+	Enabled     bool     `json:"enabled"`
+	AllowUnsafe []string `json:"allow_unsafe"`
+}
+
 // Config is the configuration object used by Tyk to set up various parameters.
 type Config struct {
 	// Force your Gateway to work only on a specific domain name. Can be overridden by API custom domain.
@@ -1117,6 +1127,24 @@ type Config struct {
 
 	// OAS holds the configuration for various OpenAPI-specific functionalities
 	OAS OASConfig `json:"oas_config"`
+
+	Streaming StreamingConfig `json:"streaming"`
+
+	Labs LabsConfig `json:"labs"`
+}
+
+// LabsConfig include config for streaming
+type LabsConfig map[string]interface{}
+
+// Decode unmarshals json config into the Labs config
+func (lc *LabsConfig) Decode(value string) error {
+	var temp map[string]interface{}
+	if err := json.Unmarshal([]byte(value), &temp); err != nil {
+		log.Error("Error unmarshalling LabsConfig: ", err)
+		return err
+	}
+	*lc = temp
+	return nil
 }
 
 // OASConfig holds the configuration for various OpenAPI-specific functionalities

gateway/api_definition.go

@@ -217,6 +217,7 @@ type APISpec struct {
 	AnalyticsPluginConfig    *GoAnalyticsPlugin
 
 	middlewareChain *ChainObject
+	unloadHooks     []func()
 
 	network analytics.NetworkStats
 
@@ -237,8 +238,16 @@ func (a *APISpec) GetSessionLifetimeRespectsKeyExpiration() bool {
 	return a.SessionLifetimeRespectsKeyExpiration
 }
 
+// AddUnloadHook adds a function to be called when the API spec is unloaded
+func (s *APISpec) AddUnloadHook(hook func()) {
+	s.unloadHooks = append(s.unloadHooks, hook)
+}
+
 // Release releases all resources associated with API spec
-func (s *APISpec) Release() {
+func (s *APISpec) Unload() {
+	s.Lock()
+	defer s.Unlock()
+
 	// release circuit breaker resources
 	for _, path := range s.RxPaths {
 		for _, urlSpec := range path {
@@ -267,6 +276,11 @@ func (s *APISpec) Release() {
 		s.HTTPTransport.transport.CloseIdleConnections()
 		s.HTTPTransport = nil
 	}
+
+	for _, hook := range s.unloadHooks {
+		hook()
+	}
+	s.unloadHooks = nil
 }
 
 // Validate returns nil if s is a valid spec and an error stating why the spec is not valid.

gateway/api_loader.go

@@ -426,6 +426,8 @@ func (gw *Gateway) processSpec(spec *APISpec, apisByListen map[string]int,
 
 	gw.mwAppendEnabled(&chainArray, &RateLimitForAPI{BaseMiddleware: baseMid})
 	gw.mwAppendEnabled(&chainArray, &GraphQLMiddleware{BaseMiddleware: baseMid})
+	gw.mwAppendEnabled(&chainArray, &StreamingMiddleware{BaseMiddleware: baseMid})
+
 	if !spec.UseKeylessAccess {
 		gw.mwAppendEnabled(&chainArray, &GraphQLComplexityMiddleware{BaseMiddleware: baseMid})
 		gw.mwAppendEnabled(&chainArray, &GraphQLGranularAccessMiddleware{BaseMiddleware: baseMid})
@@ -946,7 +948,6 @@ func (gw *Gateway) loadApps(specs []*APISpec) {
 	gw.loadControlAPIEndpoints(router)
 
 	muxer.setRouter(port, "", router, gw.GetConfig())
-
 	gs := gw.prepareStorage()
 	shouldTrace := trace.IsEnabled()
 
@@ -996,13 +997,15 @@ func (gw *Gateway) loadApps(specs []*APISpec) {
 
 	gw.DefaultProxyMux.swap(muxer, gw)
 
-	var specsToRelease []*APISpec
+	var specsToUnload []*APISpec
+
 	gw.apisMu.Lock()
 
 	for _, spec := range specs {
 		curSpec, ok := gw.apisByID[spec.APIID]
 		if ok && curSpec != nil && shouldReloadSpec(curSpec, spec) {
-			specsToRelease = append(specsToRelease, curSpec)
+			mainLog.Debugf("Spec %s has changed and needs to be reloaded", curSpec.APIID)
+			specsToUnload = append(specsToUnload, curSpec)
 		}
 
 		// Bind versions to base APIs again
@@ -1018,8 +1021,9 @@ func (gw *Gateway) loadApps(specs []*APISpec) {
 
 	gw.apisMu.Unlock()
 
-	for _, spec := range specsToRelease {
-		spec.Release()
+	for _, spec := range specsToUnload {
+		mainLog.Debugf("Unloading spec %s", spec.APIID)
+		spec.Unload()
 	}
 
 	mainLog.Debug("Checker host list")

gateway/middleware.go

@@ -55,6 +55,10 @@ type TykMiddleware interface {
 	ProcessRequest(w http.ResponseWriter, r *http.Request, conf interface{}) (error, int) // Handles request
 	EnabledForSpec() bool
 	Name() string
+
+	GetSpec() *APISpec
+
+	Unload()
 }
 
 type TraceMiddleware struct {
@@ -121,6 +125,9 @@ func (gw *Gateway) createMiddleware(actualMW TykMiddleware) func(http.Handler) h
 	mw.SetName(mw.Name())
 	mw.Logger().Debug("Init")
 
+	spec := mw.GetSpec()
+	spec.AddUnloadHook(actualMW.Unload)
+
 	// Pull the configuration
 	mwConf, err := mw.Config()
 	if err != nil {
@@ -155,14 +162,15 @@ func (gw *Gateway) createMiddleware(actualMW TykMiddleware) func(http.Handler) h
 			}
 
 			startTime := time.Now()
-			mw.Logger().WithField("ts", startTime.UnixNano()).Debug("Started")
+			mw.Logger().WithField("ts", startTime.UnixNano()).WithField("mw", mw.Name()).Debug("Started")
 
 			if mw.Base().Spec.CORS.OptionsPassthrough && r.Method == "OPTIONS" {
 				h.ServeHTTP(w, r)
 				return
 			}
 
 			err, errCode := mw.ProcessRequest(w, r, mwConf)
+
 			if err != nil {
 				writeResponse := true
 				// Prevent double error write
@@ -268,6 +276,16 @@ func (t *BaseMiddleware) Config() (interface{}, error) {
 	return nil, nil
 }
 
+// Unload unloads the middleware and frees resources
+func (t *BaseMiddleware) Unload() {
+	// methos created to satisfy middleware contract
+}
+
+// GetSpec returns the spec of the middleware
+func (t *BaseMiddleware) GetSpec() *APISpec {
+	return t.Spec
+}
+
 func (t *BaseMiddleware) OrgSession(orgID string) (user.SessionState, bool) {
 
 	if rpc.IsEmergencyMode() {

gateway/mw_context_vars.go

@@ -46,6 +46,13 @@ func (m *MiddlewareContextVars) ProcessRequest(w http.ResponseWriter, r *http.Re
 		contextDataObject[name] = c.Value
 	}
 
+	for key, vals := range r.Form {
+		name := "request_data_" + strings.Replace(key, "-", "_", -1)
+		if len(vals) > 0 {
+			contextDataObject[name] = vals[0]
+		}
+	}
+
 	ctxSetData(r, contextDataObject)
 
 	return nil, http.StatusOK

gateway/mw_context_vars_test.go

@@ -9,6 +9,8 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/google/go-cmp/cmp"
+
 	"github.com/TykTechnologies/tyk/apidef"
 	"github.com/TykTechnologies/tyk/test"
 )
@@ -88,6 +90,8 @@ func testPrepareTestContextVarsMiddleware() map[string]testContextVarsData {
 					"x": {"123"},
 					"y": {"test"},
 				},
+				"request_data_x": "123",
+				"request_data_y": "test",
 				"headers": map[string][]string{
 					"x-header-a": {"A"},
 					"x-header-b": {"B"},
@@ -120,6 +124,11 @@ func testPrepareTestContextVarsMiddleware() map[string]testContextVarsData {
 					"j":   {"2"},
 					"str": {"abc"},
 				},
+				"request_data_x":   "123",
+				"request_data_y":   "test",
+				"request_data_i":   "1",
+				"request_data_j":   "2",
+				"request_data_str": "abc",
 				"headers": map[string][]string{
 					"x-header-a":   {"A"},
 					"x-header-b":   {"B"},
@@ -155,6 +164,11 @@ func testPrepareTestContextVarsMiddleware() map[string]testContextVarsData {
 					"j":   {"2"},
 					"str": {"abc"},
 				},
+				"request_data_x":   "123",
+				"request_data_y":   "test",
+				"request_data_i":   "1",
+				"request_data_j":   "2",
+				"request_data_str": "abc",
 				"headers": map[string][]string{
 					"x-header-a":   {"A"},
 					"x-header-b":   {"B"},
@@ -209,7 +223,8 @@ func TestContextVarsMiddlewareProcessRequest(t *testing.T) {
 			delete(ctxDataObject, "request_id")
 
 			if !reflect.DeepEqual(ctxDataObject, test.ExpectedCtxDataObject) {
-				t.Errorf("Expected: %v\n Got: %v\n", test.ExpectedCtxDataObject, ctxDataObject)
+				//t.Errorf("Expected: %v\n Got: %v\n", test.ExpectedCtxDataObject, ctxDataObject)
+				t.Errorf("diff: %s", cmp.Diff(ctxDataObject, test.ExpectedCtxDataObject))
 			}
 		})
 	}