tile38/controller/scanner.go

421 lines
9.1 KiB
Go
Raw Normal View History

2016-03-05 02:08:16 +03:00
package controller
import (
"bytes"
"errors"
2017-08-10 23:31:36 +03:00
"math"
2016-03-05 02:08:16 +03:00
"strconv"
2016-04-02 17:20:30 +03:00
"sync"
2016-03-05 02:08:16 +03:00
2016-03-29 00:16:21 +03:00
"github.com/tidwall/resp"
2016-03-06 17:55:00 +03:00
"github.com/tidwall/tile38/controller/collection"
2016-07-12 22:18:16 +03:00
"github.com/tidwall/tile38/controller/glob"
2016-03-29 00:16:21 +03:00
"github.com/tidwall/tile38/controller/server"
2016-03-05 02:08:16 +03:00
"github.com/tidwall/tile38/geojson"
)
const limitItems = 100
type outputT int
const (
outputUnknown outputT = iota
outputIDs
outputObjects
outputCount
outputPoints
outputHashes
outputBounds
)
type scanWriter struct {
2016-04-02 17:20:30 +03:00
mu sync.Mutex
2016-05-23 23:01:42 +03:00
c *Controller
2016-03-05 02:08:16 +03:00
wr *bytes.Buffer
2016-03-29 00:16:21 +03:00
msg *server.Message
2016-03-05 02:08:16 +03:00
col *collection.Collection
fmap map[string]int
farr []string
fvals []float64
output outputT
wheres []whereT
numberItems uint64
nofields bool
cursor uint64
2016-03-05 02:08:16 +03:00
limit uint64
hitLimit bool
once bool
count uint64
precision uint64
2016-07-13 07:51:01 +03:00
globPattern string
2016-03-05 02:08:16 +03:00
globEverything bool
globSingle bool
fullFields bool
2016-03-29 00:16:21 +03:00
values []resp.Value
2016-07-13 06:11:02 +03:00
matchValues bool
2016-03-05 02:08:16 +03:00
}
2017-01-10 19:49:48 +03:00
type ScanWriterParams struct {
id string
o geojson.Object
fields []float64
2017-01-10 19:49:48 +03:00
distance float64
noLock bool
2017-01-10 19:49:48 +03:00
}
2016-03-05 02:08:16 +03:00
func (c *Controller) newScanWriter(
2016-07-13 06:11:02 +03:00
wr *bytes.Buffer, msg *server.Message, key string, output outputT,
precision uint64, globPattern string, matchValues bool,
cursor, limit uint64, wheres []whereT, nofields bool,
2016-03-05 02:08:16 +03:00
) (
*scanWriter, error,
) {
switch output {
default:
return nil, errors.New("invalid output type")
case outputIDs, outputObjects, outputCount, outputBounds, outputPoints, outputHashes:
}
2017-08-10 23:31:36 +03:00
if limit == 0 {
if output == outputCount {
limit = math.MaxUint64
} else {
limit = limitItems
}
}
2016-03-05 02:08:16 +03:00
sw := &scanWriter{
2016-07-13 06:11:02 +03:00
c: c,
wr: wr,
msg: msg,
cursor: cursor,
2016-07-13 07:51:01 +03:00
limit: limit,
2016-07-13 06:11:02 +03:00
wheres: wheres,
2016-07-13 07:51:01 +03:00
output: output,
2016-07-13 06:11:02 +03:00
nofields: nofields,
2016-07-13 07:51:01 +03:00
precision: precision,
globPattern: globPattern,
2016-07-13 06:11:02 +03:00
matchValues: matchValues,
2016-03-05 02:08:16 +03:00
}
2016-07-12 22:18:16 +03:00
if globPattern == "*" || globPattern == "" {
2016-03-05 02:08:16 +03:00
sw.globEverything = true
} else {
2016-07-12 22:18:16 +03:00
if !glob.IsGlob(globPattern) {
2016-03-05 02:08:16 +03:00
sw.globSingle = true
}
}
sw.col = c.getCol(key)
if sw.col != nil {
sw.fmap = sw.col.FieldMap()
sw.farr = sw.col.FieldArr()
}
sw.fvals = make([]float64, len(sw.farr))
return sw, nil
}
func (sw *scanWriter) hasFieldsOutput() bool {
switch sw.output {
default:
return false
case outputObjects, outputPoints, outputHashes, outputBounds:
return !sw.nofields
}
}
func (sw *scanWriter) writeHead() {
2016-04-02 17:20:30 +03:00
sw.mu.Lock()
defer sw.mu.Unlock()
2016-03-29 00:16:21 +03:00
switch sw.msg.OutputType {
case server.JSON:
if len(sw.farr) > 0 && sw.hasFieldsOutput() {
sw.wr.WriteString(`,"fields":[`)
for i, field := range sw.farr {
if i > 0 {
sw.wr.WriteByte(',')
}
sw.wr.WriteString(jsonString(field))
2016-03-05 02:08:16 +03:00
}
2016-03-29 00:16:21 +03:00
sw.wr.WriteByte(']')
2016-03-05 02:08:16 +03:00
}
2016-03-29 00:16:21 +03:00
switch sw.output {
case outputIDs:
sw.wr.WriteString(`,"ids":[`)
case outputObjects:
sw.wr.WriteString(`,"objects":[`)
case outputPoints:
sw.wr.WriteString(`,"points":[`)
case outputBounds:
sw.wr.WriteString(`,"bounds":[`)
case outputHashes:
sw.wr.WriteString(`,"hashes":[`)
case outputCount:
2016-03-05 02:08:16 +03:00
2016-03-29 00:16:21 +03:00
}
case server.RESP:
2016-03-05 02:08:16 +03:00
}
}
func (sw *scanWriter) writeFoot() {
2016-04-02 17:20:30 +03:00
sw.mu.Lock()
defer sw.mu.Unlock()
2017-07-24 18:42:12 +03:00
cursor := sw.cursor + sw.numberItems
2016-03-05 02:08:16 +03:00
if !sw.hitLimit {
cursor = 0
}
2016-03-29 00:16:21 +03:00
switch sw.msg.OutputType {
case server.JSON:
switch sw.output {
default:
sw.wr.WriteByte(']')
case outputCount:
2016-03-05 02:08:16 +03:00
2016-03-29 00:16:21 +03:00
}
sw.wr.WriteString(`,"count":` + strconv.FormatUint(sw.count, 10))
sw.wr.WriteString(`,"cursor":` + strconv.FormatUint(cursor, 10))
case server.RESP:
sw.wr.Reset()
2016-05-24 00:21:18 +03:00
var data []byte
var err error
if sw.output == outputCount {
data, err = resp.IntegerValue(int(sw.count)).MarshalRESP()
} else {
values := []resp.Value{
resp.IntegerValue(int(cursor)),
resp.ArrayValue(sw.values),
}
data, err = resp.ArrayValue(values).MarshalRESP()
2016-03-29 00:16:21 +03:00
}
if err != nil {
panic("Eek this is bad. Marshal resp should not fail.")
}
sw.wr.Write(data)
2016-03-05 02:08:16 +03:00
}
}
func (sw *scanWriter) fieldMatch(fields []float64, o geojson.Object) ([]float64, bool) {
var z float64
var gotz bool
if !sw.hasFieldsOutput() || sw.fullFields {
for _, where := range sw.wheres {
if where.field == "z" {
if !gotz {
z = o.CalculatedPoint().Z
}
if !where.match(z) {
return sw.fvals, false
}
continue
}
var value float64
idx, ok := sw.fmap[where.field]
if ok {
if len(fields) > idx {
value = fields[idx]
}
}
if !where.match(value) {
return sw.fvals, false
}
}
} else {
for idx := range sw.farr {
var value float64
if len(fields) > idx {
value = fields[idx]
}
sw.fvals[idx] = value
}
for _, where := range sw.wheres {
if where.field == "z" {
if !gotz {
z = o.CalculatedPoint().Z
}
if !where.match(z) {
return sw.fvals, false
}
continue
}
var value float64
idx, ok := sw.fmap[where.field]
if ok {
value = sw.fvals[idx]
}
if !where.match(value) {
return sw.fvals, false
}
}
}
return sw.fvals, true
}
2017-01-10 19:49:48 +03:00
//id string, o geojson.Object, fields []float64, noLock bool
func (sw *scanWriter) writeObject(opts ScanWriterParams) bool {
if !opts.noLock {
2016-04-03 00:13:20 +03:00
sw.mu.Lock()
defer sw.mu.Unlock()
}
2016-03-05 02:08:16 +03:00
keepGoing := true
if !sw.globEverything {
if sw.globSingle {
2017-01-10 19:49:48 +03:00
if sw.globPattern != opts.id {
2016-03-05 02:08:16 +03:00
return true
}
keepGoing = false // return current object and stop iterating
} else {
2016-07-13 06:11:02 +03:00
var val string
if sw.matchValues {
2017-01-10 19:49:48 +03:00
val = opts.o.String()
2016-07-13 06:11:02 +03:00
} else {
2017-01-10 19:49:48 +03:00
val = opts.id
2016-07-13 06:11:02 +03:00
}
2016-07-13 07:51:01 +03:00
ok, _ := glob.Match(sw.globPattern, val)
2016-03-05 02:08:16 +03:00
if !ok {
return true
}
}
}
2017-01-10 19:49:48 +03:00
nfields, ok := sw.fieldMatch(opts.fields, opts.o)
2016-03-05 02:08:16 +03:00
if !ok {
return true
}
sw.count++
2017-07-24 18:42:12 +03:00
if sw.count <= sw.cursor {
return true
}
2016-03-05 02:08:16 +03:00
if sw.output == outputCount {
2017-07-26 06:23:21 +03:00
return sw.count < sw.limit
2016-03-05 02:08:16 +03:00
}
2016-03-29 00:16:21 +03:00
switch sw.msg.OutputType {
case server.JSON:
var wr bytes.Buffer
var jsfields string
if sw.once {
wr.WriteByte(',')
} else {
sw.once = true
}
if sw.hasFieldsOutput() {
if sw.fullFields {
if len(sw.fmap) > 0 {
jsfields = `,"fields":{`
var i int
for field, idx := range sw.fmap {
2017-01-10 19:49:48 +03:00
if len(opts.fields) > idx {
if opts.fields[idx] != 0 {
2016-03-29 00:16:21 +03:00
if i > 0 {
jsfields += `,`
}
2017-01-10 19:49:48 +03:00
jsfields += jsonString(field) + ":" + strconv.FormatFloat(opts.fields[idx], 'f', -1, 64)
2016-03-29 00:16:21 +03:00
i++
2016-03-05 02:08:16 +03:00
}
}
}
2016-03-29 00:16:21 +03:00
jsfields += `}`
2016-03-05 02:08:16 +03:00
}
2016-03-29 00:16:21 +03:00
} else if len(sw.farr) > 0 {
jsfields = `,"fields":[`
for i, field := range nfields {
if i > 0 {
jsfields += ","
}
jsfields += strconv.FormatFloat(field, 'f', -1, 64)
}
jsfields += `]`
2016-03-05 02:08:16 +03:00
}
2016-03-29 00:16:21 +03:00
}
if sw.output == outputIDs {
2017-01-10 19:49:48 +03:00
wr.WriteString(jsonString(opts.id))
2016-03-29 00:16:21 +03:00
} else {
2017-01-10 19:49:48 +03:00
wr.WriteString(`{"id":` + jsonString(opts.id))
2016-03-29 00:16:21 +03:00
switch sw.output {
case outputObjects:
2017-01-10 19:49:48 +03:00
wr.WriteString(`,"object":` + opts.o.JSON())
2016-03-29 00:16:21 +03:00
case outputPoints:
2017-01-10 19:49:48 +03:00
wr.WriteString(`,"point":` + opts.o.CalculatedPoint().ExternalJSON())
2016-03-29 00:16:21 +03:00
case outputHashes:
2017-01-10 19:49:48 +03:00
p, err := opts.o.Geohash(int(sw.precision))
2016-03-29 00:16:21 +03:00
if err != nil {
p = ""
2016-03-05 02:08:16 +03:00
}
2016-03-29 00:16:21 +03:00
wr.WriteString(`,"hash":"` + p + `"`)
case outputBounds:
2017-01-10 19:49:48 +03:00
wr.WriteString(`,"bounds":` + opts.o.CalculatedBBox().ExternalJSON())
2016-03-05 02:08:16 +03:00
}
2017-01-10 19:49:48 +03:00
2016-03-29 00:16:21 +03:00
wr.WriteString(jsfields)
2017-01-10 19:49:48 +03:00
if opts.distance > 0 {
wr.WriteString(`,"distance":` + strconv.FormatFloat(opts.distance, 'f', 2, 64))
}
2016-03-29 00:16:21 +03:00
wr.WriteString(`}`)
2016-03-05 02:08:16 +03:00
}
2016-03-29 00:16:21 +03:00
sw.wr.Write(wr.Bytes())
case server.RESP:
vals := make([]resp.Value, 1, 3)
2017-01-10 19:49:48 +03:00
vals[0] = resp.StringValue(opts.id)
2016-03-29 00:16:21 +03:00
if sw.output == outputIDs {
sw.values = append(sw.values, vals[0])
} else {
switch sw.output {
case outputObjects:
2017-01-10 19:49:48 +03:00
vals = append(vals, resp.StringValue(opts.o.String()))
2016-03-29 00:16:21 +03:00
case outputPoints:
2017-01-10 19:49:48 +03:00
point := opts.o.CalculatedPoint()
2016-03-29 00:16:21 +03:00
if point.Z != 0 {
vals = append(vals, resp.ArrayValue([]resp.Value{
resp.FloatValue(point.Y),
resp.FloatValue(point.X),
resp.FloatValue(point.Z),
}))
} else {
vals = append(vals, resp.ArrayValue([]resp.Value{
resp.FloatValue(point.Y),
resp.FloatValue(point.X),
}))
}
case outputHashes:
2017-01-10 19:49:48 +03:00
p, err := opts.o.Geohash(int(sw.precision))
2016-03-29 00:16:21 +03:00
if err != nil {
p = ""
}
vals = append(vals, resp.StringValue(p))
case outputBounds:
2017-01-10 19:49:48 +03:00
bbox := opts.o.CalculatedBBox()
2016-03-29 00:16:21 +03:00
vals = append(vals, resp.ArrayValue([]resp.Value{
resp.ArrayValue([]resp.Value{
resp.FloatValue(bbox.Min.Y),
resp.FloatValue(bbox.Min.X),
}),
resp.ArrayValue([]resp.Value{
resp.FloatValue(bbox.Max.Y),
resp.FloatValue(bbox.Max.X),
}),
}))
2016-03-05 02:08:16 +03:00
}
2016-03-29 00:16:21 +03:00
if sw.hasFieldsOutput() {
fvs := orderFields(sw.fmap, opts.fields)
if len(fvs) > 0 {
fvals := make([]resp.Value, 0, len(fvs)*2)
for i, fv := range fvs {
fvals = append(fvals, resp.StringValue(fv.field), resp.StringValue(strconv.FormatFloat(fv.value, 'f', -1, 64)))
i++
}
vals = append(vals, resp.ArrayValue(fvals))
2016-03-29 00:16:21 +03:00
}
}
2017-01-10 19:49:48 +03:00
if opts.distance > 0 {
vals = append(vals, resp.FloatValue(opts.distance))
}
2016-03-29 00:16:21 +03:00
sw.values = append(sw.values, resp.ArrayValue(vals))
2016-03-05 02:08:16 +03:00
}
}
sw.numberItems++
if sw.numberItems == sw.limit {
sw.hitLimit = true
return false
}
return keepGoing
}