Adopting PathRouter to new dispatching #1 (#1072)

This is first part of integration PathRouter to new API

Author: pdelewski

quesma/frontend_connectors/basic_http_frontend_connector.go

@@ -89,6 +89,14 @@ func (router *HTTPRouter) Multiplexer() *http.ServeMux {
 	return router.mux
 }
 
+func (router *HTTPRouter) Register(pattern string, predicate quesma_api.RequestMatcher, handler quesma_api.Handler) {
+	panic("not implemented")
+}
+
+func (router *HTTPRouter) Matches(req *quesma_api.Request) (*quesma_api.HttpHandlersPipe, *quesma_api.Decision) {
+	panic("not implemented")
+}
+
 type BasicHTTPFrontendConnector struct {
 	listener *http.Server
 	router   quesma_api.Router
@@ -161,6 +169,7 @@ func (h *BasicHTTPFrontendConnector) Listen() error {
 	h.listener.Handler = h
 	go func() {
 		err := h.listener.ListenAndServe()
+		// TODO: Handle error
 		_ = err
 	}()
 

quesma/frontend_connectors/router_v2.go

@@ -225,19 +225,19 @@ func (r *RouterV2) Reroute(ctx context.Context, w http.ResponseWriter, req *http
 	quesmaRequest.ParsedBody = types.ParseRequestBody(quesmaRequest.Body)
 	var handler quesma_api.Handler
 	var decision *quesma_api.Decision
-	searchHandler, searchDecision := searchRouter.Matches(quesmaRequest)
+	searchHandlerPipe, searchDecision := searchRouter.Matches(quesmaRequest)
 	if searchDecision != nil {
 		decision = searchDecision
 	}
-	if searchHandler != nil {
-		handler = searchHandler
+	if searchHandlerPipe != nil {
+		handler = searchHandlerPipe.Handler
 	}
-	ingestHandler, ingestDecision := ingestRouter.Matches(quesmaRequest)
+	ingestHandlerPipe, ingestDecision := ingestRouter.Matches(quesmaRequest)
 	if searchDecision == nil {
 		decision = ingestDecision
 	}
-	if searchHandler == nil {
-		handler = ingestHandler
+	if searchHandlerPipe == nil && ingestHandlerPipe != nil {
+		handler = ingestHandlerPipe.Handler
 	}
 	if decision != nil {
 		w.Header().Set(QuesmaTableResolverHeader, decision.String())

quesma/quesma/dual_write_proxy.go

@@ -308,17 +308,17 @@ func (r *router) reroute(ctx context.Context, w http.ResponseWriter, req *http.R
 
 	quesmaRequest.ParsedBody = types.ParseRequestBody(quesmaRequest.Body)
 
-	handler, decision := router.Matches(quesmaRequest)
+	handlersPipe, decision := router.Matches(quesmaRequest)
 
 	if decision != nil {
 		w.Header().Set(frontend_connectors.QuesmaTableResolverHeader, decision.String())
 	} else {
 		w.Header().Set(frontend_connectors.QuesmaTableResolverHeader, "n/a")
 	}
 
-	if handler != nil {
+	if handlersPipe != nil {
 		quesmaResponse, err := recordRequestToClickhouse(req.URL.Path, r.quesmaManagementConsole, func() (*quesma_api.Result, error) {
-			return handler(ctx, quesmaRequest)
+			return handlersPipe.Handler(ctx, quesmaRequest)
 		})
 
 		zip := strings.Contains(req.Header.Get("Accept-Encoding"), "gzip")

quesma/quesma/dual_write_proxy_v2.go

@@ -100,8 +100,7 @@ func newDualWriteProxyV2(schemaLoader clickhouse.TableDiscovery, logManager *cli
 	})
 
 	elasticHttpFrontentConnector := NewElasticHttpFrontendConnector(":"+strconv.Itoa(int(config.PublicTcpPort)),
-		&routerInstance, searchRouter, ingestRouter, logManager, agent)
-
+		&routerInstance, searchRouter.(*quesma_api.PathRouter), ingestRouter.(*quesma_api.PathRouter), logManager, agent)
 	var limitedHandler http.Handler
 	if config.DisableAuth {
 		limitedHandler = newSimultaneousClientsLimiterV2(elasticHttpFrontentConnector, concurrentClientsLimitV2)

quesma/quesma/router_v2.go

@@ -33,7 +33,7 @@ import (
 	"time"
 )
 
-func ConfigureIngestRouterV2(cfg *config.QuesmaConfiguration, ip *ingest.IngestProcessor, phoneHomeAgent telemetry.PhoneHomeAgent, tableResolver table_resolver.TableResolver) *quesma_api.PathRouter {
+func ConfigureIngestRouterV2(cfg *config.QuesmaConfiguration, ip *ingest.IngestProcessor, phoneHomeAgent telemetry.PhoneHomeAgent, tableResolver table_resolver.TableResolver) quesma_api.Router {
 	// some syntactic sugar
 	method := quesma_api.IsHTTPMethod
 	and := quesma_api.And
@@ -127,7 +127,7 @@ func ConfigureIngestRouterV2(cfg *config.QuesmaConfiguration, ip *ingest.IngestP
 	return router
 }
 
-func ConfigureSearchRouterV2(cfg *config.QuesmaConfiguration, sr schema.Registry, lm *clickhouse.LogManager, console *ui.QuesmaManagementConsole, queryRunner *QueryRunner, tableResolver table_resolver.TableResolver) *quesma_api.PathRouter {
+func ConfigureSearchRouterV2(cfg *config.QuesmaConfiguration, sr schema.Registry, lm *clickhouse.LogManager, console *ui.QuesmaManagementConsole, queryRunner *QueryRunner, tableResolver table_resolver.TableResolver) quesma_api.Router {
 
 	// some syntactic sugar
 	method := quesma_api.IsHTTPMethod

quesma/v2/core/dispatch.go

@@ -3,10 +3,15 @@
 package quesma_api
 
 import (
+	"context"
 	"net/http"
 )
 
+// TODO currently there are two types of handlers, HTTPFrontendHandler and Handler
+// first one comes from v2 POC and the second one comes from v1 quesma
+// we need to unify them
 type HTTPFrontendHandler func(request *http.Request) (map[string]interface{}, any, error)
+type Handler func(ctx context.Context, req *Request) (*Result, error)
 
 type HandlersPipe struct {
 	Handler    HTTPFrontendHandler

quesma/v2/core/mux.go

@@ -3,7 +3,6 @@
 package quesma_api
 
 import (
-	"context"
 	"github.com/ucarion/urlpath"
 	"net/http"
 	"net/url"
@@ -15,12 +14,15 @@ type (
 	PathRouter struct {
 		mappings []mapping
 	}
+	HttpHandlersPipe struct {
+		Handler    Handler
+		Processors []Processor
+	}
 	mapping struct {
 		pattern      string
 		compiledPath urlpath.Path
 		predicate    RequestMatcher
-		handler      Handler
-		processors   []Processor
+		handler      *HttpHandlersPipe
 	}
 	Result struct {
 		Body       string
@@ -40,8 +42,6 @@ type (
 		ParsedBody RequestBody
 	}
 
-	Handler func(ctx context.Context, req *Request) (*Result, error)
-
 	MatchResult struct {
 		Matched  bool
 		Decision *Decision
@@ -78,14 +78,22 @@ func NewPathRouter() *PathRouter {
 	return &PathRouter{mappings: make([]mapping, 0)}
 }
 
+func (p *PathRouter) Clone() Cloner {
+	newRouter := NewPathRouter()
+	for _, mapping := range p.mappings {
+		newRouter.Register(mapping.pattern, mapping.predicate, mapping.handler.Handler)
+	}
+	return newRouter
+}
+
 func (p *PathRouter) Register(pattern string, predicate RequestMatcher, handler Handler) {
 
-	mapping := mapping{pattern, urlpath.New(pattern), predicate, handler, nil}
+	mapping := mapping{pattern, urlpath.New(pattern), predicate, &HttpHandlersPipe{Handler: handler}}
 	p.mappings = append(p.mappings, mapping)
 
 }
 
-func (p *PathRouter) Matches(req *Request) (Handler, *Decision) {
+func (p *PathRouter) Matches(req *Request) (*HttpHandlersPipe, *Decision) {
 	handler, decision := p.findHandler(req)
 	if handler != nil {
 		routerStatistics.addMatched(req.Path)
@@ -96,7 +104,7 @@ func (p *PathRouter) Matches(req *Request) (Handler, *Decision) {
 	}
 }
 
-func (p *PathRouter) findHandler(req *Request) (Handler, *Decision) {
+func (p *PathRouter) findHandler(req *Request) (*HttpHandlersPipe, *Decision) {
 	path := strings.TrimSuffix(req.Path, "/")
 	for _, m := range p.mappings {
 		meta, match := m.compiledPath.Match(path)
@@ -171,3 +179,24 @@ func (p *predicateAlways) Matches(req *Request) MatchResult {
 func Always() RequestMatcher {
 	return &predicateAlways{}
 }
+
+func (p *PathRouter) AddRoute(path string, handler HTTPFrontendHandler) {
+	// TODO: it seems that we can adapt this to register call
+	// p.Register(path, Always(), handler)
+	panic("not implemented")
+}
+func (p *PathRouter) AddFallbackHandler(handler HTTPFrontendHandler) {
+	panic("not implemented")
+}
+func (p *PathRouter) GetFallbackHandler() HTTPFrontendHandler {
+	panic("not implemented")
+}
+func (p *PathRouter) GetHandlers() map[string]HandlersPipe {
+	panic("not implemented")
+}
+func (p *PathRouter) SetHandlers(handlers map[string]HandlersPipe) {
+	panic("not implemented")
+}
+func (p *PathRouter) Multiplexer() *http.ServeMux {
+	panic("not implemented")
+}

quesma/v2/core/quesma_apis.go

@@ -16,6 +16,8 @@ type Router interface {
 	GetHandlers() map[string]HandlersPipe
 	SetHandlers(handlers map[string]HandlersPipe)
 	Multiplexer() *http.ServeMux
+	Register(pattern string, predicate RequestMatcher, handler Handler)
+	Matches(req *Request) (*HttpHandlersPipe, *Decision)
 }
 
 type FrontendConnector interface {

quesma/v2/core/quesma_pipeline.go

@@ -35,6 +35,13 @@ func (p *Pipeline) Build() PipelineBuilder {
 }
 
 func (p *Pipeline) Start() {
+	// TODO connectors for the same endpoint should be sharing the same listener
+	// This is a temporary solution to start all connectors
+	// some of them will fail to start
+	// because the port is already in use
+	// This works well from application perspective
+	// because we are copying routing table from all connectors
+	// however, bind error remains
 	for _, conn := range p.FrontendConnectors {
 		go conn.Listen()
 	}