enhance helper, db update & add worker pool util

candihelper/helper.go

@@ -700,3 +700,26 @@ func UnwrapPtr[T any](t *T) (res T) {
 func WrapPtr[T any](t T) *T {
 	return &t
 }
+
+// ToMap transform slice to map
+func ToMap[T any, K comparable](list []T, keyGetter func(T) K) map[K]T {
+	mp := make(map[K]T, len(list))
+	for _, el := range list {
+		mp[keyGetter(el)] = el
+	}
+	return mp
+}
+
+// IsExistInMap check key is exist in map
+func IsExistInMap[T any, K comparable](m map[K]T, key K) bool {
+	_, ok := m[key]
+	return ok
+}
+
+// ToKeyMapSlice transform key of map to slice
+func ToKeyMapSlice[T any, K comparable](mp map[K]T) (list []K) {
+	for k := range mp {
+		list = append(list, k)
+	}
+	return list
+}

candishared/database_update_tools.go

@@ -54,25 +54,26 @@ type DBUpdateTools struct {
 	IgnoredFields    []string
 }
 
-// ToMap method
-func (d DBUpdateTools) ToMap(data interface{}, opts ...DBUpdateOptionFunc) map[string]interface{} {
-	var (
-		o            partialUpdateOption
-		updateFields = make(map[string]interface{}, 0)
-	)
-
+func (d *DBUpdateTools) parseOption(opts ...DBUpdateOptionFunc) (o partialUpdateOption) {
 	for _, opt := range opts {
 		opt(&o)
 	}
+	return o
+}
+
+// ToMap method
+func (d DBUpdateTools) ToMap(data interface{}, opts ...DBUpdateOptionFunc) map[string]interface{} {
+	opt := d.parseOption(opts...)
 
 	dataValue := reflect.ValueOf(data)
 	dataType := reflect.TypeOf(data)
 	if dataValue.Kind() == reflect.Ptr {
 		dataValue = dataValue.Elem()
 		dataType = dataType.Elem()
 	}
-	isPartial := len(o.updateFields) > 0 || len(o.ignoreFields) > 0
+	isPartial := len(opt.updateFields) > 0 || len(opt.ignoreFields) > 0
 
+	updateFields := make(map[string]interface{}, 0)
 	for i := 0; i < dataValue.NumField(); i++ {
 		fieldValue := dataValue.Field(i)
 		fieldType := dataType.Field(i)
@@ -103,9 +104,9 @@ func (d DBUpdateTools) ToMap(data interface{}, opts ...DBUpdateOptionFunc) map[s
 			continue
 		}
 
-		_, isFieldUpdated := o.updateFields[fieldType.Name]
-		_, isFieldIgnored := o.ignoreFields[fieldType.Name]
-		if (isFieldUpdated && len(o.updateFields) > 0) || (!isFieldIgnored && len(o.ignoreFields) > 0) {
+		_, isFieldUpdated := opt.updateFields[fieldType.Name]
+		_, isFieldIgnored := opt.ignoreFields[fieldType.Name]
+		if (isFieldUpdated && len(opt.updateFields) > 0) || (!isFieldIgnored && len(opt.ignoreFields) > 0) {
 			updateFields[key] = val
 		}
 	}
@@ -115,3 +116,15 @@ func (d DBUpdateTools) ToMap(data interface{}, opts ...DBUpdateOptionFunc) map[s
 	}
 	return updateFields
 }
+
+// GetUpdatedFields method
+func (d DBUpdateTools) GetFields(opts ...DBUpdateOptionFunc) (updates, ignores []string) {
+	opt := d.parseOption(opts...)
+	for k := range opt.updateFields {
+		updates = append(updates, k)
+	}
+	for k := range opt.ignoreFields {
+		ignores = append(ignores, k)
+	}
+	return
+}

candiutils/worker_pool.go

@@ -0,0 +1,51 @@
+package candiutils
+
+import (
+	"context"
+	"sync"
+)
+
+// WorkerPool implementation
+type WorkerPool[T any] interface {
+	Dispatch(ctx context.Context, jobFunc func(context.Context, T))
+	AddJob(job T)
+	Finish()
+}
+
+type workerPool[T any] struct {
+	maxWorker int
+	wg        sync.WaitGroup
+	jobChan   chan T
+}
+
+// NewWorkerPool create an instance of WorkerPool.
+func NewWorkerPool[T any](maxWorker int) WorkerPool[T] {
+	wp := &workerPool[T]{
+		maxWorker: maxWorker,
+		wg:        sync.WaitGroup{},
+		jobChan:   make(chan T),
+	}
+
+	return wp
+}
+
+func (wp *workerPool[T]) Dispatch(ctx context.Context, jobFunc func(context.Context, T)) {
+	for i := 0; i < wp.maxWorker; i++ {
+		go func(jobFunc func(context.Context, T)) {
+			for job := range wp.jobChan {
+				jobFunc(ctx, job)
+				wp.wg.Done()
+			}
+		}(jobFunc)
+	}
+}
+
+func (wp *workerPool[T]) AddJob(job T) {
+	wp.wg.Add(1)
+	wp.jobChan <- job
+}
+
+func (wp *workerPool[T]) Finish() {
+	close(wp.jobChan)
+	wp.wg.Wait()
+}