feat(config): support configuration ReloadedEvent

config/config.go

@@ -8,17 +8,19 @@ import (
 	"sync"
 
 	"github.com/DoNewsCode/core/contract"
+	"github.com/DoNewsCode/core/events"
 	"github.com/knadh/koanf"
 	"github.com/mitchellh/mapstructure"
 )
 
 // KoanfAdapter is a implementation of contract.Config based on Koanf (https://github.com/knadh/koanf).
 type KoanfAdapter struct {
-	layers    []ProviderSet
-	watcher   contract.ConfigWatcher
-	delimiter string
-	rwlock    sync.RWMutex
-	K         *koanf.Koanf
+	layers     []ProviderSet
+	watcher    contract.ConfigWatcher
+	dispatcher contract.Dispatcher
+	delimiter  string
+	rwlock     sync.RWMutex
+	K          *koanf.Koanf
 }
 
 // ProviderSet is a configuration layer formed by a parser and a provider.
@@ -53,6 +55,13 @@ func WithDelimiter(delimiter string) Option {
 	}
 }
 
+// WithDispatcher changes the default dispatcher of Koanf.
+func WithDispatcher(dispatcher contract.Dispatcher) Option {
+	return func(option *KoanfAdapter) {
+		option.dispatcher = dispatcher
+	}
+}
+
 // NewConfig creates a new *KoanfAdapter.
 func NewConfig(options ...Option) (*KoanfAdapter, error) {
 	adapter := KoanfAdapter{delimiter: "."}
@@ -74,6 +83,10 @@ func NewConfig(options ...Option) (*KoanfAdapter, error) {
 // an error occurred, Reload will return early and abort the rest of the
 // reloading.
 func (k *KoanfAdapter) Reload() error {
+	if k.dispatcher != nil {
+		defer k.dispatcher.Dispatch(context.Background(), events.Of(ReloadedEvent{k}))
+	}
+
 	k.rwlock.Lock()
 	defer k.rwlock.Unlock()
 

config/event.go

@@ -0,0 +1,11 @@
+package config
+
+import (
+	"github.com/DoNewsCode/core/contract"
+)
+
+// ReloadedEvent is an event that triggers the configuration reloads
+type ReloadedEvent struct {
+	// NewConf is the latest configuration after the reload.
+	NewConf contract.ConfigAccessor
+}

config/module.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+
 	"github.com/DoNewsCode/core/di"
 	"gopkg.in/yaml.v3"
 
@@ -18,17 +19,20 @@ import (
 )
 
 // Module is the configuration module that bundles the reload watcher and exportConfig commands.
+// This module triggers ReloadedEvent on configuration change.
 type Module struct {
 	conf            *KoanfAdapter
 	exportedConfigs []ExportedConfig
+	dispatcher      contract.Dispatcher
 }
 
 // ConfigIn is the injection parameter for config.New.
 type ConfigIn struct {
 	di.In
 
 	Conf            contract.ConfigAccessor
-	ExportedConfigs []ExportedConfig `group:"config"`
+	Dispatcher      contract.Dispatcher `optional:"true"`
+	ExportedConfigs []ExportedConfig    `group:"config"`
 }
 
 // New creates a new config module. It contains the init command.
@@ -41,6 +45,7 @@ func New(p ConfigIn) (Module, error) {
 		return Module{}, fmt.Errorf("expects a *config.KoanfAdapter instance, but %T given", p.Conf)
 	}
 	return Module{
+		dispatcher:      p.Dispatcher,
 		conf:            adapter,
 		exportedConfigs: p.ExportedConfigs,
 	}, nil
@@ -50,6 +55,7 @@ func New(p ConfigIn) (Module, error) {
 func (m Module) ProvideRunGroup(group *run.Group) {
 	ctx, cancel := context.WithCancel(context.Background())
 	group.Add(func() error {
+		m.conf.dispatcher = m.dispatcher
 		return m.conf.Watch(ctx)
 	}, func(err error) {
 		cancel()

config/module_test.go

@@ -1,12 +1,17 @@
 package config
 
 import (
+	"context"
 	"io/ioutil"
 	"os"
 	"runtime"
 	"strings"
 	"testing"
 
+	"github.com/DoNewsCode/core/contract"
+	"github.com/DoNewsCode/core/events"
+	"github.com/knadh/koanf/providers/confmap"
+	"github.com/oklog/run"
 	"github.com/spf13/cobra"
 	"github.com/stretchr/testify/assert"
 )
@@ -97,3 +102,52 @@ func TestModule_ProvideCommand(t *testing.T) {
 		})
 	}
 }
+
+func TestModule_Watch(t *testing.T) {
+	t.Run("test without module", func(t *testing.T) {
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		dispatcher := &events.SyncDispatcher{}
+		dispatcher.Subscribe(events.Listen(events.From(ReloadedEvent{}), func(ctx context.Context, event contract.Event) error {
+			data := event.Data().(ReloadedEvent).NewConf.(*KoanfAdapter)
+			assert.Equal(t, "bar", data.String("foo"))
+			cancel()
+			return nil
+		}))
+		conf, _ := NewConfig(WithDispatcher(dispatcher), WithProviderLayer(confmap.Provider(map[string]interface{}{"foo": "bar"}, "."), nil))
+		conf.Watch(ctx)
+	})
+
+	t.Run("test with module", func(t *testing.T) {
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		dispatcher := &events.SyncDispatcher{}
+		dispatcher.Subscribe(events.Listen(events.From(ReloadedEvent{}), func(ctx context.Context, event contract.Event) error {
+			data := event.Data().(ReloadedEvent).NewConf.(*KoanfAdapter)
+			assert.Equal(t, "bar", data.String("foo"))
+			cancel()
+			return nil
+		}))
+
+		conf, _ := NewConfig(WithProviderLayer(confmap.Provider(map[string]interface{}{"foo": "bar"}, "."), nil), WithWatcher(&MockWatcher{}))
+		module, _ := New(ConfigIn{
+			Conf:       conf,
+			Dispatcher: dispatcher,
+		})
+		var g run.Group
+		g.Add(func() error {
+			<-ctx.Done()
+			return nil
+		}, func(err error) {})
+		module.ProvideRunGroup(&g)
+		g.Run()
+	})
+}
+
+type MockWatcher struct{}
+
+func (m *MockWatcher) Watch(ctx context.Context, reload func() error) error {
+	return reload()
+}