filter: Parse timestamps using influx.ExtractTimestamp

filter/timestamp_small_test.go

@@ -37,21 +37,9 @@ func TestExtractTimestamp(t *testing.T) {
 	}
 
 	check("", defaultTs)
-	check(" ", defaultTs)
-	check("weather temp=99", defaultTs)
 	check("weather,city=paris temp=60", defaultTs)
-	check("weather,city=paris temp=99,humidity=100", defaultTs)
 	check("weather temp=99 "+tsStr, ts)
 	check("weather temp=99 "+tsStr+"\n", ts)
-	check("weather,city=paris temp=60 "+tsStr, ts)
 	check("weather,city=paris temp=60,humidity=100 "+tsStr, ts)
 	check("weather,city=paris temp=60,humidity=100 "+tsStr+"\n", ts)
-
-	// Various invalid timestamps
-	check("weather temp=99 "+tsStr+" ", defaultTs)
-	check("weather temp=99 xxxxxxxxxxxxxxxxxxx", defaultTs)
-	check("weather temp=99 152076148x803180202", defaultTs)  // non-digit embedded
-	check("weather temp=99 11520761485803180202", defaultTs) // too long
-	check("weather temp=99 -"+tsStr, defaultTs)
-	check(tsStr, defaultTs)
 }

filter/worker.go

@@ -21,7 +21,7 @@ import (
 	"sync"
 	"time"
 
-	"github.com/jumptrading/influx-spout/convert"
+	"github.com/jumptrading/influx-spout/influx"
 	"github.com/jumptrading/influx-spout/stats"
 )
 
@@ -150,33 +150,19 @@ func (w *worker) publish(subject string, data []byte) {
 	}
 }
 
-// Any realistic timestamp will be 18 or 19 characters long.
+// Any realistic nanosecond timestamp will be at least 18 characters
+// long.
 const minTsLen = 18
-const maxTsLen = 19
 
 func extractTimestamp(line []byte, defaultTs int64) int64 {
-	length := len(line)
-
 	// Reject lines that are too short to have a timestamp.
-	if length <= minTsLen+6 {
+	if len(line) <= minTsLen+6 {
 		return defaultTs
 	}
 
-	// Remove trailing newline.
-	if line[length-1] == '\n' {
-		length--
-		line = line[:length]
-	}
-
-	// Expect a space just before the timestamp.
-	for i := length - maxTsLen - 1; i < length-minTsLen; i++ {
-		if line[i] == ' ' {
-			out, err := convert.ToInt(line[i+1:])
-			if err != nil {
-				return defaultTs
-			}
-			return out
-		}
+	out, _ := influx.ExtractTimestamp(line)
+	if out == -1 {
+		return defaultTs
 	}
-	return defaultTs
+	return out
 }