Function TestModule_hotReload

otkafka/integration_test.go:291–379

(t *testing.T)

}

func TestModule_hotReload(t *testing.T) {
	if os.Getenv("KAFKA_ADDR") == "" {
		t.Skip("set KAFKA_ADDR to run TestModule_ProvideRunGroup")
		return
	}
	addrs := strings.Split(os.Getenv("KAFKA_ADDR"), ",")
	t.Parallel()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	cw := &channelWatcher{}
	cw.ch = make(chan struct{})
	cw.afterReload = make(chan struct{})

	conf := map[string]interface{}{
		"http": map[string]bool{
			"disable": true,
		},
		"grpc": map[string]bool{
			"disable": true,
		},
		"cron": map[string]bool{
			"disable": true,
		},
		"kafka": map[string]interface{}{
			"reader": map[string]interface{}{
				"default": map[string]interface{}{
					"brokers": addrs,
					"topic":   "foo",
				},
			},
			"writer": map[string]interface{}{
				"default": map[string]interface{}{
					"brokers": addrs,
					"topic":   "foo",
				},
			},
		},
	}
	path := createFile(conf)
	c := core.Default(core.WithConfigStack(file.Provider(path), knoaf_json.Parser()), core.WithConfigWatcher(cw))
	c.Provide(Providers(WithReaderReload(true), WithWriterReload(true)))
	c.AddModuleFunc(config.New)

	var group run.Group
	for _, m := range c.Modules() {
		if p, ok := m.(core.RunProvider); ok {
			p.ProvideRunGroup(&group)
		}
	}
	group.Add(func() error {
		<-ctx.Done()
		return ctx.Err()
	}, func(err error) {
		cancel()
	})
	go group.Run()