Unifying HTTPFrontendHandler (#1081)

This PR unifies `HTTPFrontendHandler`. Now `POC router` and v1
`PathRouter` uses the same handler signature:
```
type HTTPFrontendHandler func(ctx context.Context, req *Request) (*Result, error)
```

Author: pdelewski

quesma/frontend_connectors/basic_http_frontend_connector.go

@@ -85,15 +85,11 @@ func (router *HTTPRouter) Unlock() {
 	router.mutex.Unlock()
 }
 
-func (router *HTTPRouter) Multiplexer() *http.ServeMux {
-	return router.mux
-}
-
-func (router *HTTPRouter) Register(pattern string, predicate quesma_api.RequestMatcher, handler quesma_api.Handler) {
+func (router *HTTPRouter) Register(pattern string, predicate quesma_api.RequestMatcher, handler quesma_api.HTTPFrontendHandler) {
 	panic("not implemented")
 }
 
-func (router *HTTPRouter) Matches(req *quesma_api.Request) (*quesma_api.HttpHandlersPipe, *quesma_api.Decision) {
+func (router *HTTPRouter) Matches(req *quesma_api.Request) (*quesma_api.HandlersPipe, *quesma_api.Decision) {
 	panic("not implemented")
 }
 
@@ -132,8 +128,8 @@ func (h *BasicHTTPFrontendConnector) ServeHTTP(w http.ResponseWriter, req *http.
 			if h.router.GetFallbackHandler() != nil {
 				fmt.Printf("No handler found for path: %s\n", req.URL.Path)
 				handler := h.router.GetFallbackHandler()
-				_, message, _ := handler(req)
-				_, err := w.Write(message.([]byte))
+				result, _ := handler(context.Background(), &quesma_api.Request{OriginalRequest: req})
+				_, err := w.Write(result.GenericResult.([]byte))
 				if err != nil {
 					fmt.Printf("Error writing response: %s\n", err)
 				}
@@ -142,9 +138,9 @@ func (h *BasicHTTPFrontendConnector) ServeHTTP(w http.ResponseWriter, req *http.
 		return
 	}
 	http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		metadata, message, _ := handlerWrapper.Handler(req)
+		result, _ := handlerWrapper.Handler(context.Background(), &quesma_api.Request{OriginalRequest: req})
 
-		_, message = dispatcher.Dispatch(handlerWrapper.Processors, metadata, message)
+		_, message := dispatcher.Dispatch(handlerWrapper.Processors, result.Meta, result.GenericResult)
 		_, err := w.Write(message.([]byte))
 		if err != nil {
 			fmt.Printf("Error writing response: %s\n", err)

quesma/frontend_connectors/elasticsearch_ingest.go

@@ -4,6 +4,7 @@
 package frontend_connectors
 
 import (
+	"context"
 	"github.com/ucarion/urlpath"
 	"net/http"
 	quesma_api "quesma_v2/core"
@@ -44,18 +45,18 @@ func setContentType(w http.ResponseWriter) http.ResponseWriter {
 	return w
 }
 
-func bulk(request *http.Request) (map[string]interface{}, any, error) {
+func bulk(_ context.Context, request *quesma_api.Request) (*quesma_api.Result, error) {
 	metadata := quesma_api.MakeNewMetadata()
 	metadata[IngestAction] = BulkIndexAction
-	metadata[IngestTargetKey] = getIndexFromRequest(request)
-	return metadata, request, nil
+	metadata[IngestTargetKey] = getIndexFromRequest(request.OriginalRequest)
+	return &quesma_api.Result{Meta: metadata, GenericResult: request}, nil
 }
 
-func doc(request *http.Request) (map[string]interface{}, any, error) {
+func doc(_ context.Context, request *quesma_api.Request) (*quesma_api.Result, error) {
 	metadata := quesma_api.MakeNewMetadata()
 	metadata[IngestAction] = DocIndexAction
-	metadata[IngestTargetKey] = getIndexFromRequest(request)
-	return metadata, request, nil
+	metadata[IngestTargetKey] = getIndexFromRequest(request.OriginalRequest)
+	return &quesma_api.Result{Meta: metadata, GenericResult: request}, nil
 }
 
 func getIndexFromRequest(request *http.Request) string {

quesma/frontend_connectors/router_v2.go

@@ -51,7 +51,7 @@ func responseFromQuesmaV2(ctx context.Context, unzipped []byte, w http.ResponseW
 	logger.Debug().Str(logger.RID, id).Msg("responding from Quesma")
 
 	for key, value := range quesmaResponse.Meta {
-		w.Header().Set(key, value)
+		w.Header().Set(key, value.(string))
 	}
 	if zip {
 		w.Header().Set("Content-Encoding", "gzip")
@@ -223,7 +223,7 @@ func (r *RouterV2) Reroute(ctx context.Context, w http.ResponseWriter, req *http
 	}
 
 	quesmaRequest.ParsedBody = types.ParseRequestBody(quesmaRequest.Body)
-	var handler quesma_api.Handler
+	var handler quesma_api.HTTPFrontendHandler
 	var decision *quesma_api.Decision
 	searchHandlerPipe, searchDecision := searchRouter.Matches(quesmaRequest)
 	if searchDecision != nil {

quesma/main_test.go

@@ -6,7 +6,6 @@ package main
 import (
 	"context"
 	"github.com/stretchr/testify/assert"
-	"net/http"
 	"os"
 	"os/signal"
 	"quesma/backend_connectors"
@@ -54,11 +53,11 @@ func Test_backendConnectorValidation(t *testing.T) {
 
 var fallbackCalled int32 = 0
 
-func fallback(request *http.Request) (map[string]interface{}, any, error) {
+func fallback(_ context.Context, _ *quesma_api.Request) (*quesma_api.Result, error) {
 	metadata := quesma_api.MakeNewMetadata()
 	atomic.AddInt32(&fallbackCalled, 1)
 	resp := []byte("unknown\n")
-	return metadata, resp, nil
+	return &quesma_api.Result{Meta: metadata, GenericResult: resp}, nil
 }
 
 func ab_testing_scenario() quesma_api.QuesmaBuilder {

quesma/quesma/dual_write_proxy.go

@@ -198,7 +198,7 @@ func responseFromQuesma(ctx context.Context, unzipped []byte, w http.ResponseWri
 	logger.Debug().Str(logger.RID, id).Msg("responding from Quesma")
 
 	for key, value := range quesmaResponse.Meta {
-		w.Header().Set(key, value)
+		w.Header().Set(key, value.(string))
 	}
 	if zip {
 		w.Header().Set("Content-Encoding", "gzip")

quesma/quesma/router.go

@@ -469,7 +469,7 @@ func elasticsearchCountResult(body int64, statusCode int) (*quesma_api.Result, e
 	if err != nil {
 		return nil, err
 	}
-	return &quesma_api.Result{Body: string(serialized), Meta: map[string]string{
+	return &quesma_api.Result{Body: string(serialized), Meta: map[string]any{
 		"Content-Type":            "application/json",
 		"X-Quesma-Headers-Source": "Quesma",
 	}, StatusCode: statusCode}, nil
@@ -486,7 +486,7 @@ type countResult struct {
 }
 
 func elasticsearchQueryResult(body string, statusCode int) *quesma_api.Result {
-	return &quesma_api.Result{Body: body, Meta: map[string]string{
+	return &quesma_api.Result{Body: body, Meta: map[string]any{
 		// TODO copy paste from the original request
 		"X-Quesma-Headers-Source": "Quesma",
 	}, StatusCode: statusCode}
@@ -556,7 +556,7 @@ func bulkInsertResult(ctx context.Context, ops []bulk.BulkItem, err error) (*que
 }
 
 func elasticsearchInsertResult(body string, statusCode int) *quesma_api.Result {
-	return &quesma_api.Result{Body: body, Meta: map[string]string{
+	return &quesma_api.Result{Body: body, Meta: map[string]any{
 		// TODO copy paste from the original request
 		frontend_connectors.ContentTypeHeaderKey: "application/json",
 		"X-Quesma-Headers-Source":                "Quesma",
@@ -575,7 +575,7 @@ func resolveIndexResult(sources elasticsearch.Sources) (*quesma_api.Result, erro
 
 	return &quesma_api.Result{
 		Body:       string(body),
-		Meta:       map[string]string{},
+		Meta:       map[string]any{},
 		StatusCode: http.StatusOK}, nil
 }
 

quesma/v2/core/dispatch.go

@@ -4,14 +4,9 @@ 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 HTTPFrontendHandler func(ctx context.Context, req *Request) (*Result, error)
 
 type HandlersPipe struct {
 	Handler    HTTPFrontendHandler

quesma/v2/core/mux.go

@@ -14,22 +14,25 @@ type (
 	PathRouter struct {
 		mappings []mapping
 	}
-	HttpHandlersPipe struct {
-		Handler    Handler
-		Processors []Processor
-	}
 	mapping struct {
 		pattern      string
 		compiledPath urlpath.Path
 		predicate    RequestMatcher
-		handler      *HttpHandlersPipe
+		handler      *HandlersPipe
 	}
+	// Result is a kind of adapter for response
+	// to uniform v1 routing
+	// GenericResult is generic result that can be used by processors
 	Result struct {
-		Body       string
-		Meta       map[string]string
-		StatusCode int
+		Body          string
+		Meta          map[string]any
+		StatusCode    int
+		GenericResult any
 	}
 
+	// Request is kind of adapter for http.Request
+	// to uniform v1 routing
+	// it stores original http request
 	Request struct {
 		Method string
 		Path   string
@@ -40,6 +43,8 @@ type (
 
 		Body       string
 		ParsedBody RequestBody
+		// OriginalRequest is the original http.Request object that was received by the server.
+		OriginalRequest *http.Request
 	}
 
 	MatchResult struct {
@@ -56,14 +61,14 @@ type RequestMatcherFunc func(req *Request) MatchResult
 func ServerErrorResult() *Result {
 	return &Result{
 		StatusCode: http.StatusInternalServerError,
-		Meta:       map[string]string{"Content-Type": "text/plain"},
+		Meta:       map[string]any{"Content-Type": "text/plain"},
 	}
 }
 
 func BadReqeustResult() *Result {
 	return &Result{
 		StatusCode: http.StatusBadRequest,
-		Meta:       map[string]string{"Content-Type": "text/plain"},
+		Meta:       map[string]any{"Content-Type": "text/plain"},
 	}
 }
 
@@ -86,14 +91,14 @@ func (p *PathRouter) Clone() Cloner {
 	return newRouter
 }
 
-func (p *PathRouter) Register(pattern string, predicate RequestMatcher, handler Handler) {
+func (p *PathRouter) Register(pattern string, predicate RequestMatcher, handler HTTPFrontendHandler) {
 
-	mapping := mapping{pattern, urlpath.New(pattern), predicate, &HttpHandlersPipe{Handler: handler}}
+	mapping := mapping{pattern, urlpath.New(pattern), predicate, &HandlersPipe{Handler: handler}}
 	p.mappings = append(p.mappings, mapping)
 
 }
 
-func (p *PathRouter) Matches(req *Request) (*HttpHandlersPipe, *Decision) {
+func (p *PathRouter) Matches(req *Request) (*HandlersPipe, *Decision) {
 	handler, decision := p.findHandler(req)
 	if handler != nil {
 		routerStatistics.addMatched(req.Path)
@@ -104,7 +109,7 @@ func (p *PathRouter) Matches(req *Request) (*HttpHandlersPipe, *Decision) {
 	}
 }
 
-func (p *PathRouter) findHandler(req *Request) (*HttpHandlersPipe, *Decision) {
+func (p *PathRouter) findHandler(req *Request) (*HandlersPipe, *Decision) {
 	path := strings.TrimSuffix(req.Path, "/")
 	for _, m := range p.mappings {
 		meta, match := m.compiledPath.Match(path)
@@ -192,11 +197,14 @@ func (p *PathRouter) GetFallbackHandler() HTTPFrontendHandler {
 	panic("not implemented")
 }
 func (p *PathRouter) GetHandlers() map[string]HandlersPipe {
-	panic("not implemented")
+	callInfos := make(map[string]HandlersPipe)
+	for _, v := range p.mappings {
+		callInfos[v.pattern] = *v.handler
+	}
+	return callInfos
 }
 func (p *PathRouter) SetHandlers(handlers map[string]HandlersPipe) {
-	panic("not implemented")
-}
-func (p *PathRouter) Multiplexer() *http.ServeMux {
-	panic("not implemented")
+	for path, handler := range handlers {
+		p.mappings = append(p.mappings, mapping{pattern: path, compiledPath: urlpath.New(path), handler: &handler})
+	}
 }

quesma/v2/core/quesma_apis.go

@@ -5,7 +5,6 @@ package quesma_api
 import (
 	"context"
 	"net"
-	"net/http"
 )
 
 type Router interface {
@@ -15,9 +14,8 @@ type Router interface {
 	GetFallbackHandler() HTTPFrontendHandler
 	GetHandlers() map[string]HandlersPipe
 	SetHandlers(handlers map[string]HandlersPipe)
-	Multiplexer() *http.ServeMux
-	Register(pattern string, predicate RequestMatcher, handler Handler)
-	Matches(req *Request) (*HttpHandlersPipe, *Decision)
+	Register(pattern string, predicate RequestMatcher, handler HTTPFrontendHandler)
+	Matches(req *Request) (*HandlersPipe, *Decision)
 }
 
 type FrontendConnector interface {

quesma/v2_test_objects.go

@@ -4,6 +4,7 @@
 package main
 
 import (
+	"context"
 	"net/http"
 	"quesma/frontend_connectors"
 	"quesma/processors"
@@ -78,41 +79,41 @@ var responses = [][]byte{
 }`),
 }
 
-func bulk(request *http.Request) (map[string]interface{}, any, error) {
-	_, err := frontend_connectors.ReadRequestBody(request)
+func bulk(_ context.Context, request *quesma_api.Request) (*quesma_api.Result, error) {
+	_, err := frontend_connectors.ReadRequestBody(request.OriginalRequest)
 	if err != nil {
-		return nil, nil, err
+		return nil, err
 	}
 	metadata := quesma_api.MakeNewMetadata()
 	metadata["level"] = 0
 	resp := []byte("bulk\n")
 	atomic.AddInt64(&correlationId, 1)
 	quesma_api.SetCorrelationId(metadata, correlationId)
-	return metadata, resp, nil
+	return &quesma_api.Result{Meta: metadata, GenericResult: resp}, nil
 }
 
-func doc(request *http.Request) (map[string]interface{}, any, error) {
-	_, err := frontend_connectors.ReadRequestBody(request)
+func doc(_ context.Context, request *quesma_api.Request) (*quesma_api.Result, error) {
+	_, err := frontend_connectors.ReadRequestBody(request.OriginalRequest)
 	if err != nil {
-		return nil, nil, err
+		return nil, err
 	}
 	metadata := quesma_api.MakeNewMetadata()
 	metadata["level"] = 0
 	atomic.AddInt64(&correlationId, 1)
 	quesma_api.SetCorrelationId(metadata, correlationId)
 	resp := []byte("doc\n")
 
-	return metadata, resp, nil
+	return &quesma_api.Result{Meta: metadata, GenericResult: resp}, nil
 }
 
 var correlationId int64 = 0
 
-func search(request *http.Request) (map[string]interface{}, any, error) {
+func search(_ context.Context, request *quesma_api.Request) (*quesma_api.Result, error) {
 	metadata := quesma_api.MakeNewMetadata()
 	metadata["level"] = 0
 	atomic.AddInt64(&correlationId, 1)
 	quesma_api.SetCorrelationId(metadata, correlationId)
-	return metadata, request, nil
+	return &quesma_api.Result{Meta: metadata, GenericResult: request.OriginalRequest}, nil
 }
 
 type IngestProcessor struct {