fix: session window inactivity timer reset mechanism (#102)

flow/session_window.go

@@ -12,9 +12,9 @@ import (
 type SessionWindow struct {
 	sync.Mutex
 	inactivityGap time.Duration
-	timer         *time.Timer
 	in            chan interface{}
 	out           chan interface{}
+	reset         chan struct{}
 	done          chan struct{}
 	buffer        []interface{}
 }
@@ -28,9 +28,9 @@ var _ streams.Flow = (*SessionWindow)(nil)
 func NewSessionWindow(inactivityGap time.Duration) *SessionWindow {
 	sessionWindow := &SessionWindow{
 		inactivityGap: inactivityGap,
-		timer:         time.NewTimer(inactivityGap),
 		in:            make(chan interface{}),
 		out:           make(chan interface{}),
+		reset:         make(chan struct{}),
 		done:          make(chan struct{}),
 	}
 	go sessionWindow.emit()
@@ -74,24 +74,41 @@ func (sw *SessionWindow) receive() {
 	for element := range sw.in {
 		sw.Lock()
 		sw.buffer = append(sw.buffer, element)
-		sw.timer.Reset(sw.inactivityGap) // reset the inactivity timer
 		sw.Unlock()
+		sw.notifyTimerReset() // signal to reset the inactivity timer
 	}
 	close(sw.done)
 }
 
+// notifyTimerReset sends a notification to reset the inactivity timer.
+func (sw *SessionWindow) notifyTimerReset() {
+	select {
+	case sw.reset <- struct{}{}:
+	default:
+	}
+}
+
 // emit captures and emits a session window based on the gap of inactivity.
 // When this period expires, the current session closes and subsequent elements
 // are assigned to a new session window.
 func (sw *SessionWindow) emit() {
-	defer sw.timer.Stop()
-
+	timer := time.NewTimer(sw.inactivityGap)
 	for {
 		select {
-		case <-sw.timer.C:
+		case <-timer.C:
 			sw.dispatchWindow()
 
+		case <-sw.reset:
+			if !timer.Stop() {
+				select {
+				case <-timer.C:
+				default:
+				}
+			}
+			timer.Reset(sw.inactivityGap)
+
 		case <-sw.done:
+			timer.Stop()
 			sw.dispatchWindow()
 			close(sw.out)
 			return

flow/session_window_test.go

@@ -41,3 +41,38 @@ func TestSessionWindow(t *testing.T) {
 	assertEquals(t, []interface{}{"d"}, outputValues[1])
 	assertEquals(t, []interface{}{"e"}, outputValues[2])
 }
+
+func TestLongSessionWindow(t *testing.T) {
+	in := make(chan interface{})
+	out := make(chan interface{})
+
+	source := ext.NewChanSource(in)
+	sessionWindow := flow.NewSessionWindow(20 * time.Millisecond)
+	sink := ext.NewChanSink(out)
+
+	inputValues := []string{"a", "b", "c", "d", "e", "f", "g"}
+	go func() {
+		for _, e := range inputValues {
+			ingestDeferred(e, in, 10*time.Millisecond)
+		}
+	}()
+	go ingestDeferred("h", in, 140*time.Millisecond)
+	go closeDeferred(in, 150*time.Millisecond)
+
+	go func() {
+		source.
+			Via(sessionWindow).
+			To(sink)
+	}()
+
+	var outputValues [][]interface{}
+	for e := range sink.Out {
+		outputValues = append(outputValues, e.([]interface{}))
+	}
+	fmt.Println(outputValues)
+
+	assertEquals(t, 2, len(outputValues)) // [[a b c d e f g] [h]]
+
+	assertEquals(t, []interface{}{"a", "b", "c", "d", "e", "f", "g"}, outputValues[0])
+	assertEquals(t, []interface{}{"h"}, outputValues[1])
+}