From 315ad19dd4afb5c4c75498334d99db4c793f1235 Mon Sep 17 00:00:00 2001
From: laodaming <11058467+laudamine@user.noreply.gitee.com>
Date: Mon, 7 Aug 2023 11:44:54 +0800
Subject: [PATCH] fix

---
 initalize/rabbitmq.go                          |  7 +++----
 server/render/consumer/assemble_render_data.go | 12 ++++++++++++
 server/render/etc/render.yaml                  |  3 ++-
 server/render/internal/config/config.go        |  6 +++---
 server/render/internal/svc/servicecontext.go   |  4 +++-
 server/render/render.go                        |  9 ++++++++-
 server/websocket/consumer/consumer.go          |  4 ----
 utils/mq_consumer_factory/mq.go                |  6 ++++++
 8 files changed, 37 insertions(+), 14 deletions(-)
 create mode 100644 server/render/consumer/assemble_render_data.go
 create mode 100644 utils/mq_consumer_factory/mq.go

diff --git a/initalize/rabbitmq.go b/initalize/rabbitmq.go
index 7fd3fd06..004524c0 100644
--- a/initalize/rabbitmq.go
+++ b/initalize/rabbitmq.go
@@ -5,7 +5,7 @@ import (
 	"crypto/tls"
 	"errors"
 	"fusenapi/constants"
-	"fusenapi/server/websocket/consumer"
+	"fusenapi/utils/mq_consumer_factory"
 	"github.com/streadway/amqp"
 	"github.com/zeromicro/go-zero/core/logx"
 	"log"
@@ -84,10 +84,10 @@ func (h *RabbitMqHandle) SendMsg(queueName constants.RABBIT_MQ, message []byte)
 }
 
 // 消费消息
-func (h *RabbitMqHandle) Consume(ctx context.Context, queueName constants.RABBIT_MQ, handle consumer.MqHandle) error {
+func (h *RabbitMqHandle) Consume(ctx context.Context, queueName constants.RABBIT_MQ, handle mq_consumer_factory.MqHandle) {
 	object, ok := mapMq[queueName]
 	if !ok {
-		return errors.New("unknown queue")
+		panic("unknown queue")
 	}
 	go func() {
 		select {
@@ -135,5 +135,4 @@ func (h *RabbitMqHandle) Consume(ctx context.Context, queueName constants.RABBIT
 		}(msg)
 	}
 	wait.Wait()
-	return nil
 }
diff --git a/server/render/consumer/assemble_render_data.go b/server/render/consumer/assemble_render_data.go
new file mode 100644
index 00000000..9f3b6187
--- /dev/null
+++ b/server/render/consumer/assemble_render_data.go
@@ -0,0 +1,12 @@
+package consumer
+
+import "fmt"
+
+// 消费渲染需要组装的数据
+type MqConsumerRenderAssemble struct {
+}
+
+func (m *MqConsumerRenderAssemble) Run(data []byte) error {
+	fmt.Println("收到消息:" + string(data))
+	return nil
+}
diff --git a/server/render/etc/render.yaml b/server/render/etc/render.yaml
index 2dba7dc7..30b4cdd7 100644
--- a/server/render/etc/render.yaml
+++ b/server/render/etc/render.yaml
@@ -5,4 +5,5 @@ SourceMysql: fusentest:XErSYmLELKMnf3Dh@tcp(110.41.19.98:3306)/fusentest
 Auth:
     AccessSecret: fusen2023
     AccessExpire: 2592000
-    RefreshAfter: 1592000
\ No newline at end of file
+    RefreshAfter: 1592000
+SourceRabbitMq: amqp://rabbit001:rabbit001129@110.41.19.98:5672
\ No newline at end of file
diff --git a/server/render/internal/config/config.go b/server/render/internal/config/config.go
index 5f9c4816..fa950a01 100644
--- a/server/render/internal/config/config.go
+++ b/server/render/internal/config/config.go
@@ -2,12 +2,12 @@ package config
 
 import (
 	"fusenapi/server/render/internal/types"
-
 	"github.com/zeromicro/go-zero/rest"
 )
 
 type Config struct {
 	rest.RestConf
-	SourceMysql string
-	Auth        types.Auth
+	SourceMysql    string
+	Auth           types.Auth
+	SourceRabbitMq string
 }
diff --git a/server/render/internal/svc/servicecontext.go b/server/render/internal/svc/servicecontext.go
index bd083461..79e04e0d 100644
--- a/server/render/internal/svc/servicecontext.go
+++ b/server/render/internal/svc/servicecontext.go
@@ -18,14 +18,16 @@ type ServiceContext struct {
 
 	MysqlConn *gorm.DB
 	AllModels *gmodel.AllModelsGen
+	RabbitMq  *initalize.RabbitMqHandle
 }
 
 func NewServiceContext(c config.Config) *ServiceContext {
-
+	initalize.InitRabbitMq(c.SourceRabbitMq, nil)
 	return &ServiceContext{
 		Config:    c,
 		MysqlConn: initalize.InitMysql(c.SourceMysql),
 		AllModels: gmodel.NewAllModels(initalize.InitMysql(c.SourceMysql)),
+		RabbitMq:  initalize.InitRabbitMq(c.SourceRabbitMq, nil),
 	}
 }
 
diff --git a/server/render/render.go b/server/render/render.go
index af3b93f1..2c052f37 100644
--- a/server/render/render.go
+++ b/server/render/render.go
@@ -1,8 +1,11 @@
 package main
 
 import (
+	"context"
 	"flag"
 	"fmt"
+	"fusenapi/constants"
+	"fusenapi/server/render/consumer"
 	"net/http"
 	"time"
 
@@ -30,7 +33,11 @@ func main() {
 
 	ctx := svc.NewServiceContext(c)
 	handler.RegisterHandlers(server, ctx)
-
+	//消费渲染前组装数据队列
+	ctx1 := context.Background()
+	ctx2, cancel := context.WithCancel(ctx1)
+	defer cancel()
+	go ctx.RabbitMq.Consume(ctx2, constants.RABBIT_MQ_ASSEMBLE_RENDER_DATA, &consumer.MqConsumerRenderAssemble{})
 	fmt.Printf("Starting server at %s:%d...\n", c.Host, c.Port)
 	server.Start()
 }
diff --git a/server/websocket/consumer/consumer.go b/server/websocket/consumer/consumer.go
index 69363bf0..16a7c8a9 100644
--- a/server/websocket/consumer/consumer.go
+++ b/server/websocket/consumer/consumer.go
@@ -2,10 +2,6 @@ package consumer
 
 import "fmt"
 
-type MqHandle interface {
-	Run(data []byte) error
-}
-
 // 消费渲染结果数据
 type MqConsumerRenderResult struct {
 }
diff --git a/utils/mq_consumer_factory/mq.go b/utils/mq_consumer_factory/mq.go
new file mode 100644
index 00000000..0f3bc97d
--- /dev/null
+++ b/utils/mq_consumer_factory/mq.go
@@ -0,0 +1,6 @@
+package mq_consumer_factory
+
+// 消费mq消息要实现对应Run方法
+type MqHandle interface {
+	Run(data []byte) error
+}