refactor(BUX-411): redis options by WithRedis

bux_suite_mocks_test.go

@@ -12,13 +12,13 @@ type taskManagerMockBase struct{}
 
 func (tm *taskManagerMockBase) Info(context.Context, string, ...interface{}) {}
 
-func (tm *taskManagerMockBase) RegisterTask(*taskmanager.Task) error {
+func (tm *taskManagerMockBase) RegisterTask(string, interface{}) error {
 	return nil
 }
 
 func (tm *taskManagerMockBase) ResetCron() {}
 
-func (tm *taskManagerMockBase) RunTask(context.Context, *taskmanager.TaskOptions) error {
+func (tm *taskManagerMockBase) RunTask(context.Context, *taskmanager.TaskRunOptions) error {
 	return nil
 }
 
@@ -55,6 +55,6 @@ func (tm *taskManagerMockBase) CronJobsInit(cronJobsMap taskmanager.CronJobs) er
 // Sets custom task manager only for testing
 func withTaskManagerMockup() ClientOps {
 	return func(c *clientOptions) {
-		c.taskManager.TaskManagerInterface = &taskManagerMockBase{}
+		c.taskManager.Tasker = &taskManagerMockBase{}
 	}
 }

bux_suite_test.go

@@ -326,7 +326,7 @@ func (ts *EmbeddedDBTestSuite) genericDBClient(t *testing.T, database datastore.
 		WithAutoMigrate(&PaymailAddress{}),
 	)
 	if taskManagerEnabled {
-		opts = append(opts, WithTaskqConfig(taskmanager.DefaultTaskQConfig(prefix+"_queue", nil)))
+		opts = append(opts, WithTaskqConfig(taskmanager.DefaultTaskQConfig(prefix+"_queue")))
 	} else {
 		opts = append(opts, withTaskManagerMockup())
 	}

bux_test.go

@@ -59,7 +59,7 @@ func (tc *TestingClient) Close(ctx context.Context) {
 
 // DefaultClientOpts will return a default set of client options required to load the new client
 func DefaultClientOpts(debug, shared bool) []ClientOps {
-	tqc := taskmanager.DefaultTaskQConfig(tester.RandomTablePrefix(), nil)
+	tqc := taskmanager.DefaultTaskQConfig(tester.RandomTablePrefix())
 	tqc.MaxNumWorker = 2
 	tqc.MaxNumFetcher = 2
 

client.go

@@ -113,10 +113,10 @@ type (
 
 	// taskManagerOptions holds the configuration for taskmanager
 	taskManagerOptions struct {
-		taskmanager.TaskManagerInterface                          // Client for TaskManager
-		cronJobs                         taskmanager.CronJobs     // List of cron jobs
-		options                          []taskmanager.ClientOps  // List of options
-		cronCustomPeriods                map[string]time.Duration // will override the default period of cronJob
+		taskmanager.Tasker                          // Client for TaskManager
+		cronJobs           taskmanager.CronJobs     // List of cron jobs
+		options            []taskmanager.ClientOps  // List of options
+		cronCustomPeriods  map[string]time.Duration // will override the default period of cronJob
 	}
 )
 
@@ -303,7 +303,7 @@ func (c *Client) Close(ctx context.Context) error {
 		if err := tm.Close(ctx); err != nil {
 			return err
 		}
-		c.options.taskManager.TaskManagerInterface = nil
+		c.options.taskManager.Tasker = nil
 	}
 	return nil
 }
@@ -445,9 +445,9 @@ func (c *Client) SetNotificationsClient(client notifications.ClientInterface) {
 }
 
 // Taskmanager will return the Taskmanager if it exists
-func (c *Client) Taskmanager() taskmanager.TaskManagerInterface {
-	if c.options.taskManager != nil && c.options.taskManager.TaskManagerInterface != nil {
-		return c.options.taskManager.TaskManagerInterface
+func (c *Client) Taskmanager() taskmanager.Tasker {
+	if c.options.taskManager != nil && c.options.taskManager.Tasker != nil {
+		return c.options.taskManager.Tasker
 	}
 	return nil
 }

client_internal.go

@@ -114,8 +114,8 @@ func (c *Client) loadPaymailClient() (err error) {
 // loadTaskmanager will load the TaskManager and start the TaskManager client
 func (c *Client) loadTaskmanager(ctx context.Context) (err error) {
 	// Load if a custom interface was NOT provided
-	if c.options.taskManager.TaskManagerInterface == nil {
-		c.options.taskManager.TaskManagerInterface, err = taskmanager.NewClient(
+	if c.options.taskManager.Tasker == nil {
+		c.options.taskManager.Tasker, err = taskmanager.NewClient(
 			ctx, c.options.taskManager.options...,
 		)
 	}

client_options.go

@@ -105,8 +105,8 @@ func defaultClientOptions() *clientOptions {
 
 		// Blank TaskManager config
 		taskManager: &taskManagerOptions{
-			TaskManagerInterface: nil,
-			cronCustomPeriods:    map[string]time.Duration{},
+			Tasker:            nil,
+			cronCustomPeriods: map[string]time.Duration{},
 		},
 
 		// Default user agent

client_options_test.go

@@ -264,7 +264,7 @@ func TestWithRedis(t *testing.T) {
 
 		tc, err := NewClient(
 			tester.GetNewRelicCtx(t, defaultNewRelicApp, defaultNewRelicTx),
-			WithTaskqConfig(taskmanager.DefaultTaskQConfig(tester.RandomTablePrefix(), nil)),
+			WithTaskqConfig(taskmanager.DefaultTaskQConfig(tester.RandomTablePrefix())),
 			WithRedis(&cachestore.RedisConfig{
 				URL: cachestore.RedisPrefix + "localhost:6379",
 			}),
@@ -287,7 +287,7 @@ func TestWithRedis(t *testing.T) {
 
 		tc, err := NewClient(
 			tester.GetNewRelicCtx(t, defaultNewRelicApp, defaultNewRelicTx),
-			WithTaskqConfig(taskmanager.DefaultTaskQConfig(tester.RandomTablePrefix(), nil)),
+			WithTaskqConfig(taskmanager.DefaultTaskQConfig(tester.RandomTablePrefix())),
 			WithRedis(&cachestore.RedisConfig{
 				URL: "localhost:6379",
 			}),
@@ -314,7 +314,7 @@ func TestWithRedisConnection(t *testing.T) {
 	t.Run("using a nil connection", func(t *testing.T) {
 		tc, err := NewClient(
 			tester.GetNewRelicCtx(t, defaultNewRelicApp, defaultNewRelicTx),
-			WithTaskqConfig(taskmanager.DefaultTaskQConfig(tester.RandomTablePrefix(), nil)),
+			WithTaskqConfig(taskmanager.DefaultTaskQConfig(tester.RandomTablePrefix())),
 			WithRedisConnection(nil),
 			WithSQLite(tester.SQLiteTestConfig(false, true)),
 			WithMinercraft(&chainstate.MinerCraftBase{}),
@@ -335,7 +335,7 @@ func TestWithRedisConnection(t *testing.T) {
 
 		tc, err := NewClient(
 			tester.GetNewRelicCtx(t, defaultNewRelicApp, defaultNewRelicTx),
-			WithTaskqConfig(taskmanager.DefaultTaskQConfig(tester.RandomTablePrefix(), nil)),
+			WithTaskqConfig(taskmanager.DefaultTaskQConfig(tester.RandomTablePrefix())),
 			WithRedisConnection(client),
 			WithSQLite(tester.SQLiteTestConfig(false, true)),
 			WithMinercraft(&chainstate.MinerCraftBase{}),
@@ -363,7 +363,7 @@ func TestWithFreeCache(t *testing.T) {
 		tc, err := NewClient(
 			tester.GetNewRelicCtx(t, defaultNewRelicApp, defaultNewRelicTx),
 			WithFreeCache(),
-			WithTaskqConfig(taskmanager.DefaultTaskQConfig(testQueueName, nil)),
+			WithTaskqConfig(taskmanager.DefaultTaskQConfig(testQueueName)),
 			WithSQLite(&datastore.SQLiteConfig{Shared: true}),
 			WithMinercraft(&chainstate.MinerCraftBase{}))
 		require.NoError(t, err)
@@ -391,7 +391,7 @@ func TestWithFreeCacheConnection(t *testing.T) {
 		tc, err := NewClient(
 			tester.GetNewRelicCtx(t, defaultNewRelicApp, defaultNewRelicTx),
 			WithFreeCacheConnection(nil),
-			WithTaskqConfig(taskmanager.DefaultTaskQConfig(testQueueName, nil)),
+			WithTaskqConfig(taskmanager.DefaultTaskQConfig(testQueueName)),
 			WithSQLite(&datastore.SQLiteConfig{Shared: true}),
 			WithMinercraft(&chainstate.MinerCraftBase{}),
 			WithLogger(&logger),
@@ -412,7 +412,7 @@ func TestWithFreeCacheConnection(t *testing.T) {
 		tc, err := NewClient(
 			tester.GetNewRelicCtx(t, defaultNewRelicApp, defaultNewRelicTx),
 			WithFreeCacheConnection(fc),
-			WithTaskqConfig(taskmanager.DefaultTaskQConfig(testQueueName, nil)),
+			WithTaskqConfig(taskmanager.DefaultTaskQConfig(testQueueName)),
 			WithSQLite(&datastore.SQLiteConfig{Shared: true}),
 			WithMinercraft(&chainstate.MinerCraftBase{}),
 			WithLogger(&logger),
@@ -497,9 +497,7 @@ func TestWithTaskQ(t *testing.T) {
 		tc, err := NewClient(
 			tester.GetNewRelicCtx(t, defaultNewRelicApp, defaultNewRelicTx),
 			WithTaskqConfig(
-				taskmanager.DefaultTaskQConfig(tester.RandomTablePrefix(), &taskmanager.SimplifiedRedisOptions{
-					Addr: "localhost:6379",
-				}),
+				taskmanager.DefaultTaskQConfig(tester.RandomTablePrefix(), taskmanager.WithRedis("localhost:6379")),
 			),
 			WithRedis(&cachestore.RedisConfig{
 				URL: cachestore.RedisPrefix + "localhost:6379",

examples/client/redis/redis.go

@@ -15,9 +15,8 @@ func main() {
 		context.Background(), // Set context
 		bux.WithRedis(&cachestore.RedisConfig{URL: redisURL}), // Cache
 		bux.WithTaskqConfig( // Tasks
-			taskmanager.DefaultTaskQConfig("example_queue", &taskmanager.SimplifiedRedisOptions{
-				Addr: redisURL,
-			})),
+			taskmanager.DefaultTaskQConfig("example_queue", taskmanager.WithRedis(redisURL)),
+		),
 	)
 	if err != nil {
 		log.Fatalln("error: " + err.Error())

interface.go

@@ -66,7 +66,7 @@ type ClientService interface {
 	Logger() *zerolog.Logger
 	Notifications() notifications.ClientInterface
 	PaymailClient() paymail.ClientInterface
-	Taskmanager() taskmanager.TaskManagerInterface
+	Taskmanager() taskmanager.Tasker
 }
 
 // DestinationService is the destination actions

paymail_test.go

@@ -219,7 +219,7 @@ func Test_getCapabilities(t *testing.T) {
 
 		tc, err := NewClient(context.Background(),
 			WithRedisConnection(redisClient),
-			WithTaskqConfig(taskmanager.DefaultTaskQConfig(testQueueName, nil)),
+			WithTaskqConfig(taskmanager.DefaultTaskQConfig(testQueueName)),
 			WithSQLite(&datastore.SQLiteConfig{Shared: true}),
 			WithChainstateOptions(false, false, false, false),
 			WithDebugging(),
@@ -261,7 +261,7 @@ func Test_getCapabilities(t *testing.T) {
 
 		tc, err := NewClient(context.Background(),
 			WithRedisConnection(redisClient),
-			WithTaskqConfig(taskmanager.DefaultTaskQConfig(testQueueName, nil)),
+			WithTaskqConfig(taskmanager.DefaultTaskQConfig(testQueueName)),
 			WithSQLite(&datastore.SQLiteConfig{Shared: true}),
 			WithChainstateOptions(false, false, false, false),
 			WithDebugging(),
@@ -374,7 +374,7 @@ func Test_resolvePaymailAddress(t *testing.T) {
 
 		tc, err := NewClient(context.Background(),
 			WithRedisConnection(redisClient),
-			WithTaskqConfig(taskmanager.DefaultTaskQConfig(testQueueName, nil)),
+			WithTaskqConfig(taskmanager.DefaultTaskQConfig(testQueueName)),
 			WithSQLite(&datastore.SQLiteConfig{Shared: true}),
 			WithChainstateOptions(false, false, false, false),
 			WithDebugging(),

taskmanager/cron_jobs.go

@@ -31,23 +31,19 @@ func (tm *Client) CronJobsInit(cronJobsMap CronJobs) (err error) {
 
 	for name, taskDef := range cronJobsMap {
 		handler := taskDef.Handler
-		if err = tm.RegisterTask(&Task{
-			Name:       name,
-			RetryLimit: 1,
-			Handler: func() error {
-				if taskErr := handler(ctx); taskErr != nil {
-					if tm.options.logger != nil {
-						tm.options.logger.Error().Msgf("error running %v task: %v", name, taskErr.Error())
-					}
+		if err = tm.RegisterTask(name, func() error {
+			if taskErr := handler(ctx); taskErr != nil {
+				if tm.options.logger != nil {
+					tm.options.logger.Error().Msgf("error running %v task: %v", name, taskErr.Error())
 				}
-				return nil
-			},
+			}
+			return nil
 		}); err != nil {
 			return
 		}
 
 		// Run the task periodically
-		if err = tm.RunTask(ctx, &TaskOptions{
+		if err = tm.RunTask(ctx, &TaskRunOptions{
 			RunEveryPeriod: taskDef.Period,
 			TaskName:       name,
 		}); err != nil {

taskmanager/interface.go

@@ -6,11 +6,11 @@ import (
 	taskq "github.com/vmihailenco/taskq/v3"
 )
 
-// TaskManagerInterface is the taskmanager client interface
-type TaskManagerInterface interface {
-	RegisterTask(task *Task) error
+// Tasker is the taskmanager client interface
+type Tasker interface {
+	RegisterTask(name string, handler interface{}) error
 	ResetCron()
-	RunTask(ctx context.Context, options *TaskOptions) error
+	RunTask(ctx context.Context, options *TaskRunOptions) error
 	Tasks() map[string]*taskq.Task
 	CronJobsInit(cronJobsMap CronJobs) error
 	Close(ctx context.Context) error

taskmanager/options.go

@@ -22,7 +22,7 @@ func defaultClientOptions() *clientOptions {
 		newRelicEnabled: false,
 		taskq: &taskqOptions{
 			tasks:  make(map[string]*taskq.Task),
-			config: DefaultTaskQConfig("taskq", nil),
+			config: DefaultTaskQConfig("taskq"),
 		},
 	}
 }
@@ -52,6 +52,7 @@ func WithDebugging() ClientOps {
 	}
 }
 
+// WithTaskqConfig will set the taskq custom config
 func WithTaskqConfig(config *taskq.QueueOptions) ClientOps {
 	return func(c *clientOptions) {
 		if config != nil {

taskmanager/options_test.go

@@ -61,7 +61,7 @@ func TestWithTaskQ(t *testing.T) {
 		options := &clientOptions{
 			taskq: &taskqOptions{},
 		}
-		opt := WithTaskqConfig(DefaultTaskQConfig(testQueueName, nil))
+		opt := WithTaskqConfig(DefaultTaskQConfig(testQueueName))
 		opt(options)
 		assert.NotNil(t, options.taskq.config)
 	})

taskmanager/taskmanager.go

@@ -40,7 +40,7 @@ type (
 //
 // If no options are given, it will use the defaultClientOptions()
 // ctx may contain a NewRelic txn (or one will be created)
-func NewClient(_ context.Context, opts ...ClientOps) (TaskManagerInterface, error) {
+func NewClient(_ context.Context, opts ...ClientOps) (Tasker, error) {
 	// Create a new client with defaults
 	client := &Client{options: defaultClientOptions()}