1
0
mirror of https://github.com/mainflux/mainflux.git synced 2025-05-01 13:48:56 +08:00
Manuel Imperiale c8069827bb MF-506 - Fix influx & mongo readers queries to return most recent messages (#510)
* MF-506 - Fix fluxdb-reader query to return recent messages

Signed-off-by: Manuel Imperiale <manuel.imperiale@gmail.com>

* MF-506 - Fix mongodb-reader query to return messages sorted by descendent time

Signed-off-by: Manuel Imperiale <manuel.imperiale@gmail.com>

* Fix readers tests

Signed-off-by: Manuel Imperiale <manuel.imperiale@gmail.com>
2018-12-18 15:17:55 +01:00

136 lines
3.0 KiB
Go

package influxdb
import (
"encoding/json"
"fmt"
"reflect"
"strconv"
"strings"
"time"
"github.com/mainflux/mainflux/readers"
influxdata "github.com/influxdata/influxdb/client/v2"
"github.com/mainflux/mainflux"
)
const maxLimit = 100
var _ readers.MessageRepository = (*influxRepository)(nil)
type influxRepository struct {
database string
client influxdata.Client
}
// New returns new InfluxDB reader.
func New(client influxdata.Client, database string) (readers.MessageRepository, error) {
return &influxRepository{database, client}, nil
}
func (repo *influxRepository) ReadAll(chanID string, offset, limit uint64) []mainflux.Message {
if limit > maxLimit {
limit = maxLimit
}
cmd := fmt.Sprintf(`SELECT * from messages WHERE channel='%s' ORDER BY time DESC LIMIT %d OFFSET %d`, chanID, limit, offset)
q := influxdata.Query{
Command: cmd,
Database: repo.database,
}
ret := []mainflux.Message{}
resp, err := repo.client.Query(q)
if err != nil || resp.Error() != nil {
return ret
}
if len(resp.Results) < 1 || len(resp.Results[0].Series) < 1 {
return ret
}
result := resp.Results[0].Series[0]
for _, v := range result.Values {
ret = append(ret, parseMessage(result.Columns, v))
}
return ret
}
// ParseMessage and parseValues are util methods. Since InfluxDB client returns
// results in form of rows and columns, this obscure message conversion is needed
// to return actual []mainflux.Message from the query result.
func parseValues(value interface{}, name string, msg *mainflux.Message) {
if name == "valueSum" && value != nil {
if sum, ok := value.(json.Number); ok {
valSum, err := sum.Float64()
if err != nil {
return
}
msg.ValueSum = &mainflux.SumValue{Value: valSum}
}
return
}
if strings.HasSuffix(strings.ToLower(name), "value") {
switch value.(type) {
case bool:
msg.Value = &mainflux.Message_BoolValue{BoolValue: value.(bool)}
case json.Number:
num, err := value.(json.Number).Float64()
if err != nil {
return
}
msg.Value = &mainflux.Message_FloatValue{FloatValue: num}
case string:
if strings.HasPrefix(name, "string") {
msg.Value = &mainflux.Message_StringValue{StringValue: value.(string)}
return
}
if strings.HasPrefix(name, "data") {
msg.Value = &mainflux.Message_DataValue{DataValue: value.(string)}
}
}
}
}
func parseMessage(names []string, fields []interface{}) mainflux.Message {
m := mainflux.Message{}
v := reflect.ValueOf(&m).Elem()
for i, name := range names {
parseValues(fields[i], name, &m)
msgField := v.FieldByName(strings.Title(name))
if !msgField.IsValid() {
continue
}
f := msgField.Interface()
switch f.(type) {
case string:
if s, ok := fields[i].(string); ok {
msgField.SetString(s)
}
case float64:
if name == "time" {
t, err := time.Parse(time.RFC3339, fields[i].(string))
if err != nil {
continue
}
v := float64(t.Unix())
msgField.SetFloat(v)
continue
}
val, _ := strconv.ParseFloat(fields[i].(string), 64)
msgField.SetFloat(val)
}
}
return m
}