Merge pull request #14 from bigbluedisco/master

Fix TimeWindow not resetting buckets when re-using them.

time.go

@@ -79,7 +79,11 @@ func (w *TimePolicy) AppendWithTimestamp(value float64, timestamp time.Time) {
 
 	var adjustedTime, windowOffset = w.selectBucket(timestamp)
 	w.keepConsistent(adjustedTime, windowOffset)
-	w.window[windowOffset] = append(w.window[windowOffset], value)
+	if w.lastWindowOffset != windowOffset {
+		w.window[windowOffset] = []float64{value}
+	} else {
+		w.window[windowOffset] = append(w.window[windowOffset], value)
+	}
 	w.lastWindowTime = adjustedTime
 	w.lastWindowOffset = windowOffset
 }

time_test.go

@@ -28,6 +28,24 @@ func TestTimeWindow(t *testing.T) {
 	if final != float64(numberBuckets) {
 		t.Fatal(final)
 	}
+
+	for x := 0; x < numberBuckets; x = x + 1 {
+		p.Append(2)
+		time.Sleep(bucketSize)
+	}
+
+	final = p.Reduce(func(w Window) float64 {
+		var result float64
+		for _, bucket := range w {
+			for _, point := range bucket {
+				result = result + point
+			}
+		}
+		return result
+	})
+	if final != 2*float64(numberBuckets) {
+		t.Fatal("got", final, "expected", 2*float64(numberBuckets))
+	}
 }
 
 func TestTimeWindowSelectBucket(t *testing.T) {