Quesma v2 new frontend (#1057)

This PR is about few things:
- moving frontend, backend connectors, processors back to `quesma`
module
- refactoring to split ingest and search routing
- moving some code regarding http frontend and routing and doing some
cleanup

quesma/v2/frontend_connectors/basic_http_frontend_connector.go → quesma/frontend_connectors/basic_http_frontend_connector.go

@@ -14,9 +14,10 @@ import (
 )
 
 type HTTPRouter struct {
-	mux      *http.ServeMux                     // Default HTTP multiplexer
-	handlers map[string]quesma_api.HandlersPipe // Map to store custom route handlers
-	mutex    sync.RWMutex                       // Mutex for concurrent access to handlers
+	mux             *http.ServeMux                     // Default HTTP multiplexer
+	handlers        map[string]quesma_api.HandlersPipe // Map to store custom route handlers
+	fallbackHandler quesma_api.HTTPFrontendHandler
+	mutex           sync.RWMutex // Mutex for concurrent access to handlers
 }
 
 func NewHTTPRouter() *HTTPRouter {
@@ -34,13 +35,26 @@ func (router *HTTPRouter) AddRoute(path string, handler quesma_api.HTTPFrontendH
 	fmt.Printf("Added route: %s\n", path)
 }
 
+func (router *HTTPRouter) AddFallbackHandler(handler quesma_api.HTTPFrontendHandler) {
+	router.mutex.Lock()
+	defer router.mutex.Unlock()
+	router.fallbackHandler = handler
+}
+
+func (router *HTTPRouter) GetFallbackHandler() quesma_api.HTTPFrontendHandler {
+	router.mutex.RLock()
+	defer router.mutex.RUnlock()
+	return router.fallbackHandler
+}
+
 func (router *HTTPRouter) Clone() quesma_api.Cloner {
 	newRouter := NewHTTPRouter()
 	router.mutex.Lock()
 	defer router.mutex.Unlock()
 	for path, handler := range router.handlers {
 		newRouter.handlers[path] = handler
 	}
+	newRouter.fallbackHandler = router.fallbackHandler
 	return newRouter
 }
 
@@ -97,15 +111,25 @@ func (h *BasicHTTPFrontendConnector) GetRouter() quesma_api.Router {
 
 func (h *BasicHTTPFrontendConnector) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 	handlerWrapper, exists := h.router.GetHandlers()[req.URL.Path]
+	dispatcher := &quesma_api.Dispatcher{}
 	if !exists {
-		h.router.Multiplexer().ServeHTTP(w, req)
+		http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			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))
+				if err != nil {
+					fmt.Printf("Error writing response: %s\n", err)
+				}
+			}
+		}).ServeHTTP(w, req)
 		return
 	}
-	dispatcher := &quesma_api.Dispatcher{}
 	http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		metadata, message, _ := handlerWrapper.Handler(req)
 
-		metadata, message = dispatcher.Dispatch(handlerWrapper.Processors, metadata, message)
+		_, message = dispatcher.Dispatch(handlerWrapper.Processors, metadata, message)
 		_, err := w.Write(message.([]byte))
 		if err != nil {
 			fmt.Printf("Error writing response: %s\n", err)

quesma/v2/frontend_connectors/tcp_postgres_connection_handler.go → quesma/frontend_connectors/tcp_postgres_connection_handler.go

@@ -32,9 +32,12 @@ func (p *TcpPostgresConnectionHandler) HandleConnection(conn net.Conn) error {
 		}
 		var resp any = msg
 		metadata := make(map[string]interface{})
-		metadata, resp = dispatcher.Dispatch(p.processors, metadata, resp)
+		_, resp = dispatcher.Dispatch(p.processors, metadata, resp)
 		if resp != nil {
 			_, err = conn.Write(resp.([]byte))
+			if err != nil {
+				return fmt.Errorf("error sending response: %w", err)
+			}
 		}
 	}
 }

quesma/go.mod

@@ -10,12 +10,16 @@ require (
 	github.com/DataDog/go-sqllexer v0.0.17
 	github.com/barkimedes/go-deepcopy v0.0.0-20220514131651-17c30cfc62df
 	github.com/coreos/go-semver v0.3.1
+	github.com/go-sql-driver/mysql v1.8.1
 	github.com/goccy/go-json v0.10.3
+	github.com/google/go-cmp v0.6.0
 	github.com/google/uuid v1.6.0
 	github.com/gorilla/mux v1.8.1
 	github.com/gorilla/securecookie v1.1.2
 	github.com/gorilla/sessions v1.4.0
 	github.com/hashicorp/go-multierror v1.1.1
+	github.com/jackc/pgx/v4 v4.18.3
+	github.com/jackc/pgx/v5 v5.7.1
 	github.com/k0kubun/pp v3.0.1+incompatible
 	github.com/knadh/koanf/parsers/json v0.1.0
 	github.com/knadh/koanf/parsers/yaml v0.1.0
@@ -34,9 +38,16 @@ require (
 )
 
 require (
+	filippo.io/edwards25519 v1.1.0 // indirect
 	github.com/go-viper/mapstructure/v2 v2.2.1 // indirect
 	github.com/hashicorp/errwrap v1.0.0 // indirect
-	github.com/jackc/pgx/v5 v5.7.1 // indirect
+	github.com/jackc/chunkreader/v2 v2.0.1 // indirect
+	github.com/jackc/pgconn v1.14.3 // indirect
+	github.com/jackc/pgio v1.0.0 // indirect
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgproto3/v2 v2.3.3 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
+	github.com/jackc/pgtype v1.14.0 // indirect
 	github.com/k0kubun/colorstring v0.0.0-20150214042306-9440f1994b88 // indirect
 	github.com/knadh/koanf/maps v0.1.1 // indirect
 	github.com/kr/text v0.2.0 // indirect
@@ -46,6 +57,8 @@ require (
 	github.com/tidwall/gjson v1.18.0 // indirect
 	github.com/tidwall/match v1.1.1 // indirect
 	github.com/tidwall/pretty v1.2.1 // indirect
+	golang.org/x/crypto v0.28.0 // indirect
+	golang.org/x/text v0.19.0 // indirect
 )
 
 require (