Merge pull request #228 from emqx/uint64_overflow

fix for #223

fvt_scripts/edgex/pub.go

@@ -48,8 +48,9 @@ func pubEventClientZeroMq() {
 
 				r4 := models.Reading{Name:"i1", Value:fmt.Sprintf("%d", i)}
 				r5 := models.Reading{Name:"f1", Value:fmt.Sprintf("%.2f", float64(i)/2.0)}
+				r6 := models.Reading{Name:"ui64", Value:"10796529505058023104"}
 
-				testEvent.Readings = append(testEvent.Readings, r1, r2, r3, r4, r5)
+				testEvent.Readings = append(testEvent.Readings, r1, r2, r3, r4, r5, r6)
 
 				data, err := client.MarshalEvent(testEvent)
 				if err != nil {

fvt_scripts/edgex/valuedesc/vd_server.go

@@ -16,6 +16,7 @@ const (
 	desc4 = "Int descriptor"
 	desc5 = "Float descriptor"
 	desc6 = "String descriptor"
+	desc7 = "UInt64 descriptor"
 )
 
 var vd1 = models.ValueDescriptor{Id: "Temperature", Created: 123, Modified: 123, Origin: 123, Name: "Temperature",
@@ -30,6 +31,7 @@ var vd3 = models.ValueDescriptor{Id: "b1", Name: "b1", Formatting: "%t", Type:"B
 var vd4 = models.ValueDescriptor{Id: "i1", Name: "i1", Formatting: "%d", Type:"UINT8", MediaType: clients.ContentTypeJSON}
 var vd5 = models.ValueDescriptor{Id: "f1", Name: "f1", Formatting: "%f", Type:"FLOAT64", MediaType: clients.ContentTypeJSON}
 var vd6 = models.ValueDescriptor{Id: "s1", Name: "s1", Formatting: "%s", Type:"String", MediaType: clients.ContentTypeJSON}
+var vd7 = models.ValueDescriptor{Id: "ui64", Name: "ui64", Formatting: "%d", Type:"UINT64", MediaType: clients.ContentTypeJSON}
 
 func main() {
 	http.HandleFunc(clients.ApiValueDescriptorRoute, Hello)
@@ -57,7 +59,10 @@ func Hello(w http.ResponseWriter, req *http.Request) {
 	descriptor6 := vd6
 	descriptor6.Description = desc6
 
-	descriptors := []models.ValueDescriptor{descriptor1, descriptor2, descriptor3, descriptor4, descriptor5, descriptor6}
+	descriptor7 := vd7
+	descriptor7.Description = desc7
+
+	descriptors := []models.ValueDescriptor{descriptor1, descriptor2, descriptor3, descriptor4, descriptor5, descriptor6, descriptor7}
 
 	data, err := json.Marshal(descriptors)
 	if err != nil {

fvt_scripts/select_edgex_condition_rule.jmx

@@ -459,6 +459,15 @@
             <boolProp name="ISREGEX">false</boolProp>
           </JSONPathAssertion>
           <hashTree/>
+          <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="uint64 Assertion" enabled="true">
+            <stringProp name="JSON_PATH">$[0].ui64</stringProp>
+            <stringProp name="EXPECTED_VALUE">10796529505058023104</stringProp>
+            <boolProp name="JSONVALIDATION">true</boolProp>
+            <boolProp name="EXPECT_NULL">false</boolProp>
+            <boolProp name="INVERT">false</boolProp>
+            <boolProp name="ISREGEX">false</boolProp>
+          </JSONPathAssertion>
+          <hashTree/>
           <BeanShellAssertion guiclass="BeanShellAssertionGui" testclass="BeanShellAssertion" testname="temperature value assertion" enabled="true">
             <stringProp name="BeanShellAssertion.query">import net.sf.json.JSONArray;
 import net.sf.json.JSONObject;

xsql/plans/preprocessor.go

@@ -6,6 +6,7 @@ import (
 	"github.com/emqx/kuiper/common"
 	"github.com/emqx/kuiper/xsql"
 	"github.com/emqx/kuiper/xstream/api"
+	"math"
 	"reflect"
 	"strconv"
 	"strings"
@@ -112,13 +113,21 @@ func (p *Preprocessor) addRecField(ft xsql.FieldType, r map[string]interface{},
 				if jtype == reflect.Int {
 					r[n] = t.(int)
 				} else if jtype == reflect.Float64 {
-					r[n] = int(t.(float64))
+					if tt, ok1 := t.(float64); ok1 {
+						if tt > math.MaxInt64 {
+							r[n] = uint64(tt)
+						} else {
+							r[n] = int(tt)
+						}
+					}
 				} else if jtype == reflect.String {
 					if i, err := strconv.Atoi(t.(string)); err != nil {
 						return fmt.Errorf("invalid data type for %s, expect bigint but found %[2]T(%[2]v)", n, t)
 					} else {
 						r[n] = i
 					}
+				} else if jtype == reflect.Uint64 {
+					r[n] = t.(uint64)
 				} else {
 					return fmt.Errorf("invalid data type for %s, expect bigint but found %[2]T(%[2]v)", n, t)
 				}

xstream/extensions/edgex_source.go

@@ -182,12 +182,18 @@ func (es *EdgexSource) getValue(r models.Reading, logger api.Logger) (interface{
 		} else {
 			return r, nil
 		}
-	case "INT8", "INT16", "INT32", "INT64", "UINT8", "UINT16", "UINT32", "UINT64":
+	case "INT8", "INT16", "INT32", "INT64", "UINT8", "UINT16", "UINT32":
 		if r, err := strconv.Atoi(v); err != nil {
 			return nil, err
 		} else {
 			return r, nil
 		}
+	case "UINT64":
+		if u64, err := strconv.ParseUint(v, 10, 64); err != nil {
+			return nil, err
+		} else {
+			return u64, nil
+		}
 	case "FLOAT32", "FLOAT64":
 		if r, err := strconv.ParseFloat(v, 64); err != nil {
 			return nil, err

xstream/extensions/edgex_source_test.go

@@ -18,16 +18,16 @@ var es = EdgexSource{valueDescs: map[string]string{
 	"i5" : "UINT8",
 	"i6" : "UINT16",
 	"i7" : "UINT32",
-	"i8" : "UINT64",
 	"f1" : "FLOAT32",
 	"f2" : "FLOAT64",
 	"s1" : "String",
+	"i8" : "UINT64", //i8 will be handled by special case
 	},
 }
 
 func TestGetValue_Int(t *testing.T) {
 	var testEvent = models.Event{Device: "test"}
-	for i := 1; i < 9; i++{
+	for i := 1; i < 8; i++{
 		r1 := models.Reading{Name: fmt.Sprintf("i%d", i), Value: "1"}
 		testEvent.Readings = append(testEvent.Readings, r1)
 	}
@@ -39,6 +39,17 @@ func TestGetValue_Int(t *testing.T) {
 			expectOne(t, v)
 		}
 	}
+
+	r1 := models.Reading{Name: "i8", Value: "10796529505058023104"}
+	if v, e := es.getValue(r1, common.Log); e != nil {
+		t.Errorf("%s", e)
+	} else {
+		if v1, ok := v.(uint64); ok {
+			if v1 != 10796529505058023104 {
+				t.Errorf("expected 10796529505058023104, but it's %d.", v1)
+			}
+		}
+	}
 }
 
 func expectOne(t *testing.T, expected interface{}) {
@@ -47,7 +58,7 @@ func expectOne(t *testing.T, expected interface{}) {
 			t.Errorf("expected 1, but it's %d.", v1)
 		}
 	} else {
-		t.Errorf("expected int type, but it's %t.", expected)
+		t.Errorf("expected int type, but it's %T.", expected)
 	}
 }