Commit 882341c7 authored by Juan Batiz-Benet's avatar Juan Batiz-Benet

query: send back errors when channels somehow

this is not a good inteface. should fix it
parent a706e049
...@@ -79,6 +79,15 @@ func (d *datastore) Query(q dsq.Query) (*dsq.Results, error) { ...@@ -79,6 +79,15 @@ func (d *datastore) Query(q dsq.Query) (*dsq.Results, error) {
} }
i := d.DB.NewIterator(rnge, nil) i := d.DB.NewIterator(rnge, nil)
// buffer this channel so that we dont totally block leveldb if client
// is not reading from chan.
ch := make(chan dsq.Entry, 1000)
qr := dsq.ResultsWithEntriesChan(q, ch)
// qr := dsq.ResultsWithEntries(q, es)
go func() {
defer close(ch)
// offset // offset
if q.Offset > 0 { if q.Offset > 0 {
for j := 0; j < q.Offset; j++ { for j := 0; j < q.Offset; j++ {
...@@ -86,11 +95,11 @@ func (d *datastore) Query(q dsq.Query) (*dsq.Results, error) { ...@@ -86,11 +95,11 @@ func (d *datastore) Query(q dsq.Query) (*dsq.Results, error) {
} }
} }
var es []dsq.Entry sent := 0
for i.Next() { for i.Next() {
// limit // limit
if q.Limit > 0 && len(es) >= q.Limit { if q.Limit > 0 && sent >= q.Limit {
break break
} }
...@@ -103,20 +112,20 @@ func (d *datastore) Query(q dsq.Query) (*dsq.Results, error) { ...@@ -103,20 +112,20 @@ func (d *datastore) Query(q dsq.Query) (*dsq.Results, error) {
e.Value = buf e.Value = buf
} }
es = append(es, e) ch <- e
sent++
} }
i.Release() i.Release()
if err := i.Error(); err != nil { if err := i.Error(); err != nil {
return nil, err qr.Err() <- err
} }
}()
// Now, apply remaining pieces. // Now, apply remaining pieces.
q2 := q q2 := q
q2.Offset = 0 // already applied q2.Offset = 0 // already applied
q2.Limit = 0 // already applied q2.Limit = 0 // already applied
// TODO: make this async with:
// qr := dsq.ResultsWithEntriesChan(q, ch)
qr := dsq.ResultsWithEntries(q, es)
qr = q2.ApplyTo(qr) qr = q2.ApplyTo(qr)
qr.Query = q // set it back qr.Query = q // set it back
return qr, nil return qr, nil
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment