update

2025-10-09 08:50:46 +08:00 · 2025-08-06 10:48:37 +05:45
parent 29d09d2d11
commit f728f89ff9
3 changed files with 169 additions and 62 deletions
--- a/handlers/split_join_handler.go
+++ b/handlers/split_join_handler.go
@@ -10,12 +10,12 @@ import (
 	"github.com/oarkflow/mq/dag"
 )

-// SplitJoinHandler handles string split and join operations
-type SplitJoinHandler struct {
+// SplitHandler handles string split operations
+type SplitHandler struct {
 	dag.Operation
 }

-func (h *SplitJoinHandler) ProcessTask(ctx context.Context, task *mq.Task) mq.Result {
+func (h *SplitHandler) ProcessTask(ctx context.Context, task *mq.Task) mq.Result {
 	var data map[string]any
 	err := json.Unmarshal(task.Payload, &data)
 	if err != nil {
@@ -31,12 +31,8 @@ func (h *SplitJoinHandler) ProcessTask(ctx context.Context, task *mq.Task) mq.Re
 	switch operation {
 	case "split":
 		result = h.splitOperation(data)
-	case "join":
-		result = h.joinOperation(data)
 	case "split_to_array":
 		result = h.splitToArrayOperation(data)
-	case "join_from_array":
-		result = h.joinFromArrayOperation(data)
 	default:
 		return mq.Result{Error: fmt.Errorf("unsupported operation: %s", operation)}
 	}
@@ -49,11 +45,13 @@ func (h *SplitJoinHandler) ProcessTask(ctx context.Context, task *mq.Task) mq.Re
 	return mq.Result{Payload: resultPayload, Ctx: ctx}
 }

-func (h *SplitJoinHandler) splitOperation(data map[string]any) map[string]any {
+func (h *SplitHandler) splitOperation(data map[string]any) map[string]any {
 	result := make(map[string]any)
 	fields := h.getTargetFields()
 	separator := h.getSeparator()

+	fmt.Printf("Split Operation: Fields=%v, Separator='%s'\n", fields, separator)
+
 	// Copy all original data
 	for key, value := range data {
 		result[key] = value
@@ -64,6 +62,8 @@ func (h *SplitJoinHandler) splitOperation(data map[string]any) map[string]any {
 			if str, ok := val.(string); ok {
 				parts := strings.Split(str, separator)

+				fmt.Printf("Splitting field '%s': Original='%s', Parts=%v\n", field, str, parts)
+
 				// Create individual fields for each part
 				for i, part := range parts {
 					result[fmt.Sprintf("%s_%d", field, i)] = strings.TrimSpace(part)
@@ -79,32 +79,7 @@ func (h *SplitJoinHandler) splitOperation(data map[string]any) map[string]any {
 	return result
 }

-func (h *SplitJoinHandler) joinOperation(data map[string]any) map[string]any {
-	result := make(map[string]any)
-	targetField := h.getTargetField()
-	separator := h.getSeparator()
-	sourceFields := h.getSourceFields()
-
-	// Copy all original data
-	for key, value := range data {
-		result[key] = value
-	}
-
-	var parts []string
-	for _, field := range sourceFields {
-		if val, ok := data[field]; ok && val != nil {
-			parts = append(parts, fmt.Sprintf("%v", val))
-		}
-	}
-
-	if len(parts) > 0 {
-		result[targetField] = strings.Join(parts, separator)
-	}
-
-	return result
-}
-
-func (h *SplitJoinHandler) splitToArrayOperation(data map[string]any) map[string]any {
+func (h *SplitHandler) splitToArrayOperation(data map[string]any) map[string]any {
 	result := make(map[string]any)
 	fields := h.getTargetFields()
 	separator := h.getSeparator()
@@ -130,7 +105,100 @@ func (h *SplitJoinHandler) splitToArrayOperation(data map[string]any) map[string
 	return result
 }

-func (h *SplitJoinHandler) joinFromArrayOperation(data map[string]any) map[string]any {
+func (h *SplitHandler) getTargetFields() []string {
+	if fields, ok := h.Payload.Data["fields"].([]string); ok {
+		return fields
+	}
+	if fields, ok := h.Payload.Data["fields"].([]interface{}); ok {
+		var result []string
+		for _, field := range fields {
+			if str, ok := field.(string); ok {
+				result = append(result, str)
+			}
+		}
+		return result
+	}
+	return nil
+}
+
+func (h *SplitHandler) getSeparator() string {
+	if sep, ok := h.Payload.Data["separator"].(string); ok {
+		return sep
+	}
+	return "," // Default separator
+}
+
+func NewSplitHandler(id string) *SplitHandler {
+	return &SplitHandler{
+		Operation: dag.Operation{ID: id, Key: "split", Type: dag.Function, Tags: []string{"data", "transformation", "string"}},
+	}
+}
+
+// JoinHandler handles string join operations
+type JoinHandler struct {
+	dag.Operation
+}
+
+func (h *JoinHandler) ProcessTask(ctx context.Context, task *mq.Task) mq.Result {
+	var data map[string]any
+	err := json.Unmarshal(task.Payload, &data)
+	if err != nil {
+		return mq.Result{Error: fmt.Errorf("failed to unmarshal task payload: %w", err)}
+	}
+
+	operation, ok := h.Payload.Data["operation"].(string)
+	if !ok {
+		return mq.Result{Error: fmt.Errorf("operation not specified")}
+	}
+
+	var result map[string]any
+	switch operation {
+	case "join":
+		result = h.joinOperation(data)
+	case "join_from_array":
+		result = h.joinFromArrayOperation(data)
+	default:
+		return mq.Result{Error: fmt.Errorf("unsupported operation: %s", operation)}
+	}
+
+	resultPayload, err := json.Marshal(result)
+	if err != nil {
+		return mq.Result{Error: fmt.Errorf("failed to marshal result: %w", err)}
+	}
+
+	return mq.Result{Payload: resultPayload, Ctx: ctx}
+}
+
+func (h *JoinHandler) joinOperation(data map[string]any) map[string]any {
+	result := make(map[string]any)
+	targetField := h.getTargetField()
+	separator := h.getSeparator()
+	sourceFields := h.getSourceFields()
+
+	fmt.Printf("Join Operation: TargetField='%s', Separator='%s', SourceFields=%v\n", targetField, separator, sourceFields)
+
+	// Copy all original data
+	for key, value := range data {
+		result[key] = value
+	}
+
+	var parts []string
+	for _, field := range sourceFields {
+		if val, ok := data[field]; ok && val != nil {
+			parts = append(parts, fmt.Sprintf("%v", val))
+		}
+	}
+
+	fmt.Printf("Joining fields: Parts=%v\n", parts)
+
+	if len(parts) > 0 {
+		result[targetField] = strings.Join(parts, separator)
+	}
+
+	return result
+}
+
+func (h *JoinHandler) joinFromArrayOperation(data map[string]any) map[string]any {
 	result := make(map[string]any)
 	targetField := h.getTargetField()
 	separator := h.getSeparator()
@@ -156,34 +224,21 @@ func (h *SplitJoinHandler) joinFromArrayOperation(data map[string]any) map[strin
 	return result
 }

-func (h *SplitJoinHandler) getTargetFields() []string {
-	if fields, ok := h.Payload.Data["fields"].([]interface{}); ok {
-		var result []string
-		for _, field := range fields {
-			if str, ok := field.(string); ok {
-				result = append(result, str)
-			}
-		}
-		return result
-	}
-	return nil
-}
-
-func (h *SplitJoinHandler) getTargetField() string {
+func (h *JoinHandler) getTargetField() string {
 	if field, ok := h.Payload.Data["target_field"].(string); ok {
 		return field
 	}
 	return "joined_field"
 }

-func (h *SplitJoinHandler) getSourceField() string {
+func (h *JoinHandler) getSourceField() string {
 	if field, ok := h.Payload.Data["source_field"].(string); ok {
 		return field
 	}
 	return ""
 }

-func (h *SplitJoinHandler) getSourceFields() []string {
+func (h *JoinHandler) getSourceFields() []string {
 	if fields, ok := h.Payload.Data["source_fields"].([]interface{}); ok {
 		var result []string
 		for _, field := range fields {
@@ -196,15 +251,15 @@ func (h *SplitJoinHandler) getSourceFields() []string {
 	return nil
 }

-func (h *SplitJoinHandler) getSeparator() string {
+func (h *JoinHandler) getSeparator() string {
 	if sep, ok := h.Payload.Data["separator"].(string); ok {
 		return sep
 	}
 	return "," // Default separator
 }

-func NewSplitJoinHandler(id string) *SplitJoinHandler {
-	return &SplitJoinHandler{
-		Operation: dag.Operation{ID: id, Key: "split_join", Type: dag.Function, Tags: []string{"data", "transformation", "string"}},
+func NewJoinHandler(id string) *JoinHandler {
+	return &JoinHandler{
+		Operation: dag.Operation{ID: id, Key: "join", Type: dag.Function, Tags: []string{"data", "transformation", "string"}},
 	}
 }