events_test.go

@@ -70,6 +70,16 @@ func TestEvents_Unwatch_Good_Idempotent(t *testing.T) {
 	storeInstance.Unwatch("g", events)
 }
 
+func TestEvents_Close_Good_ClosesWatcherChannels(t *testing.T) {
+	storeInstance, _ := New(":memory:")
+
+	events := storeInstance.Watch("g")
+	require.NoError(t, storeInstance.Close())
+
+	_, open := <-events
+	assert.False(t, open, "channel should be closed after Close")
+}
+
 func TestEvents_Unwatch_Good_NilChannel(t *testing.T) {
 	storeInstance, _ := New(":memory:")
 	defer storeInstance.Close()

store.go

@@ -226,6 +226,20 @@ func (storeInstance *Store) Close() error {
 		storeInstance.cancelPurge()
 	}
 	storeInstance.purgeWaitGroup.Wait()
+
+	storeInstance.watchersLock.Lock()
+	for groupName, registeredEvents := range storeInstance.watchers {
+		for _, registeredEventChannel := range registeredEvents {
+			close(registeredEventChannel)
+		}
+		delete(storeInstance.watchers, groupName)
+	}
+	storeInstance.watchersLock.Unlock()
+
+	storeInstance.callbacksLock.Lock()
+	storeInstance.callbacks = nil
+	storeInstance.callbacksLock.Unlock()
+
 	if storeInstance.database == nil {
 		return nil
 	}