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

update

Browse Source
This commit is contained in:
Oarkflow
2025-08-09 18:15:07 +05:45
parent 84577dfc57
commit e51641f402
9 changed files with 504 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
handlers/common_handler.go
View File

@@ -44,7 +44,7 @@ func (e *Condition) ProcessTask(ctx context.Context, task *mq.Task) mq.Result {
var conditionStatus string var conditionStatus string
_, ok := e.conditions[defaultKey] _, ok := e.conditions[defaultKey]
for status, condition := range e.conditions { for status, condition := range e.conditions {
if status != defaultKey { if status != defaultKey && condition != nil {
if condition.Match(data) { if condition.Match(data) {
conditionStatus = status conditionStatus = status
} }

39
handlers/html_handler.go
View File

@@ -8,6 +8,7 @@ import (
"html/template" "html/template"
"os" "os"
"github.com/oarkflow/jet"
"github.com/oarkflow/mq" "github.com/oarkflow/mq"
"github.com/oarkflow/mq/consts" "github.com/oarkflow/mq/consts"
"github.com/oarkflow/mq/dag" "github.com/oarkflow/mq/dag"
@@ -26,24 +27,30 @@ func (c *RenderHTMLNode) ProcessTask(ctx context.Context, task *mq.Task) mq.Resu
templateStr, _ = data["template"].(string) templateStr, _ = data["template"].(string)
templateFile, _ = data["template_file"].(string) templateFile, _ = data["template_file"].(string)
) )
var templateData map[string]any var templateData map[string]any
if len(task.Payload) > 0 { if len(task.Payload) > 0 {
if err := json.Unmarshal(task.Payload, &templateData); err != nil { if err := json.Unmarshal(task.Payload, &templateData); err != nil {
return mq.Result{Payload: task.Payload, Error: err, Ctx: ctx, ConditionStatus: "invalid"} return mq.Result{Payload: task.Payload, Error: err, Ctx: ctx}
} }
} }
if templateData == nil { if templateData == nil {
templateData = make(map[string]any) templateData = make(map[string]any)
} }
delete(templateData, "html_content")
if c.Payload.Mapping != nil {
for k, v := range c.Payload.Mapping {
_, val := dag.GetVal(ctx, v, templateData)
templateData[k] = val
}
}
templateData["task_id"] = ctx.Value("task_id") templateData["task_id"] = ctx.Value("task_id")
var renderedHTML string var renderedHTML string
var err error var err error
parser := jet.NewWithMemory(jet.WithDelims("{{", "}}"))
switch { switch {
// 1. JSONSchema + HTML Template // 1. JSONSchema + HTML Template
case schemaFile != "" && templateStr != "": case schemaFile != "" && templateStr != "" && templateFile == "":
fmt.Println("Using JSONSchema and inline HTML template", c.ID)
if c.renderer == nil { if c.renderer == nil {
renderer, err := renderer.GetFromFile(schemaFile, templateStr) renderer, err := renderer.GetFromFile(schemaFile, templateStr)
if err != nil { if err != nil {
@@ -53,7 +60,8 @@ func (c *RenderHTMLNode) ProcessTask(ctx context.Context, task *mq.Task) mq.Resu
} }
renderedHTML, err = c.renderer.RenderFields(templateData) renderedHTML, err = c.renderer.RenderFields(templateData)
// 2. JSONSchema + HTML File // 2. JSONSchema + HTML File
case schemaFile != "" && templateFile != "": case schemaFile != "" && templateFile != "" && templateStr == "":
fmt.Println("Using JSONSchema and HTML file", c.ID)
if c.renderer == nil { if c.renderer == nil {
renderer, err := renderer.GetFromFile(schemaFile, "", templateFile) renderer, err := renderer.GetFromFile(schemaFile, "", templateFile)
if err != nil { if err != nil {
@@ -63,7 +71,8 @@ func (c *RenderHTMLNode) ProcessTask(ctx context.Context, task *mq.Task) mq.Resu
} }
renderedHTML, err = c.renderer.RenderFields(templateData) renderedHTML, err = c.renderer.RenderFields(templateData)
// 3. Only JSONSchema // 3. Only JSONSchema
case schemaFile != "" || c.renderer != nil: case (schemaFile != "" || c.renderer != nil) && templateStr == "" && templateFile == "":
fmt.Println("Using only JSONSchema", c.ID)
if c.renderer == nil { if c.renderer == nil {
renderer, err := renderer.GetFromFile(schemaFile, "") renderer, err := renderer.GetFromFile(schemaFile, "")
if err != nil { if err != nil {
@@ -73,7 +82,8 @@ func (c *RenderHTMLNode) ProcessTask(ctx context.Context, task *mq.Task) mq.Resu
} }
renderedHTML, err = c.renderer.RenderFields(templateData) renderedHTML, err = c.renderer.RenderFields(templateData)
// 4. Only HTML Template // 4. Only HTML Template
case templateStr != "": case templateStr != "" && templateFile == "" && schemaFile == "":
fmt.Println("Using inline HTML template", c.ID)
tmpl, err := template.New("inline").Parse(templateStr) tmpl, err := template.New("inline").Parse(templateStr)
if err != nil { if err != nil {
return mq.Result{Error: fmt.Errorf("failed to parse template: %v", err), Ctx: ctx} return mq.Result{Error: fmt.Errorf("failed to parse template: %v", err), Ctx: ctx}
@@ -85,21 +95,16 @@ func (c *RenderHTMLNode) ProcessTask(ctx context.Context, task *mq.Task) mq.Resu
} }
renderedHTML = buf.String() renderedHTML = buf.String()
// 5. Only HTML File // 5. Only HTML File
case templateFile != "": case templateFile != "" && templateStr == "" && schemaFile == "":
fmt.Println("Using HTML file", c.ID)
fileContent, err := os.ReadFile(templateFile) fileContent, err := os.ReadFile(templateFile)
if err != nil { if err != nil {
return mq.Result{Error: fmt.Errorf("failed to read template file: %v", err), Ctx: ctx} return mq.Result{Error: fmt.Errorf("failed to read template file: %v", err), Ctx: ctx}
} }
tmpl, err := template.New("file").Parse(string(fileContent)) renderedHTML, err = parser.ParseTemplate(string(fileContent), templateData)
if err != nil { if err != nil {
return mq.Result{Error: fmt.Errorf("failed to parse template file: %v", err), Ctx: ctx} return mq.Result{Error: err, Ctx: ctx}
} }
var buf bytes.Buffer
err = tmpl.Execute(&buf, templateData)
if err != nil {
return mq.Result{Error: fmt.Errorf("failed to execute template file: %v", err), Ctx: ctx}
}
renderedHTML = buf.String()
default: default:
return mq.Result{Error: fmt.Errorf("no valid rendering approach found"), Ctx: ctx} return mq.Result{Error: fmt.Errorf("no valid rendering approach found"), Ctx: ctx}
} }

42
services/examples/app/templates/basic.html Normal file
View File

@@ -0,0 +1,42 @@
<!DOCTYPE html>
<html>
<head>
<title>Basic Template</title>
<script src="https://cdn.tailwindcss.com"></script>
<link rel="stylesheet" href="form.css">
<style>
.required {
color: #dc3545;
}
.group-header {
font-weight: bold;
margin-top: 0.5rem;
margin-bottom: 0.5rem;
}
.section-title {
color: #0d6efd;
border-bottom: 2px solid #0d6efd;
padding-bottom: 0.5rem;
}
.form-group-fields>div {
margin-bottom: 1rem;
}
</style>
</head>
<body class="bg-gray-100">
<form {{form_attributes}}>
<div class="form-container p-4 bg-white shadow-md rounded">
{{form_groups}}
<div class="mt-4 flex gap-2">
{{form_buttons}}
</div>
</div>
</form>
</body>
</html>

134
services/examples/app/templates/error.html Normal file
View File

@@ -0,0 +1,134 @@
<!DOCTYPE html>
<html>
<head>
<title>Email Error</title>
<style>
body {
font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;
max-width: 700px;
margin: 50px auto;
padding: 20px;
background: linear-gradient(135deg, #FF6B6B 0%, #FF5722 100%);
color: white;
}
.error-container {
background: rgba(255, 255, 255, 0.1);
padding: 40px;
border-radius: 20px;
backdrop-filter: blur(15px);
box-shadow: 0 12px 40px rgba(0, 0, 0, 0.4);
text-align: center;
}
.error-icon {
font-size: 80px;
margin-bottom: 20px;
animation: shake 0.5s ease-in-out infinite alternate;
}
@keyframes shake {
0% {
transform: translateX(0);
}
100% {
transform: translateX(5px);
}
}
h1 {
margin-bottom: 30px;
text-shadow: 2px 2px 4px rgba(0, 0, 0, 0.3);
font-size: 2.5em;
}
.error-message {
background: rgba(255, 255, 255, 0.2);
padding: 25px;
border-radius: 12px;
margin: 25px 0;
font-size: 18px;
border-left: 6px solid #FFB6B6;
line-height: 1.6;
}
.error-details {
background: rgba(255, 255, 255, 0.15);
padding: 20px;
border-radius: 12px;
margin: 25px 0;
text-align: left;
}
.actions {
margin-top: 40px;
}
.btn {
background: linear-gradient(45deg, #4ECDC4, #44A08D);
color: white;
padding: 15px 30px;
border: none;
border-radius: 25px;
cursor: pointer;
font-size: 16px;
font-weight: bold;
margin: 0 15px;
text-decoration: none;
display: inline-block;
transition: all 0.3s ease;
text-transform: uppercase;
letter-spacing: 1px;
}
.btn:hover {
transform: translateY(-3px);
box-shadow: 0 8px 25px rgba(0, 0, 0, 0.3);
}
.retry-btn {
background: linear-gradient(45deg, #FFA726, #FF9800);
}
</style>
</head>
<body>
<div class="error-container">
<div class="error-icon">❌</div>
<h1>Email Processing Error</h1>
<div class="error-message">
{{error_message}}
</div>
{{if error_field}}
<div class="error-details">
<strong>🎯 Error Field:</strong> {{error_field}}<br>
<strong>⚡ Action Required:</strong> Please correct the highlighted field and try again.<br>
<strong>💡 Tip:</strong> Make sure all required fields are properly filled out.
</div>
{{end}}
{{if retry_suggested}}
<div class="error-details">
<strong>⚠️ Temporary Issue:</strong> This appears to be a temporary system issue.
Please try sending your message again in a few moments.<br>
<strong>🔄 Auto-Retry:</strong> Our system will automatically retry failed deliveries.
</div>
{{end}}
<div class="actions">
<a href="/" class="btn retry-btn">🔄 Try Again</a>
<a href="/api/status" class="btn">📊 Check Status</a>
</div>
<div style="margin-top: 30px; font-size: 14px; opacity: 0.8;">
🔄 DAG Error Handler | Email Notification Workflow Failed<br>
Our advanced routing system ensures reliable message delivery.
</div>
</div>
</body>
</html>

71
services/examples/json/login.json Normal file
View File

@@ -0,0 +1,71 @@
{
"name": "Login Flow",
"key": "login:flow",
"nodes": [
{
"id": "LoginForm",
"first_node": true,
"node": "render-html",
"data": {
"additional_data": {
"schema_file": "login.json",
"template_file": "app/templates/basic.html"
}
}
},
{
"id": "ValidateLogin",
"node": "condition",
"data": {
"mapping": {
"username": "username",
"password": "password"
},
"additional_data": {
"conditions": {
"invalid": {
"id": "condition:invalid_login",
"node": "error-page",
"group": {
"reverse": true,
"filters": [
{
"field": "username",
"operator": "eq",
"value": "admin"
},
{
"field": "password",
"operator": "eq",
"value": "password"
}
]
}
}
}
}
}
},
{
"id": "error-page",
"node": "render-html",
"data": {
"mapping": {
"error_message": "eval.{{'Invalid login credentials.'}}",
"error_field": "eval.{{'username'}}",
"retry_suggested": "eval.{{true}}"
},
"additional_data": {
"template_file": "app/templates/error.html"
}
}
}
],
"edges": [
{
"source": "LoginForm",
"target": [ "ValidateLogin" ]
}
]
}

37
services/examples/json_email.go Normal file
View File

@@ -0,0 +1,37 @@
package main
import (
"context"
"encoding/json"
"fmt"
"os"
"github.com/oarkflow/mq"
"github.com/oarkflow/mq/dag"
"github.com/oarkflow/mq/handlers"
"github.com/oarkflow/mq/services"
)
func main() {
handlerBytes, err := os.ReadFile("json/login.json")
if err != nil {
panic(err)
}
var handler services.Handler
err = json.Unmarshal(handlerBytes, &handler)
if err != nil {
panic(err)
}
brokerAddr := ":8081"
flow := services.SetupHandler(handler, brokerAddr)
if flow.Error != nil {
fmt.Println("Error setting up handler:", flow.Error)
return
}
flow.Start(context.Background(), ":5000")
}
func init() {
dag.AddHandler("render-html", func(id string) mq.Processor { return handlers.NewRenderHTMLNode(id) })
dag.AddHandler("condition", func(id string) mq.Processor { return handlers.NewCondition(id) })
}

63
services/examples/login.json Normal file
View File

@@ -0,0 +1,63 @@
{
"type": "object",
"properties": {
"username": {
"type": "string",
"title": "Username or Email",
"order": 1,
"ui": {
"element": "input",
"type": "text",
"class": "form-group",
"name": "username",
"placeholder": "Enter your username or email"
}
},
"password": {
"type": "string",
"title": "Password",
"order": 2,
"ui": {
"element": "input",
"type": "password",
"class": "form-group",
"name": "password",
"placeholder": "Enter your password"
}
},
"remember_me": {
"type": "boolean",
"title": "Remember Me",
"order": 3,
"ui": {
"element": "input",
"type": "checkbox",
"class": "form-check",
"name": "remember_me"
}
}
},
"required": [ "username", "password" ],
"form": {
"class": "form-horizontal",
"action": "/process?task_id={{task_id}}&next=true",
"method": "POST",
"enctype": "application/x-www-form-urlencoded",
"groups": [
{
"title": "Login Credentials",
"fields": [ "username", "password", "remember_me" ]
}
],
"submit": {
"type": "submit",
"label": "Log In",
"class": "btn btn-primary"
},
"reset": {
"type": "reset",
"label": "Clear",
"class": "btn btn-secondary"
}
}
}

105
services/examples/schema.json Normal file
View File

@@ -0,0 +1,105 @@
{
"type": "object",
"properties": {
"first_name": {
"type": "string",
"title": "First Name",
"order": 1,
"ui": {
"element": "input",
"class": "form-group",
"name": "first_name"
}
},
"last_name": {
"type": "string",
"title": "Last Name",
"order": 2,
"ui": {
"element": "input",
"class": "form-group",
"name": "last_name"
}
},
"email": {
"type": "email",
"title": "Email Address",
"order": 3,
"ui": {
"element": "input",
"type": "email",
"class": "form-group",
"name": "email"
}
},
"user_type": {
"type": "string",
"title": "User Type",
"order": 4,
"ui": {
"element": "select",
"class": "form-group",
"name": "user_type",
"options": [ "new", "premium", "standard" ]
}
},
"priority": {
"type": "string",
"title": "Priority Level",
"order": 5,
"ui": {
"element": "select",
"class": "form-group",
"name": "priority",
"options": [ "low", "medium", "high", "urgent" ]
}
},
"subject": {
"type": "string",
"title": "Subject",
"order": 6,
"ui": {
"element": "input",
"class": "form-group",
"name": "subject"
}
},
"message": {
"type": "textarea",
"title": "Message",
"order": 7,
"ui": {
"element": "textarea",
"class": "form-group",
"name": "message"
}
}
},
"required": [ "first_name", "last_name", "email", "user_type", "priority", "subject", "message" ],
"form": {
"class": "form-horizontal",
"action": "/process?task_id={{task_id}}&next=true",
"method": "POST",
"enctype": "application/x-www-form-urlencoded",
"groups": [
{
"title": "User Information",
"fields": [ "first_name", "last_name", "email" ]
},
{
"title": "Ticket Details",
"fields": [ "user_type", "priority", "subject", "message" ]
}
],
"submit": {
"type": "submit",
"label": "Submit",
"class": "btn btn-primary"
},
"reset": {
"type": "reset",
"label": "Reset",
"class": "btn btn-secondary"
}
}
}

33
services/setup.go
View File

@@ -77,10 +77,17 @@ func SetupHandler(handler Handler, brokerAddr string, async ...bool) *dag.DAG {
return flow return flow
} }
type FilterGroup struct {
Operator string `json:"operator"`
Reverse bool `json:"reverse"`
Filters []*filters.Filter `json:"filters"`
}
type Filter struct { type Filter struct {
Filter *filters.Filter `json:"condition"` Filter *filters.Filter `json:"condition"`
Node string `json:"node"` FilterGroup *FilterGroup `json:"group"`
ID string `json:"id"` Node string `json:"node"`
ID string `json:"id"`
} }
func prepareNode(flow *dag.DAG, node Node) error { func prepareNode(flow *dag.DAG, node Node) error {
@@ -108,7 +115,25 @@ func prepareNode(flow *dag.DAG, node Node) error {
conditions := make(map[string]dag.Condition) conditions := make(map[string]dag.Condition)
for key, cond := range fil { for key, cond := range fil {
condition[key] = cond.Node condition[key] = cond.Node
conditions[key] = cond.Filter if cond.Filter != nil {
conditions[key] = cond.Filter
} else if cond.FilterGroup != nil {
cond.FilterGroup.Operator = strings.ToUpper(cond.FilterGroup.Operator)
if !slices.Contains([]string{"AND", "OR"}, cond.FilterGroup.Operator) {
cond.FilterGroup.Operator = "AND"
}
var fillers []filters.Condition
for _, f := range cond.FilterGroup.Filters {
if f != nil {
fillers = append(fillers, f)
}
}
conditions[key] = &filters.FilterGroup{
Operator: filters.Boolean(cond.FilterGroup.Operator),
Reverse: cond.FilterGroup.Reverse,
Filters: fillers,
}
}
} }
flow.AddCondition(node.ID, condition) flow.AddCondition(node.ID, condition)
nodeHandler.SetConditions(conditions) nodeHandler.SetConditions(conditions)