package/mq
1
0
Fork 0
mirror of https://github.com/oarkflow/mq.git synced 2025-10-16 03:10:51 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat: [wip] - Implement html node

Browse Source
This commit is contained in:
sujit
2024-11-18 08:55:44 +05:45
parent f9b0afdac6
commit f9b09272e7
1 changed files with 5 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
examples/v2.go
View File

@@ -206,12 +206,10 @@ func (tm *TaskManager) aggregateResults(parentNode string, taskID string) {
defer tm.mu.Unlock() defer tm.mu.Unlock()
state := tm.TaskStates[taskID][parentNode] state := tm.TaskStates[taskID][parentNode]
if len(state.targetResults) > 1 { if len(state.targetResults) > 1 {
aggregatedData := make([]any, len(state.targetResults)) aggregatedData := make([]json.RawMessage, len(state.targetResults))
i := 0 i := 0
for _, result := range state.targetResults { for _, result := range state.targetResults {
var data map[string]any aggregatedData[i] = result.Data
json.Unmarshal(result.Data, &data)
aggregatedData[i] = data
i++ i++
} }
aggregatedPayload, _ := json.Marshal(aggregatedData) aggregatedPayload, _ := json.Marshal(aggregatedData)
@@ -232,7 +230,7 @@ func finalResultCallback(taskID string, result Result) {
} }
func generateTaskID() string { func generateTaskID() string {
return strconv.Itoa(rand.Intn(100000)) // Random taskID generation return strconv.Itoa(rand.Intn(100000))
} }
func (tm *TaskManager) formHandler(w http.ResponseWriter, r *http.Request) { func (tm *TaskManager) formHandler(w http.ResponseWriter, r *http.Request) {
@@ -243,7 +241,7 @@ func (tm *TaskManager) formHandler(w http.ResponseWriter, r *http.Request) {
email := r.FormValue("email") email := r.FormValue("email")
age := r.FormValue("age") age := r.FormValue("age")
gender := r.FormValue("gender") gender := r.FormValue("gender")
taskID := generateTaskID() // Generate TaskID on form submission taskID := generateTaskID()
payload := fmt.Sprintf(`{"email": "%s", "age": "%s", "gender": "%s"}`, email, age, gender) payload := fmt.Sprintf(`{"email": "%s", "age": "%s", "gender": "%s"}`, email, age, gender)
tm.Trigger(taskID, "NodeA", json.RawMessage(payload)) tm.Trigger(taskID, "NodeA", json.RawMessage(payload))
http.Redirect(w, r, "/result?taskID="+taskID, http.StatusFound) http.Redirect(w, r, "/result?taskID="+taskID, http.StatusFound)
@@ -260,17 +258,13 @@ func (tm *TaskManager) taskStatusHandler(w http.ResponseWriter, r *http.Request)
http.Error(w, "taskID is missing", http.StatusBadRequest) http.Error(w, "taskID is missing", http.StatusBadRequest)
return return
} }
tm.mu.Lock() tm.mu.Lock()
state := tm.TaskStates[taskID] state := tm.TaskStates[taskID]
tm.mu.Unlock() tm.mu.Unlock()
if state == nil { if state == nil {
http.Error(w, "Invalid taskID", http.StatusNotFound) http.Error(w, "Invalid taskID", http.StatusNotFound)
return return
} }
// Return the final result as JSON or other response format
w.Header().Set("Content-Type", "application/json") w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(state) json.NewEncoder(w).Encode(state)
} }
@@ -307,21 +301,16 @@ func main() {
tm.AddNode("Result", func(payload json.RawMessage) Result { tm.AddNode("Result", func(payload json.RawMessage) Result {
var data map[string]any var data map[string]any
json.Unmarshal(payload, &data) json.Unmarshal(payload, &data)
// Show the final result
// You can render the data as an HTML result here
return Result{Data: payload, Status: StatusCompleted} return Result{Data: payload, Status: StatusCompleted}
}) })
tm.AddEdge("Form", "NodeA") tm.AddEdge("Form", "NodeA")
tm.AddEdge("NodeA", "NodeB") tm.AddEdge("NodeA", "NodeB")
tm.AddEdge("NodeB", "NodeC") tm.AddEdge("NodeB", "NodeC")
tm.AddEdge("NodeC", "Result") tm.AddEdge("NodeC", "Result")
http.HandleFunc("/form", tm.formHandler) http.HandleFunc("/form", tm.formHandler)
http.HandleFunc("/result", tm.resultHandler) http.HandleFunc("/result", tm.resultHandler)
http.HandleFunc("/task-result", tm.taskStatusHandler) http.HandleFunc("/task-result", tm.taskStatusHandler)
go tm.Run() go tm.Run()
http.ListenAndServe(":8080", nil) http.ListenAndServe(":8080", nil)
} }