refactor

gateway/api_definition.go

@@ -428,11 +428,15 @@ func (a APIDefinitionLoader) FromDashboardService(endpoint, secret string) ([]*A
 }
 
 // FromCloud will connect and download ApiDefintions from a Mongo DB instance.
-func (a APIDefinitionLoader) FromRPC(orgId string) ([]*APISpec, error) {
+func (a APIDefinitionLoader) FromRPC(orgId string, gw Gateway) ([]*APISpec, error) {
 	if rpc.IsEmergencyMode() {
 		return LoadDefinitionsFromRPCBackup()
 	}
-	store := RPCStorageHandler{}
+
+	store := RPCStorageHandler{
+		DoReload:gw.DoReload,
+	}
+
 	if !store.Connect() {
 		return nil, errors.New("Can't connect RPC layer")
 	}

gateway/api_loader.go

@@ -603,14 +603,14 @@ func sanitizeProxyPaths(apiSpec *APISpec, request *http.Request) {
 	request.URL.RawPath = apiSpec.StripListenPath(request, request.URL.RawPath)
 }
 
-func loadGlobalApps() {
+func(gw Gateway) loadGlobalApps() {
 	// we need to make a full copy of the slice, as loadApps will
 	// use in-place to sort the apis.
 	apisMu.RLock()
 	specs := make([]*APISpec, len(apiSpecs))
 	copy(specs, apiSpecs)
 	apisMu.RUnlock()
-	loadApps(specs)
+	gw.loadApps(specs)
 }
 
 func trimCategories(name string) string {
@@ -736,7 +736,7 @@ func loadGraphQLPlayground(spec *APISpec, router *mux.Router) {
 }
 
 // Create the individual API (app) specs based on live configurations and assign middleware
-func loadApps(specs []*APISpec) {
+func(gw Gateway) loadApps(specs []*APISpec) {
 	mainLog.Info("Loading API configurations.")
 
 	tmpSpecRegister := make(map[string]*APISpec)
@@ -761,7 +761,7 @@ func loadApps(specs []*APISpec) {
 	muxer := &proxyMux{}
 	router := mux.NewRouter()
 	router.NotFoundHandler = http.HandlerFunc(muxer.handle404)
-	loadControlAPIEndpoints(router)
+	gw.loadControlAPIEndpoints(router)
 
 	muxer.setRouter(port, "", router)
 
@@ -807,7 +807,7 @@ func loadApps(specs []*APISpec) {
 		}()
 	}
 
-	defaultProxyMux.swap(muxer)
+	gw.DefaultProxyMux.swap(muxer)
 
 	// Swap in the new register
 	apisMu.Lock()

gateway/proxy_muxer.go

@@ -86,9 +86,10 @@ type proxyMux struct {
 	again   again.Again
 }
 
+/*
 var defaultProxyMux = &proxyMux{
 	again: again.New(),
-}
+}*/
 
 func (m *proxyMux) getProxy(listenPort int) *proxy {
 	if listenPort == 0 {

gateway/rpc_storage_handler.go

@@ -93,6 +93,7 @@ type RPCStorageHandler struct {
 	KeyPrefix        string
 	HashKeys         bool
 	SuppressRegister bool
+	DoReload func()
 }
 
 var RPCGlobalCache = cache.New(30*time.Second, 15*time.Second)
@@ -125,7 +126,7 @@ func (r *RPCStorageHandler) Connect() bool {
 		func() {
 			reloadURLStructure(nil)
 		},
-		DoReload,
+		r.DoReload,
 	)
 }
 

gateway/server.go

@@ -270,7 +270,7 @@ func buildConnStr(resource string) string {
 	return config.Global().DBAppConfOptions.ConnectionString + resource
 }
 
-func syncAPISpecs() (int, error) {
+func(gw Gateway) syncAPISpecs() (int, error) {
 	loader := APIDefinitionLoader{}
 	apisMu.Lock()
 	defer apisMu.Unlock()
@@ -290,7 +290,7 @@ func syncAPISpecs() (int, error) {
 		mainLog.Debug("Using RPC Configuration")
 
 		var err error
-		s, err = loader.FromRPC(config.Global().SlaveOptions.RPCKey)
+		s, err = loader.FromRPC(config.Global().SlaveOptions.RPCKey, gw)
 		if err != nil {
 			return 0, err
 		}
@@ -396,15 +396,15 @@ func controlAPICheckClientCertificate(certLevel string, next http.Handler) http.
 }
 
 // loadControlAPIEndpoints loads the endpoints used for controlling the Gateway.
-func loadControlAPIEndpoints(muxer *mux.Router) {
+func(gw Gateway) loadControlAPIEndpoints(muxer *mux.Router) {
 	hostname := config.Global().HostName
 	if config.Global().ControlAPIHostname != "" {
 		hostname = config.Global().ControlAPIHostname
 	}
 
 	if muxer == nil {
 		cp := config.Global().ControlAPIPort
-		muxer = defaultProxyMux.router(cp, "")
+		muxer = gw.DefaultProxyMux.router(cp, "")
 		if muxer == nil {
 			if cp != 0 {
 				log.Error("Can't find control API router")
@@ -700,7 +700,7 @@ func rpcReloadLoop(rpcKey string) {
 
 var reloadMu sync.Mutex
 
-func DoReload() {
+func(gw Gateway) DoReload() {
 	reloadMu.Lock()
 	defer reloadMu.Unlock()
 
@@ -716,7 +716,7 @@ func DoReload() {
 	}
 
 	// load the specs
-	if count, err := syncAPISpecs(); err != nil {
+	if count, err := gw.syncAPISpecs(); err != nil {
 		mainLog.Error("Error during syncing apis:", err.Error())
 		return
 	} else {
@@ -727,7 +727,7 @@ func DoReload() {
 			return
 		}
 	}
-	loadGlobalApps()
+	gw.loadGlobalApps()
 
 	mainLog.Info("API reload complete")
 }
@@ -745,7 +745,7 @@ func shouldReload() ([]func(), bool) {
 	return n, true
 }
 
-func reloadLoop(ctx context.Context, tick <-chan time.Time, complete ...func()) {
+func(gw Gateway) reloadLoop(ctx context.Context, tick <-chan time.Time, complete ...func()) {
 	for {
 		select {
 		case <-ctx.Done():
@@ -760,7 +760,7 @@ func reloadLoop(ctx context.Context, tick <-chan time.Time, complete ...func())
 			}
 			start := time.Now()
 			mainLog.Info("reload: initiating")
-			DoReload()
+			gw.DoReload()
 			mainLog.Info("reload: complete")
 			mainLog.Info("Initiating coprocess reload")
 			DoCoprocessReload()
@@ -1196,6 +1196,14 @@ func getGlobalStorageHandler(keyPrefix string, hashKeys bool) storage.Handler {
 	return &storage.RedisCluster{KeyPrefix: keyPrefix, HashKeys: hashKeys}
 }
 
+func NewGateway()Gateway{
+	return Gateway{
+		DefaultProxyMux:&proxyMux{
+			again: again.New(),
+		},
+	}
+}
+
 func Start() {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -1206,6 +1214,8 @@ func Start() {
 		os.Exit(0)
 	}
 
+	gw := NewGateway()
+
 	SetNodeID("solo-" + uuid.NewV4().String())
 
 	if err := initialiseSystem(ctx); err != nil {
@@ -1237,7 +1247,7 @@ func Start() {
 			os.Setenv("TYK_SERVICE_NODEID", GetNodeID())
 		}
 	}
-	err := again.ListenFrom(&defaultProxyMux.again, onFork)
+	err := again.ListenFrom(&gw.DefaultProxyMux.again, onFork)
 	if err != nil {
 		mainLog.Errorf("Initializing again %s", err)
 	}
@@ -1246,7 +1256,7 @@ func Start() {
 		trace.SetLogger(mainLog)
 		defer trace.Close()
 	}
-	start(ctx)
+	gw.start(ctx)
 	go storage.ConnectToRedis(ctx, func() {
 		reloadURLStructure(func() {})
 	})
@@ -1279,17 +1289,17 @@ func Start() {
 
 	// TODO: replace goagain with something that support multiple listeners
 	// Example: https://gravitational.com/blog/golang-ssh-bastion-graceful-restarts/
-	startServer()
+	gw.startServer()
 
 	if again.Child() {
 		// This is a child process, we need to murder the parent now
 		if err := again.Kill(); err != nil {
 			mainLog.Fatal(err)
 		}
 	}
-	again.Wait(&defaultProxyMux.again)
+	again.Wait(&gw.DefaultProxyMux.again)
 	mainLog.Info("Stop signal received.")
-	if err := defaultProxyMux.again.Close(); err != nil {
+	if err := gw.DefaultProxyMux.again.Close(); err != nil {
 		mainLog.Error("Closing listeners: ", err)
 	}
 	// stop analytics workers
@@ -1335,7 +1345,7 @@ func writeProfiles() {
 	}
 }
 
-func start(ctx context.Context) {
+func (gw Gateway) start(ctx context.Context) {
 	// Set up a default org manager so we can traverse non-live paths
 	if !config.Global().SupressDefaultOrgStore {
 		mainLog.Debug("Initialising default org store")
@@ -1364,7 +1374,7 @@ func start(ctx context.Context) {
 
 	// 1s is the minimum amount of time between hot reloads. The
 	// interval counts from the start of one reload to the next.
-	go reloadLoop(ctx, time.Tick(time.Second))
+	go gw.reloadLoop(ctx, time.Tick(time.Second))
 	go reloadQueueLoop(ctx)
 }
 
@@ -1428,19 +1438,24 @@ func setupPortsWhitelist() {
 	config.SetGlobal(globalConf)
 }
 
-func startServer() {
+type Gateway struct {
+	DefaultProxyMux *proxyMux
+}
+
+func (gw Gateway) startServer() {
 	// Ensure that Control listener and default http listener running on first start
 	muxer := &proxyMux{}
 
 	router := mux.NewRouter()
-	loadControlAPIEndpoints(router)
+	gw.loadControlAPIEndpoints(router)
 	muxer.setRouter(config.Global().ControlAPIPort, "", router)
 
 	if muxer.router(config.Global().ListenPort, "") == nil {
 		muxer.setRouter(config.Global().ListenPort, "", mux.NewRouter())
 	}
 
-	defaultProxyMux.swap(muxer)
+//	defaultProxyMux.swap(muxer)
+	gw.DefaultProxyMux.swap(muxer)
 
 	// handle dashboard registration and nonces if available
 	handleDashboardRegistration()
@@ -1457,6 +1472,6 @@ func startServer() {
 	mainLog.Info("--> Listening on port: ", config.Global().ListenPort)
 	mainLog.Info("--> PID: ", hostDetails.PID)
 	if !rpc.IsEmergencyMode() {
-		DoReload()
+		gw.DoReload()
 	}
 }

gateway/testutil.go

@@ -78,6 +78,8 @@ type ReloadMachinery struct {
 	stop       chan struct{}
 }
 
+var globalGateway Gateway
+
 func NewReloadMachinery() *ReloadMachinery {
 	return &ReloadMachinery{
 		reloadTick: make(chan time.Time),
@@ -218,6 +220,7 @@ func (r *ReloadMachinery) TickOk(t *testing.T) {
 
 func InitTestMain(ctx context.Context, m *testing.M, genConf ...func(globalConf *config.Config)) int {
 	setTestMode(true)
+	globalGateway = NewGateway()
 	testServerRouter = testHttpHandler()
 	testServer := &http.Server{
 		Addr:           testHttpListen,
@@ -288,7 +291,7 @@ func InitTestMain(ctx context.Context, m *testing.M, genConf ...func(globalConf
 	cli.Init(VERSION, confPaths)
 	initialiseSystem(ctx)
 	// Small part of start()
-	loadControlAPIEndpoints(mainRouter())
+	globalGateway.loadControlAPIEndpoints(mainRouter())
 	if analytics.GeoIPDB == nil {
 		panic("GeoIPDB was not initialized")
 	}
@@ -305,7 +308,7 @@ func InitTestMain(ctx context.Context, m *testing.M, genConf ...func(globalConf
 		time.Sleep(10 * time.Millisecond)
 	}
 	go startPubSubLoop()
-	go reloadLoop(ctx, ReloadTestCase.ReloadTicker(), ReloadTestCase.OnReload)
+	go globalGateway.reloadLoop(ctx, ReloadTestCase.ReloadTicker(), ReloadTestCase.OnReload)
 	go reloadQueueLoop(ctx, ReloadTestCase.OnQueued)
 	go reloadSimulation()
 	exitCode := m.Run()
@@ -395,15 +398,15 @@ func bundleHandleFunc(w http.ResponseWriter, r *http.Request) {
 	z.Close()
 }
 func mainRouter() *mux.Router {
-	return getMainRouter(defaultProxyMux)
+	return getMainRouter(globalGateway.DefaultProxyMux)
 }
 
 func mainProxy() *proxy {
-	return defaultProxyMux.getProxy(config.Global().ListenPort)
+	return globalGateway.DefaultProxyMux.getProxy(config.Global().ListenPort)
 }
 
 func controlProxy() *proxy {
-	p := defaultProxyMux.getProxy(config.Global().ControlAPIPort)
+	p := globalGateway.DefaultProxyMux.getProxy(config.Global().ControlAPIPort)
 	if p != nil {
 		return p
 	}
@@ -818,7 +821,7 @@ func CreateDefinitionFromString(defStr string) *APISpec {
 
 func LoadSampleAPI(def string) (spec *APISpec) {
 	spec = CreateDefinitionFromString(def)
-	loadApps([]*APISpec{spec})
+	globalGateway.loadApps([]*APISpec{spec})
 	return
 }
 
@@ -859,7 +862,7 @@ func (s *Test) Start(slavedClusterConfig *SlaveDataCenter) {
 	l.Close()
 	globalConf := config.Global()
 	globalConf.ListenPort, _ = strconv.Atoi(port)
-
+	gw := NewGateway()
 	if s.config.sepatateControlAPI {
 		l, _ := net.Listen("tcp", "127.0.0.1:0")
 
@@ -884,7 +887,7 @@ func (s *Test) Start(slavedClusterConfig *SlaveDataCenter) {
 
 	setupPortsWhitelist()
 
-	startServer()
+	gw.startServer()
 	ctx, cancel := context.WithCancel(context.Background())
 	s.cacnel = cancel
 	setupGlobals(ctx)
@@ -937,7 +940,7 @@ func (s *Test) Close() {
 	if s.cacnel != nil {
 		s.cacnel()
 	}
-	defaultProxyMux.swap(&proxyMux{})
+	globalGateway.DefaultProxyMux.swap(&proxyMux{})
 	if s.config.sepatateControlAPI {
 		globalConf := config.Global()
 		globalConf.ControlAPIPort = 0
@@ -1230,7 +1233,7 @@ func LoadAPI(specs ...*APISpec) (out []*APISpec) {
 		}
 	}
 
-	DoReload()
+	globalGateway.DoReload()
 
 	for _, spec := range specs {
 		out = append(out, getApiSpec(spec.APIID))