feat: glue all changes

worksinmagic · Sep 25, 2020 · 02d48f7 · 02d48f7
1 parent 1aba1df
commit 02d48f7
Show file tree

Hide file tree

Showing 3 changed files with 18 additions and 18 deletions.
diff --git a/app/ytfeed/ytfeed.go b/app/ytfeed/ytfeed.go
@@ -46,7 +46,7 @@ func Run(ctx context.Context, logger mainytfeed.Logger) (err error) {
 
 	var streamScheduler *streamschedule.StreamSchedule
 	if cfg.BoltDBPath != "" {
-		streamScheduler, err = streamschedule.New(logger, cfg.BoltDBPath, cfg.ResubCallbackAddr, cfg.StreamSchedulerRetryDelay, cfg.StreamSchedulerWorkerInterval, cfg.StreamSchedulerMaxRetries)
+		streamScheduler, err = streamschedule.New(logger, cfg.BoltDBPath, cfg.StreamSchedulerWorkerInterval)
 		if err != nil {
 			err = errors.Wrap(err, "failed to create stream scheduler service")
 			return
@@ -178,7 +178,7 @@ func Run(ctx context.Context, logger mainytfeed.Logger) (err error) {
 	}
 
 	// run workers
-	subscriber := autosubscribefeed.New(logger, cfg.VerificationToken, cfg.ResubTargetAddr, cfg.ResubTopic, cfg.ResubCallbackAddr, cfg.ResubInterval)
+	subscriber := autosubscribefeed.New(logger, cfg.VerificationToken, cfg.VerificationSecret, cfg.ResubTargetAddr, cfg.ResubCallbackAddr, cfg.ResubTopics, cfg.ResubInterval)
 	go func(ctx context.Context, subscriber *autosubscribefeed.Subscriber) {
 		err := subscriber.Subscribe(ctx)
 		if err != nil {
@@ -190,6 +190,7 @@ func Run(ctx context.Context, logger mainytfeed.Logger) (err error) {
 
 	if streamScheduler != nil {
 		go func(ctx context.Context, streamScheduler *streamschedule.StreamSchedule) {
+			streamScheduler.RegisterDataHandler(dataHandlers...)
 			err := streamScheduler.RunWorker(ctx)
 			if err != nil {
 				err = errors.Wrap(err, "stream scheduler worker exited with error")
@@ -201,7 +202,7 @@ func Run(ctx context.Context, logger mainytfeed.Logger) (err error) {
 
 	// declare handler functions
 	http.HandleFunc("/health", health.Handler)
-	http.HandleFunc("/", rss.Handler(ctx, logger, cfg.VerificationToken, dataHandlers...))
+	http.HandleFunc("/", rss.Handler(ctx, logger, cfg.VerificationToken, cfg.VerificationSecret, dataHandlers...))
 
 	// listen
 	errCh := make(chan error, 1)

diff --git a/config/config.go b/config/config.go
@@ -10,16 +10,14 @@ import (
 )
 
 const (
-	DefaultVersion                       = "v1.0.0"
+	DefaultVersion                       = "v0.2.0"
 	DefaultHost                          = ":8123"
 	DefaultResubTargetAddr               = "https://pubsubhubbub.appspot.com/subscribe"
 	DefaultResubInterval                 = 24 * 3 * time.Hour
 	DefaultFileNameTemplate              = "{{.ChannelID}}/{{.PublishedYear}}/{{.PublishedMonth}}/{{.PublishedDay}}/{{.PublishedTimeZone}}/{{.VideoID}}.{{.VideoExtension}}"
 	DefaultFormatQuality                 = "720"
 	DefaultFormatExtension               = "webm"
 	DefaultRedisChannel                  = "ytfeed"
-	DefaultStreamSchedulerRetryDelay     = 1 * time.Minute
-	DefaultStreamSchedulerMaxRetries     = 5
 	DefaultStreamSchedulerWorkerInterval = 1 * time.Minute
 	DefaultVideoDownloadMaxRetries       = 5
 	DefaultTemporaryFileDir              = "./"
@@ -55,6 +53,7 @@ func init() {
 	handleError(viper.BindEnv("host"))
 	handleError(viper.BindEnv("youtube_api_key"))
 	handleError(viper.BindEnv("verification_token"))
+	handleError(viper.BindEnv("verification_secret"))
 
 	handleError(viper.BindEnv("resub_interval"))
 	handleError(viper.BindEnv("resub_target_addr"))
@@ -99,9 +98,7 @@ func init() {
 	handleError(viper.BindEnv("redis_idle_check_frequency"))
 
 	handleError(viper.BindEnv("boltdb_path"))
-	handleError(viper.BindEnv("stream_scheduler_retry_delay"))
 	handleError(viper.BindEnv("stream_scheduler_worker_interval"))
-	handleError(viper.BindEnv("stream_scheduler_max_retries"))
 
 	handleError(viper.BindEnv("amqp_dsn"))
 	handleError(viper.BindEnv("amqp_exchange"))
@@ -122,9 +119,7 @@ func init() {
 	viper.SetDefault("video_format_quality", DefaultFormatQuality)
 	viper.SetDefault("video_format_extension", DefaultFormatExtension)
 	viper.SetDefault("redis_channel", DefaultRedisChannel)
-	viper.SetDefault("stream_scheduler_retry_delay", DefaultStreamSchedulerRetryDelay)
 	viper.SetDefault("stream_scheduler_worker_interval", DefaultStreamSchedulerWorkerInterval)
-	viper.SetDefault("stream_scheduler_max_retries", DefaultStreamSchedulerMaxRetries)
 	viper.SetDefault("video_download_max_retries", DefaultVideoDownloadMaxRetries)
 	viper.SetDefault("temporary_file_dir", DefaultTemporaryFileDir)
 	viper.SetDefault("amqp_exchange", DefaultAMQPExchange)
@@ -151,11 +146,12 @@ type Validator interface {
 type Configuration struct {
 	validator Validator `validate:"required"`
 
-	YoutubeAPIKey     string `validate:"required"`
-	VerificationToken string `validate:"required"`
+	YoutubeAPIKey      string `validate:"required"`
+	VerificationToken  string `validate:"required"`
+	VerificationSecret string `validate:"required"`
 
 	ResubTargetAddr   string        `validate:"required"`
-	ResubTopic        string        `validate:"required"`
+	ResubTopics       []string      `validate:"required"`
 	ResubCallbackAddr string        `validate:"required"`
 	ResubInterval     time.Duration `validate:"required"`
 
@@ -197,9 +193,7 @@ type Configuration struct {
 	RedisIdleCheckFrequency time.Duration `validate:""`
 
 	BoltDBPath                    string        `validate:"omitempty,file"`
-	StreamSchedulerRetryDelay     time.Duration `validate:"required,min=0"`
 	StreamSchedulerWorkerInterval time.Duration `validate:"required,min=1000000000"`
-	StreamSchedulerMaxRetries     int           `validate:""`
 
 	AMQPDSN                string `validate:""`
 	AMQPExchange           string `validate:"required"`
@@ -219,10 +213,11 @@ func New() (c *Configuration) {
 
 	c.YoutubeAPIKey = viper.GetString("youtube_api_key")
 	c.VerificationToken = viper.GetString("verification_token")
+	c.VerificationSecret = viper.GetString("verification_secret")
 
 	c.ResubCallbackAddr = viper.GetString("resub_callback_addr")
 	c.ResubTargetAddr = viper.GetString("resub_target_addr")
-	c.ResubTopic = viper.GetString("resub_topic")
+	c.ResubTopics = viper.GetStringSlice("resub_topic")
 	c.ResubInterval = viper.GetDuration("resub_interval")
 
 	c.S3Endpoint = viper.GetString("s3_endpoint")
@@ -263,9 +258,7 @@ func New() (c *Configuration) {
 	c.RedisIdleCheckFrequency = viper.GetDuration("redis_idle_check_frequency")
 
 	c.BoltDBPath = viper.GetString("boltdb_path")
-	c.StreamSchedulerRetryDelay = viper.GetDuration("stream_scheduler_retry_delay")
 	c.StreamSchedulerWorkerInterval = viper.GetDuration("stream_scheduler_worker_interval")
-	c.StreamSchedulerMaxRetries = viper.GetInt("stream_scheduler_max_retries")
 
 	c.AMQPDSN = viper.GetString("amqp_dsn")
 	c.AMQPExchange = viper.GetString("amqp_exchange")

diff --git a/config/config_test.go b/config/config_test.go
@@ -2,6 +2,7 @@ package config
 
 import (
 	"errors"
+	"os"
 	"testing"
 
 	"github.com/golang/mock/gomock"
@@ -18,6 +19,9 @@ func TestConfig(t *testing.T) {
 	mockValidator := mock.NewMockValidator(ctrl)
 
 	t.Run("Validate success", func(t *testing.T) {
+		version := "v100.0.0"
+		os.Setenv("YTFEED_VERSION", version)
+
 		cfg = New()
 		require.NotNil(t, cfg)
 		cfg.StorageBackend = StorageBackendDisk
@@ -28,6 +32,8 @@ func TestConfig(t *testing.T) {
 
 		err := cfg.Validate()
 		require.NoError(t, err)
+		require.Equal(t, version, cfg.Version)
+		require.Equal(t, DefaultHost, cfg.Host)
 	})
 
 	t.Run("Validate failed", func(t *testing.T) {