package/mq
1
0
Fork 0
mirror of https://github.com/oarkflow/mq.git synced 2026-04-22 22:47:06 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat: update

Browse Source
This commit is contained in:
sujit
2025-10-07 16:07:11 +05:45
parent 98e60906ce
commit 9a2af7761f
2 changed files with 73 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dag/task_manager.go
+15 -2
View File
@@ -461,11 +461,19 @@ func (tm *TaskManager) areDependenciesMet(nodeID string) bool {
return true
})
}
if !exists || state.Status != mq.Completed {
if !exists {
tm.dag.Logger().Debug("Dependency not met",
logger.Field{Key: "nodeID", Value: nodeID},
logger.Field{Key: "dependency", Value: prevNode.ID},
logger.Field{Key: "stateExists", Value: exists},
logger.Field{Key: "stateExists", Value: false},
logger.Field{Key: "taskID", Value: tm.taskID})
return false
}
if state == nil || state.Status != mq.Completed {
tm.dag.Logger().Debug("Dependency not met",
logger.Field{Key: "nodeID", Value: nodeID},
logger.Field{Key: "dependency", Value: prevNode.ID},
logger.Field{Key: "stateExists", Value: true},
logger.Field{Key: "stateStatus", Value: string(state.Status)},
logger.Field{Key: "taskID", Value: tm.taskID})
return false
@@ -1032,6 +1040,11 @@ func (tm *TaskManager) processSingleEdge(currentResult nodeResult, edge Edge) {
tm.parentNodes.Set(childNode, parentNode)
}
// Handle dependencies for conditional edges
if strings.HasPrefix(edge.Label, "condition:") {
tm.handleTargetNodeDependencies(edge.To.ID, currentResult)
}
tm.enqueueTask(ctx, edge.To.ID, tm.taskID, currentResult.result.Payload)
}
}
examples/bulk_sms.go
+58 -5
View File
@@ -14,6 +14,7 @@ import (
"time"
"github.com/gofiber/fiber/v2/middleware/session"
"github.com/oarkflow/mq/dag"
"github.com/oarkflow/mq/utils"
@@ -968,11 +969,63 @@ func (p *BulkLoginPage) ProcessTask(ctx context.Context, task *mq.Task) mq.Resul
if sess, ok := ctx.Value("session").(*session.Session); ok {
if authenticated, exists := sess.Get("authenticated").(bool); exists && authenticated {
// User is already authenticated, show auto-submit page
htmlContentBytes, err := os.ReadFile("views/bulk_login_page.html")
if err != nil {
return mq.Result{Error: fmt.Errorf("failed to load HTML template: %v", err), Ctx: ctx}
}
htmlContent := string(htmlContentBytes)
htmlContent := `
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Phone Processing System - Already Authenticated</title>
<style>
body {
font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;
background: #f5f5f5;
margin: 0;
padding: 0;
min-height: 100vh;
display: flex;
align-items: center;
justify-content: center;
flex-direction: column;
}
.spinner {
width: 40px;
height: 40px;
border: 4px solid #f3f3f3;
border-top: 4px solid #667eea;
border-radius: 50%;
animation: spin 1s linear infinite;
}
.status-text {
margin-top: 12px;
color: #4a5568;
font-size: 14px;
}
@keyframes spin {
0% { transform: rotate(0deg); }
100% { transform: rotate(360deg); }
}
</style>
</head>
<body>
<div class="spinner"></div>
<div class="status-text">You are already authenticated. Redirecting to the next step...</div>
<form id="autoSubmitForm" method="post" action="/process?task_id=` + fmt.Sprintf("%v", ctx.Value("task_id")) + `&next=true" style="display: none;">
<input type="hidden" name="authenticated" value="true">
<input type="hidden" name="username" value="` + fmt.Sprintf("%v", sess.Get("username")) + `">
<input type="hidden" name="user_role" value="` + fmt.Sprintf("%v", sess.Get("user_role")) + `">
<input type="hidden" name="auth_token" value="` + fmt.Sprintf("%v", sess.Get("auth_token")) + `">
<input type="hidden" name="skip_login" value="true">
</form>
<script>
setTimeout(() => {
document.getElementById('autoSubmitForm').submit();
}, 1000);
</script>
</body>
</html>`
ctx = context.WithValue(ctx, consts.ContentType, consts.TypeHtml)
resultData := map[string]any{