Michael McGuinness
87 lines
1.7 KiB
Go
87 lines
1.7 KiB
Go
package queryQueue
|
|
|
|
import (
|
|
"context"
|
|
"queryorchestration/internal/database"
|
|
"queryorchestration/internal/query"
|
|
)
|
|
|
|
func (q *Queue) getUnsyncedQueries() {
|
|
for _, query := range *q.collectorQueries {
|
|
isSynced := false
|
|
for _, result := range *q.results {
|
|
if result.QueryID != query.ID || result.QueryVersion != query.Version {
|
|
continue
|
|
}
|
|
isSynced = true
|
|
break
|
|
}
|
|
if isSynced {
|
|
continue
|
|
}
|
|
|
|
q.Add(&query)
|
|
}
|
|
}
|
|
|
|
func (c *Queue) getCollectorQueries(ctx context.Context) error {
|
|
if c.collectorQueries != nil {
|
|
return nil
|
|
}
|
|
|
|
id := database.MustToDBUUID(c.collector.ID)
|
|
|
|
queries, err := c.db.GetCollectorQueries(ctx, id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
cleanQueries := make([]query.Query, len(queries))
|
|
for index, dbQuery := range queries {
|
|
cleanQuery, err := query.ParseDBQuery(&dbQuery)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
cleanQueries[index] = *cleanQuery
|
|
}
|
|
|
|
c.collectorQueries = &cleanQueries
|
|
|
|
return nil
|
|
}
|
|
|
|
func (q *Queue) Add(qu *query.Query) {
|
|
dependentQueries := []query.Query{}
|
|
requiredIndex := -1
|
|
|
|
if q.unsyncedQueue == nil {
|
|
q.unsyncedQueue = &[]query.Query{}
|
|
} else {
|
|
for index, entry := range *q.unsyncedQueue {
|
|
if entry.ID == qu.ID {
|
|
return
|
|
}
|
|
if entry.ID == qu.RequiredQueryID {
|
|
requiredIndex = index
|
|
}
|
|
}
|
|
}
|
|
|
|
for _, entry := range *q.collectorQueries {
|
|
if entry.RequiredQueryID == qu.ID {
|
|
dependentQueries = append(dependentQueries, entry)
|
|
}
|
|
}
|
|
|
|
if requiredIndex != -1 {
|
|
*q.unsyncedQueue = append((*q.unsyncedQueue)[:requiredIndex+1], append([]query.Query{*qu}, (*q.unsyncedQueue)[requiredIndex+1:]...)...)
|
|
} else {
|
|
*q.unsyncedQueue = append([]query.Query{*qu}, *q.unsyncedQueue...)
|
|
}
|
|
|
|
for _, entry := range dependentQueries {
|
|
q.Add(&entry)
|
|
}
|
|
}
|