fixed kafka message queue non-existent

0987363 · Dec 19, 2023 · f121542 · f121542
1 parent d073d0f
commit f121542
Showing 1 changed file with 4 additions and 1 deletion.
diff --git a/server/rpc_router.go b/server/rpc_router.go
@@ -527,9 +527,12 @@ func (router *router) ProcessMessage(ctx context.Context, msg Message) (err erro
 
 	// we may have multiple subscribers for the topic
 	for _, sub := range subs {
+		fmt.Println(msg.Header())
+		fmt.Println(sub.opts)
+
 		// we may have multiple handlers per subscriber
 		for i := 0; i < len(sub.handlers); i++ {
-			if sub.opts.Queue != msg.Header()["Queue"] {
+			if msg.Header()["Queue"] != "" && sub.opts.Queue != msg.Header()["Queue"] {
 				continue
 			}