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

feat: separate broker

Browse Source
This commit is contained in:
Oarkflow
2024-10-09 11:29:06 +05:45
parent 9c8712994d
commit aa324f4435
1 changed files with 77 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

127
README.md
View File

@@ -28,45 +28,63 @@ import (
"context" "context"
"encoding/json" "encoding/json"
"fmt" "fmt"
"log"
"github.com/oarkflow/mq" "github.com/oarkflow/mq"
) )
func Node1(ctx context.Context, task mq.Task) mq.Result { func Node1(ctx context.Context, task *mq.Task) mq.Result {
fmt.Println("Processing queue1") return mq.Result{Payload: task.Payload, TaskID: task.ID}
return mq.Result{Payload: task.Payload, MessageID: task.ID}
} }
func Node2(ctx context.Context, task mq.Task) mq.Result { func Node2(ctx context.Context, task *mq.Task) mq.Result {
return mq.Result{Payload: task.Payload, MessageID: task.ID} return mq.Result{Payload: task.Payload, TaskID: task.ID}
} }
func Node3(ctx context.Context, task mq.Task) mq.Result { func Node3(ctx context.Context, task *mq.Task) mq.Result {
var data map[string]any var user map[string]any
err := json.Unmarshal(task.Payload, &data) json.Unmarshal(task.Payload, &user)
if err != nil { age := int(user["age"].(float64))
return mq.Result{Error: err} status := "FAIL"
if age > 20 {
status = "PASS"
} }
data["salary"] = fmt.Sprintf("12000%v", data["user_id"]) user["status"] = status
bt, _ := json.Marshal(data) resultPayload, _ := json.Marshal(user)
return mq.Result{Payload: bt, MessageID: task.ID} return mq.Result{Payload: resultPayload, Status: status}
} }
func Node4(ctx context.Context, task mq.Task) mq.Result { func Node4(ctx context.Context, task *mq.Task) mq.Result {
var data []map[string]any var user map[string]any
err := json.Unmarshal(task.Payload, &data) json.Unmarshal(task.Payload, &user)
if err != nil { user["final"] = "D"
return mq.Result{Error: err} resultPayload, _ := json.Marshal(user)
return mq.Result{Payload: resultPayload}
} }
payload := map[string]any{"storage": data}
bt, _ := json.Marshal(payload) func Node5(ctx context.Context, task *mq.Task) mq.Result {
return mq.Result{Payload: bt, MessageID: task.ID} var user map[string]any
json.Unmarshal(task.Payload, &user)
user["salary"] = "E"
resultPayload, _ := json.Marshal(user)
return mq.Result{Payload: resultPayload}
}
func Node6(ctx context.Context, task *mq.Task) mq.Result {
var user map[string]any
json.Unmarshal(task.Payload, &user)
resultPayload, _ := json.Marshal(map[string]any{"storage": user})
return mq.Result{Payload: resultPayload}
} }
func Callback(ctx context.Context, task mq.Result) mq.Result { func Callback(ctx context.Context, task mq.Result) mq.Result {
fmt.Println("Received task", task.MessageID, "Payload", string(task.Payload), task.Error, task.Queue) fmt.Println("Received task", task.TaskID, "Payload", string(task.Payload), task.Error, task.Topic)
return mq.Result{} return mq.Result{}
} }
func NotifyResponse(ctx context.Context, result mq.Result) {
log.Printf("DAG Final response: TaskID: %s, Payload: %s, Topic: %s", result.TaskID, result.Payload, result.Topic)
}
``` ```
## Start Server ## Start Server
@@ -102,14 +120,16 @@ import (
"context" "context"
"github.com/oarkflow/mq" "github.com/oarkflow/mq"
"github.com/oarkflow/mq/examples/tasks" "github.com/oarkflow/mq/examples/tasks"
) )
func main() { func main() {
consumer := mq.NewConsumer("consumer-1") consumer1 := mq.NewConsumer("consumer-1", "queue1", tasks.Node1)
consumer.RegisterHandler("queue1", tasks.Node1) consumer2 := mq.NewConsumer("consumer-2", "queue2", tasks.Node2)
consumer.RegisterHandler("queue2", tasks.Node2) // consumer := mq.NewConsumer("consumer-1", mq.WithTLS(true, "./certs/server.crt", "./certs/server.key"))
consumer.Consume(context.Background()) go consumer1.Consume(context.Background())
consumer2.Consume(context.Background())
} }
``` ```
@@ -164,38 +184,38 @@ package main
import ( import (
"context" "context"
"encoding/json" "encoding/json"
"github.com/oarkflow/mq/consts"
"github.com/oarkflow/mq/examples/tasks"
"io" "io"
"log"
"net/http" "net/http"
"github.com/oarkflow/mq" "github.com/oarkflow/mq"
"github.com/oarkflow/mq/dag" "github.com/oarkflow/mq/dag"
"github.com/oarkflow/mq/examples/tasks"
) )
var d *dag.DAG var (
d = dag.NewDAG(mq.WithSyncMode(false), mq.WithNotifyResponse(tasks.NotifyResponse))
// d = dag.NewDAG(mq.WithSyncMode(true), mq.WithTLS(true, "./certs/server.crt", "./certs/server.key"), mq.WithCAPath("./certs/ca.cert"))
)
func main() { func main() {
d = dag.New() d.AddNode("A", tasks.Node1, true)
d.AddNode("queue1", tasks.Node1) d.AddNode("B", tasks.Node2)
d.AddNode("queue2", tasks.Node2) d.AddNode("C", tasks.Node3)
d.AddNode("queue3", tasks.Node3) d.AddNode("D", tasks.Node4)
d.AddNode("queue4", tasks.Node4) d.AddNode("E", tasks.Node5)
d.AddNode("F", tasks.Node6)
d.AddEdge("queue1", "queue2") d.AddEdge("A", "B", dag.LoopEdge)
d.AddLoop("queue2", "queue3") d.AddCondition("C", map[string]string{"PASS": "D", "FAIL": "E"})
d.AddEdge("queue2", "queue4") d.AddEdge("B", "C")
d.Prepare() d.AddEdge("D", "F")
go func() { d.AddEdge("E", "F")
err := d.Start(context.TODO()) http.HandleFunc("POST /publish", requestHandler("publish"))
http.HandleFunc("POST /request", requestHandler("request"))
err := d.Start(context.TODO(), ":8083")
if err != nil { if err != nil {
panic(err) panic(err)
} }
}()
http.HandleFunc("/publish", requestHandler("publish"))
http.HandleFunc("/request", requestHandler("request"))
log.Println("HTTP server started on http://localhost:8083")
log.Fatal(http.ListenAndServe(":8083", nil))
} }
func requestHandler(requestType string) func(w http.ResponseWriter, r *http.Request) { func requestHandler(requestType string) func(w http.ResponseWriter, r *http.Request) {
@@ -217,19 +237,14 @@ func requestHandler(requestType string) func(w http.ResponseWriter, r *http.Requ
http.Error(w, "Empty request body", http.StatusBadRequest) http.Error(w, "Empty request body", http.StatusBadRequest)
return return
} }
var rs mq.Result ctx := context.Background()
if requestType == "request" { if requestType == "request" {
rs = d.Request(context.Background(), payload) ctx = mq.SetHeaders(ctx, map[string]string{consts.AwaitResponseKey: "true"})
} else {
rs = d.Send(context.Background(), payload)
} }
// ctx = context.WithValue(ctx, "initial_node", "E")
rs := d.ProcessTask(ctx, payload)
w.Header().Set("Content-Type", "application/json") w.Header().Set("Content-Type", "application/json")
result := map[string]any{ json.NewEncoder(w).Encode(rs)
"message_id": rs.MessageID,
"payload": string(rs.Payload),
"error": rs.Error,
}
json.NewEncoder(w).Encode(result)
} }
} }
``` ```