Skip to content
This repository was archived by the owner on Nov 7, 2025. It is now read-only.

Commit 0ba6514

Browse files
pdelewskipdelewski
authored
PathRouter has to be moved up to v2 (#1065)
In order to use it in v2 `quesma_api` I had to move it up to v2
1 parent 0923f81 commit 0ba6514

33 files changed

+301
-491
lines changed

‎quesma/go.mod‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,6 @@ require (
3131
github.com/stretchr/testify v1.10.0
3232
github.com/tailscale/hujson v0.0.0-20221223112325-20486734a56a
3333
github.com/tidwall/sjson v1.2.5
34-
github.com/ucarion/urlpath v0.0.0-20200424170820-7ccc79b76bbb
3534
golang.org/x/exp v0.0.0-20241108190413-2d47ceb2692f
3635
golang.org/x/oauth2 v0.24.0
3736
quesma_v2 v0.0.0-00010101000000-000000000000
@@ -57,6 +56,7 @@ require (
5756
github.com/tidwall/gjson v1.18.0 // indirect
5857
github.com/tidwall/match v1.1.1 // indirect
5958
github.com/tidwall/pretty v1.2.1 // indirect
59+
github.com/ucarion/urlpath v0.0.0-20200424170820-7ccc79b76bbb // indirect
6060
golang.org/x/crypto v0.28.0 // indirect
6161
golang.org/x/text v0.19.0 // indirect
6262
)

‎quesma/ingest/common_table_test.go‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -9,13 +9,13 @@ import (
99
"github.com/stretchr/testify/assert"
1010
"quesma/clickhouse"
1111
"quesma/common_table"
12-
"quesma/frontend_connectors"
1312
"quesma/jsonprocessor"
1413
"quesma/persistence"
1514
"quesma/quesma/config"
1615
"quesma/quesma/types"
1716
"quesma/schema"
1817
"quesma/table_resolver"
18+
"quesma_v2/core/mux"
1919
"testing"
2020
)
2121

@@ -192,9 +192,9 @@ func TestIngestToCommonTable(t *testing.T) {
192192

193193
resolver := table_resolver.NewEmptyTableResolver()
194194

195-
decision := &frontend_connectors.Decision{
196-
UseConnectors: []frontend_connectors.ConnectorDecision{
197-
&frontend_connectors.ConnectorDecisionClickhouse{
195+
decision := &mux.Decision{
196+
UseConnectors: []mux.ConnectorDecision{
197+
&mux.ConnectorDecisionClickhouse{
198198
ClickhouseTableName: common_table.TableName,
199199
ClickhouseTables: []string{indexName},
200200
IsCommonTable: true,

‎quesma/ingest/ingest_validator_test.go‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -9,11 +9,11 @@ import (
99
"github.com/DATA-DOG/go-sqlmock"
1010
"github.com/stretchr/testify/assert"
1111
"quesma/clickhouse"
12-
"quesma/frontend_connectors"
1312
"quesma/quesma/config"
1413
"quesma/quesma/types"
1514
"quesma/table_resolver"
1615
"quesma/util"
16+
"quesma_v2/core/mux"
1717
"strings"
1818
"testing"
1919
)
@@ -172,8 +172,8 @@ func TestIngestValidation(t *testing.T) {
172172
ip.tableDiscovery = clickhouse.NewTableDiscoveryWith(&config.QuesmaConfiguration{}, nil, *tableMap)
173173

174174
resolver := table_resolver.NewEmptyTableResolver()
175-
decision := &frontend_connectors.Decision{
176-
UseConnectors: []frontend_connectors.ConnectorDecision{&frontend_connectors.ConnectorDecisionClickhouse{
175+
decision := &mux.Decision{
176+
UseConnectors: []mux.ConnectorDecision{&mux.ConnectorDecisionClickhouse{
177177
ClickhouseTableName: "test_table",
178178
}}}
179179
resolver.Decisions["test_table"] = decision

‎quesma/ingest/insert_test.go‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -8,14 +8,14 @@ import (
88
"github.com/DATA-DOG/go-sqlmock"
99
"github.com/stretchr/testify/assert"
1010
"quesma/clickhouse"
11-
"quesma/frontend_connectors"
1211
"quesma/jsonprocessor"
1312
"quesma/persistence"
1413
"quesma/quesma/config"
1514
"quesma/quesma/types"
1615
"quesma/schema"
1716
"quesma/table_resolver"
1817
"quesma/util"
18+
"quesma_v2/core/mux"
1919
"slices"
2020
"strconv"
2121
"strings"
@@ -240,8 +240,8 @@ func TestProcessInsertQuery(t *testing.T) {
240240
db, mock := util.InitSqlMockWithPrettyPrint(t, true)
241241
ip.ip.chDb = db
242242
resolver := table_resolver.NewEmptyTableResolver()
243-
decision := &frontend_connectors.Decision{
244-
UseConnectors: []frontend_connectors.ConnectorDecision{&frontend_connectors.ConnectorDecisionClickhouse{
243+
decision := &mux.Decision{
244+
UseConnectors: []mux.ConnectorDecision{&mux.ConnectorDecisionClickhouse{
245245
ClickhouseTableName: "test_table",
246246
}}}
247247
resolver.Decisions["test_table"] = decision
@@ -425,8 +425,8 @@ func TestCreateTableIfSomeFieldsExistsInSchemaAlready(t *testing.T) {
425425
schemaRegistry.Tables[schema.TableName(indexName)] = indexSchema
426426

427427
resolver := table_resolver.NewEmptyTableResolver()
428-
decision := &frontend_connectors.Decision{
429-
UseConnectors: []frontend_connectors.ConnectorDecision{&frontend_connectors.ConnectorDecisionClickhouse{
428+
decision := &mux.Decision{
429+
UseConnectors: []mux.ConnectorDecision{&mux.ConnectorDecisionClickhouse{
430430
ClickhouseTableName: "test_index",
431431
}}}
432432
resolver.Decisions["test_index"] = decision

‎quesma/ingest/processor.go‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,6 @@ import (
1212
"quesma/comment_metadata"
1313
"quesma/common_table"
1414
"quesma/end_user_errors"
15-
"quesma/frontend_connectors"
1615
"quesma/jsonprocessor"
1716
"quesma/logger"
1817
"quesma/model"
@@ -25,6 +24,7 @@ import (
2524
"quesma/table_resolver"
2625
"quesma/telemetry"
2726
"quesma/util"
27+
"quesma_v2/core/mux"
2828
"slices"
2929
"sort"
3030
"strings"
@@ -696,7 +696,7 @@ func (lm *IngestProcessor) ProcessInsertQuery(ctx context.Context, tableName str
696696
jsonData []types.JSON, transformer jsonprocessor.IngestTransformer,
697697
tableFormatter TableColumNameFormatter) error {
698698

699-
decision := lm.tableResolver.Resolve(frontend_connectors.IngestPipeline, tableName)
699+
decision := lm.tableResolver.Resolve(mux.IngestPipeline, tableName)
700700

701701
if decision.Err != nil {
702702
return decision.Err
@@ -712,10 +712,10 @@ func (lm *IngestProcessor) ProcessInsertQuery(ctx context.Context, tableName str
712712

713713
for _, connectorDecision := range decision.UseConnectors {
714714

715-
var clickhouseDecision *frontend_connectors.ConnectorDecisionClickhouse
715+
var clickhouseDecision *mux.ConnectorDecisionClickhouse
716716

717717
var ok bool
718-
if clickhouseDecision, ok = connectorDecision.(*frontend_connectors.ConnectorDecisionClickhouse); !ok {
718+
if clickhouseDecision, ok = connectorDecision.(*mux.ConnectorDecisionClickhouse); !ok {
719719
continue
720720
}
721721

‎quesma/quesma/dual_write_proxy.go‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -15,9 +15,6 @@ import (
1515
"quesma/elasticsearch"
1616
"quesma/end_user_errors"
1717
"quesma/feature"
18-
"quesma/frontend_connectors"
19-
"quesma/frontend_connectors/mux"
20-
"quesma/frontend_connectors/routes"
2118
"quesma/ingest"
2219
"quesma/logger"
2320
"quesma/queryparser"
@@ -32,6 +29,8 @@ import (
3229
"quesma/telemetry"
3330
"quesma/tracing"
3431
"quesma/util"
32+
"quesma_v2/core/mux"
33+
"quesma_v2/core/routes"
3534
"strconv"
3635
"strings"
3736
"sync/atomic"
@@ -54,7 +53,6 @@ func newSimultaneousClientsLimiter(handler http.Handler, limit int64) *simultane
5453
}
5554

5655
func (c *simultaneousClientsLimiter) ServeHTTP(w http.ResponseWriter, r *http.Request) {
57-
5856
current := c.counter.Load()
5957
// this is hard limit, we should not allow to go over it
6058
if current >= c.limit {
@@ -369,7 +367,7 @@ func (r *router) reroute(ctx context.Context, w http.ResponseWriter, req *http.R
369367
}
370368

371369
for _, connector := range decision.UseConnectors {
372-
if _, ok := connector.(*frontend_connectors.ConnectorDecisionElastic); ok {
370+
if _, ok := connector.(*mux.ConnectorDecisionElastic); ok {
373371
// this is desired elastic call
374372
sendToElastic = true
375373
break

‎quesma/quesma/dual_write_proxy_v2.go‎

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -15,9 +15,6 @@ import (
1515
"quesma/elasticsearch"
1616
"quesma/end_user_errors"
1717
"quesma/feature"
18-
"quesma/frontend_connectors"
19-
"quesma/frontend_connectors/mux"
20-
"quesma/frontend_connectors/routes"
2118
"quesma/ingest"
2219
"quesma/logger"
2320
"quesma/queryparser"
@@ -32,6 +29,8 @@ import (
3229
"quesma/telemetry"
3330
"quesma/tracing"
3431
"quesma/util"
32+
"quesma_v2/core/mux"
33+
"quesma_v2/core/routes"
3534
"strconv"
3635
"strings"
3736
"sync/atomic"
@@ -276,7 +275,7 @@ func (*routerV2) closedIndexResponse(ctx context.Context, w http.ResponseWriter,
276275

277276
}
278277

279-
func (r *routerV2) elasticFallback(decision *frontend_connectors.Decision,
278+
func (r *routerV2) elasticFallback(decision *mux.Decision,
280279
ctx context.Context, w http.ResponseWriter,
281280
req *http.Request, reqBody []byte, logManager *clickhouse.LogManager) {
282281

@@ -307,7 +306,7 @@ func (r *routerV2) elasticFallback(decision *frontend_connectors.Decision,
307306
}
308307

309308
for _, connector := range decision.UseConnectors {
310-
if _, ok := connector.(*frontend_connectors.ConnectorDecisionElastic); ok {
309+
if _, ok := connector.(*mux.ConnectorDecisionElastic); ok {
311310
// this is desired elastic call
312311
sendToElastic = true
313312
break
@@ -360,7 +359,7 @@ func (r *routerV2) reroute(ctx context.Context, w http.ResponseWriter, req *http
360359

361360
quesmaRequest.ParsedBody = types.ParseRequestBody(quesmaRequest.Body)
362361
var handler mux.Handler
363-
var decision *frontend_connectors.Decision
362+
var decision *mux.Decision
364363
searchHandler, searchDecision := searchRouter.Matches(quesmaRequest)
365364
if searchDecision != nil {
366365
decision = searchDecision

‎quesma/quesma/elastic_http_frontend_connector.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,9 +7,9 @@ import (
77
"net/http"
88
"quesma/clickhouse"
99
"quesma/frontend_connectors"
10-
"quesma/frontend_connectors/mux"
1110
"quesma/quesma/recovery"
1211
"quesma/telemetry"
12+
"quesma_v2/core/mux"
1313
)
1414

1515
type ElasticHttpFrontendConnector struct {

‎quesma/quesma/functionality/bulk/bulk.go‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,6 @@ import (
1111
"quesma/clickhouse"
1212
"quesma/elasticsearch"
1313
"quesma/end_user_errors"
14-
"quesma/frontend_connectors"
1514
"quesma/ingest"
1615
"quesma/logger"
1716
"quesma/queryparser"
@@ -21,6 +20,7 @@ import (
2120
"quesma/stats"
2221
"quesma/table_resolver"
2322
"quesma/telemetry"
23+
"quesma_v2/core/mux"
2424
"sort"
2525
"strings"
2626
"sync"
@@ -139,7 +139,7 @@ func splitBulk(ctx context.Context, defaultIndex *string, bulk types.NDJSON, bul
139139
}
140140
}
141141

142-
decision := tableResolver.Resolve(frontend_connectors.IngestPipeline, index)
142+
decision := tableResolver.Resolve(mux.IngestPipeline, index)
143143

144144
if decision.Err != nil {
145145
return decision.Err
@@ -181,7 +181,7 @@ func splitBulk(ctx context.Context, defaultIndex *string, bulk types.NDJSON, bul
181181

182182
switch connector.(type) {
183183

184-
case *frontend_connectors.ConnectorDecisionElastic:
184+
case *mux.ConnectorDecisionElastic:
185185
// Bulk entry for Elastic - forward the request as-is
186186
opBytes, err := rawOp.Bytes()
187187
if err != nil {
@@ -199,7 +199,7 @@ func splitBulk(ctx context.Context, defaultIndex *string, bulk types.NDJSON, bul
199199

200200
elasticBulkEntries = append(elasticBulkEntries, entryWithResponse)
201201

202-
case *frontend_connectors.ConnectorDecisionClickhouse:
202+
case *mux.ConnectorDecisionClickhouse:
203203

204204
// Bulk entry for Clickhouse
205205
if operation != "create" && operation != "index" {

‎quesma/quesma/matchers.go‎

Lines changed: 7 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -3,13 +3,12 @@
33
package quesma
44

55
import (
6-
"quesma/frontend_connectors"
7-
"quesma/frontend_connectors/mux"
86
"quesma/logger"
97
"quesma/quesma/config"
108
"quesma/quesma/types"
119
"quesma/table_resolver"
1210
"quesma/tracing"
11+
"quesma_v2/core/mux"
1312
"strings"
1413
)
1514

@@ -34,15 +33,15 @@ func matchedAgainstBulkBody(configuration *config.QuesmaConfiguration, tableReso
3433
if idx%2 == 0 {
3534
name := extractIndexName(s)
3635

37-
decision := tableResolver.Resolve(frontend_connectors.IngestPipeline, name)
36+
decision := tableResolver.Resolve(mux.IngestPipeline, name)
3837

3938
if decision.IsClosed {
4039
return mux.MatchResult{Matched: true, Decision: decision}
4140
}
4241

4342
// if have any enabled Clickhouse connector, then return true
4443
for _, connector := range decision.UseConnectors {
45-
if _, ok := connector.(*frontend_connectors.ConnectorDecisionClickhouse); ok {
44+
if _, ok := connector.(*mux.ConnectorDecisionClickhouse); ok {
4645
return mux.MatchResult{Matched: true, Decision: decision}
4746
}
4847
}
@@ -57,7 +56,7 @@ func matchedAgainstBulkBody(configuration *config.QuesmaConfiguration, tableReso
5756

5857
// Query path only (looks at QueryTarget)
5958
func matchedAgainstPattern(indexRegistry table_resolver.TableResolver) mux.RequestMatcher {
60-
return matchAgainstTableResolver(indexRegistry, frontend_connectors.QueryPipeline)
59+
return matchAgainstTableResolver(indexRegistry, mux.QueryPipeline)
6160
}
6261

6362
// check whether exact index name is enabled
@@ -71,7 +70,7 @@ func matchAgainstTableResolver(indexRegistry table_resolver.TableResolver, pipel
7170
return mux.MatchResult{Matched: false, Decision: decision}
7271
}
7372
for _, connector := range decision.UseConnectors {
74-
if _, ok := connector.(*frontend_connectors.ConnectorDecisionClickhouse); ok {
73+
if _, ok := connector.(*mux.ConnectorDecisionClickhouse); ok {
7574
return mux.MatchResult{Matched: true, Decision: decision}
7675
}
7776
}
@@ -80,11 +79,11 @@ func matchAgainstTableResolver(indexRegistry table_resolver.TableResolver, pipel
8079
}
8180

8281
func matchedExactQueryPath(indexRegistry table_resolver.TableResolver) mux.RequestMatcher {
83-
return matchAgainstTableResolver(indexRegistry, frontend_connectors.QueryPipeline)
82+
return matchAgainstTableResolver(indexRegistry, mux.QueryPipeline)
8483
}
8584

8685
func matchedExactIngestPath(indexRegistry table_resolver.TableResolver) mux.RequestMatcher {
87-
return matchAgainstTableResolver(indexRegistry, frontend_connectors.IngestPipeline)
86+
return matchAgainstTableResolver(indexRegistry, mux.IngestPipeline)
8887
}
8988

9089
// Returns false if the body contains a Kibana internal search.

0 commit comments

Comments
 (0)