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

update: dependencies

Browse Source
This commit is contained in:
Oarkflow
2024-11-14 13:05:39 +05:45
parent 824bdb35d5
commit 29aa276f01
1 changed files with 31 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
examples/html.go
View File

@@ -2,6 +2,7 @@ package main
import ( import (
"context" "context"
"encoding/json"
"fmt" "fmt"
"log" "log"
"math/rand" "math/rand"
@@ -29,7 +30,7 @@ type Result struct {
type Task struct { type Task struct {
ID string ID string
CurrentNodeID string CurrentNodeID string
Inputs map[string]string Payload json.RawMessage
FinalResult string FinalResult string
} }
@@ -39,7 +40,8 @@ type PageNode struct {
} }
func (n *PageNode) ProcessTask(ctx context.Context, task *Task) Result { func (n *PageNode) ProcessTask(ctx context.Context, task *Task) Result {
return Result{Message: n.Content} contentWithTaskID := strings.ReplaceAll(n.Content, "{{taskID}}", task.ID)
return Result{Message: contentWithTaskID}
} }
func (n *PageNode) GetNodeType() string { func (n *PageNode) GetNodeType() string {
@@ -106,7 +108,6 @@ func (tm *TaskManager) StartTask() *Task {
task := &Task{ task := &Task{
ID: taskID, ID: taskID,
CurrentNodeID: "customRegistration", CurrentNodeID: "customRegistration",
Inputs: make(map[string]string),
} }
tm.tasks[taskID] = task tm.tasks[taskID] = task
return task return task
@@ -144,52 +145,36 @@ func generateTaskID() string {
func processNode(w http.ResponseWriter, r *http.Request, task *Task, tm *TaskManager) { func processNode(w http.ResponseWriter, r *http.Request, task *Task, tm *TaskManager) {
for { for {
log.Printf("Processing taskID: %s, Current Node: %s", task.ID, task.CurrentNodeID) log.Printf("Processing taskID: %s, Current Node: %s", task.ID, task.CurrentNodeID)
// Retrieve the current node
node, exists := tm.graph.Nodes[task.CurrentNodeID] node, exists := tm.graph.Nodes[task.CurrentNodeID]
if !exists { if !exists {
http.Error(w, "Node not found", http.StatusInternalServerError) http.Error(w, "Node not found", http.StatusInternalServerError)
return return
} }
// Process the task with the current node
result := node.ProcessTask(context.Background(), task) result := node.ProcessTask(context.Background(), task)
log.Printf("Node %s processed. Result ConditionStatus: %s", task.CurrentNodeID, result.ConditionStatus) log.Printf("Node %s processed. Result ConditionStatus: %s", task.CurrentNodeID, result.ConditionStatus)
// If the node type is PageType, render the content with the task ID and return
if node.GetNodeType() == PageType { if node.GetNodeType() == PageType {
contentWithTaskID := strings.ReplaceAll(result.Message, "{{taskID}}", task.ID) contentWithTaskID := strings.ReplaceAll(result.Message, "{{taskID}}", task.ID)
fmt.Fprintf(w, contentWithTaskID) fmt.Fprintf(w, contentWithTaskID)
return return
} }
// Handle any processing error
if result.Error != nil { if result.Error != nil {
http.Error(w, result.Error.Error(), http.StatusInternalServerError) http.Error(w, result.Error.Error(), http.StatusInternalServerError)
return return
} }
// Determine the next node ID
nextNodeID := result.ConditionStatus nextNodeID := result.ConditionStatus
if nextNodeID == "" { if nextNodeID == "" {
// If ConditionStatus is empty, try to find the next node in the DAG's edges
edges := tm.graph.Edges[task.CurrentNodeID] edges := tm.graph.Edges[task.CurrentNodeID]
if len(edges) > 0 { if len(edges) > 0 {
nextNodeID = edges[0] // Get the first edge's destination as the next node nextNodeID = edges[0]
log.Printf("No ConditionStatus found, following edge to next node: %s", nextNodeID) log.Printf("No ConditionStatus found, following edge to next node: %s", nextNodeID)
} else { } else {
// If no edges are found, the process is complete
log.Printf("Task %s completed. Final result: %s", task.ID, task.FinalResult) log.Printf("Task %s completed. Final result: %s", task.ID, task.FinalResult)
fmt.Fprintf(w, "<html><body><h1>Process Completed</h1><p>%s</p></body></html>", task.FinalResult) fmt.Fprintf(w, "<html><body><h1>Process Completed</h1><p>%s</p></body></html>", task.FinalResult)
return return
} }
} }
// Update the task with the new node ID and persist the task state
task.CurrentNodeID = nextNodeID task.CurrentNodeID = nextNodeID
tm.UpdateTask(task) tm.UpdateTask(task)
// Check if the next node is a PageType and handle redirection if necessary
if nextNode, nextExists := tm.graph.Nodes[nextNodeID]; nextExists && nextNode.GetNodeType() == PageType { if nextNode, nextExists := tm.graph.Nodes[nextNodeID]; nextExists && nextNode.GetNodeType() == PageType {
log.Printf("Redirecting to next page: %s", nextNodeID) log.Printf("Redirecting to next page: %s", nextNodeID)
http.Redirect(w, r, fmt.Sprintf("/render?taskID=%s", task.ID), http.StatusFound) http.Redirect(w, r, fmt.Sprintf("/render?taskID=%s", task.ID), http.StatusFound)
@@ -220,26 +205,26 @@ func submitHandler(w http.ResponseWriter, r *http.Request, tm *TaskManager) {
http.Error(w, "Failed to parse form data", http.StatusBadRequest) http.Error(w, "Failed to parse form data", http.StatusBadRequest)
return return
} }
task.Inputs = make(map[string]string) inputData := make(map[string]string)
for key, values := range r.Form { for key, values := range r.Form {
if len(values) > 0 { if len(values) > 0 {
task.Inputs[key] = values[0] inputData[key] = values[0]
} }
} }
nextNodes, exists := tm.GetNextNode(task) rawInputs, _ := json.Marshal(inputData)
task.Payload = rawInputs
nextNode, exists := tm.GetNextNode(task)
if !exists { if !exists {
log.Printf("Task %s completed. Final result: %s", task.ID, task.FinalResult) log.Printf("Task %s completed. Final result: %s", task.ID, task.FinalResult)
fmt.Fprintf(w, "<html><body><h1>Process Completed</h1><p>%s</p></body></html>", task.FinalResult) fmt.Fprintf(w, "<html><body><h1>Process Completed</h1><p>%s</p></body></html>", task.FinalResult)
return return
} }
switch nextNode := nextNode.(type) {
switch nextNodes := nextNodes.(type) {
case *PageNode: case *PageNode:
task.CurrentNodeID = nextNodes.ID task.CurrentNodeID = nextNode.ID
case *FunctionNode: case *FunctionNode:
task.CurrentNodeID = nextNodes.ID task.CurrentNodeID = nextNode.ID
} }
fmt.Println(task.CurrentNodeID)
processNode(w, r, task, tm) processNode(w, r, task, tm)
} }
@@ -249,11 +234,11 @@ func startTaskHandler(w http.ResponseWriter, r *http.Request, tm *TaskManager) {
} }
func isValidEmail(email string) bool { func isValidEmail(email string) bool {
return true return email != ""
} }
func isValidPhone(phone string) bool { func isValidPhone(phone string) bool {
return true return phone != ""
} }
func verifyHandler(w http.ResponseWriter, r *http.Request, tm *TaskManager) { func verifyHandler(w http.ResponseWriter, r *http.Request, tm *TaskManager) {
@@ -267,7 +252,11 @@ func verifyHandler(w http.ResponseWriter, r *http.Request, tm *TaskManager) {
http.Error(w, "Task not found", http.StatusNotFound) http.Error(w, "Task not found", http.StatusNotFound)
return return
} }
task.Inputs["email_verified"] = "true" data := map[string]any{
"email_verified": "true",
}
bt, _ := json.Marshal(data)
task.Payload = bt
log.Printf("Email for taskID %s successfully verified.", task.ID) log.Printf("Email for taskID %s successfully verified.", task.ID)
nextNode, exists := tm.graph.Nodes["dashboard"] nextNode, exists := tm.graph.Nodes["dashboard"]
if !exists { if !exists {
@@ -293,7 +282,12 @@ func main() {
checkValidityNode := &FunctionNode{ checkValidityNode := &FunctionNode{
ID: "checkValidity", ID: "checkValidity",
Func: func(task *Task) Result { Func: func(task *Task) Result {
email, phone := task.Inputs["email"], task.Inputs["phone"] var inputs map[string]string
if err := json.Unmarshal(task.Payload, &inputs); err != nil {
return Result{ConditionStatus: "customRegistration", Message: "Invalid input format"}
}
email, phone := inputs["email"], inputs["phone"]
if !isValidEmail(email) || !isValidPhone(phone) { if !isValidEmail(email) || !isValidPhone(phone) {
return Result{ return Result{
ConditionStatus: "customRegistration", ConditionStatus: "customRegistration",
@@ -306,7 +300,11 @@ func main() {
checkManualVerificationNode := &FunctionNode{ checkManualVerificationNode := &FunctionNode{
ID: "checkManualVerification", ID: "checkManualVerification",
Func: func(task *Task) Result { Func: func(task *Task) Result {
city := task.Inputs["city"] var inputs map[string]string
if err := json.Unmarshal(task.Payload, &inputs); err != nil {
return Result{ConditionStatus: "customRegistration", Message: "Invalid input format"}
}
city := inputs["city"]
if city != "Kathmandu" { if city != "Kathmandu" {
return Result{ConditionStatus: "manualVerificationPage"} return Result{ConditionStatus: "manualVerificationPage"}
} }