发消息测试通过

functions/queue_functions.go

@@ -55,7 +55,7 @@ func CreateQueue(rabbitFactory *rabbitmq.RabbitMQFactory, req types.QueueRequest
 }
 
 // BindQueue 绑定队列到交换机
-func BindQueue(rabbitFactory *rabbitmq.RabbitMQFactory, req types.BindRequest) *types.QueryResult {
+func BindQueue(rabbitFactory *rabbitmq.RabbitMQFactory, req types.QueueBindRequest) *types.QueryResult {
 	// 设置默认值
 	if req.ChannelName == "" {
 		req.ChannelName = "default"

main.go

@@ -101,7 +101,7 @@ func registerRabbitMQRoutes(webService web.Service, rabbitFactory *rabbitmq.Rabb
 	})))
 
 	// 发送原始消息（字节流）
-	webService.Handle("/api/rabbitmq/message/send-bytes", middleware.JWTAuthMiddleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+	webService.Handle("/api/rabbitmq/message/send/bytes", middleware.JWTAuthMiddleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		myhandle.QueryHandlerJson(w, r, rabbitFactory, functions.SendBytesMessage)
 	})))
 

test/my_create_exchange_test.go

@@ -8,34 +8,49 @@ import (
 	"net/http"
 	"testing"
 	"time"
+
+	"git.x2erp.com/qdy/go-base/types"
 )
 
-type ExchangeRequest struct {
-	ChannelName  string            `json:"channel_name"`
-	ExchangeName string            `json:"exchange_name"`
-	ExchangeType string            `json:"exchange_type"`
-	Durable      bool              `json:"durable"`
-	Internal     bool              `json:"internal"`
-	AutoDelete   bool              `json:"auto_delete"`
-	NoWait       bool              `json:"no_wait"`
-	Arguments    map[string]string `json:"arguments"`
-}
+// type ExchangeRequest struct {
+// 	ChannelName  string            `json:"channel_name"`
+// 	ExchangeName string            `json:"exchange_name"`
+// 	ExchangeType string            `json:"exchange_type"`
+// 	Durable      bool              `json:"durable"`
+// 	Internal     bool              `json:"internal"`
+// 	AutoDelete   bool              `json:"auto_delete"`
+// 	NoWait       bool              `json:"no_wait"`
+// 	Arguments    map[string]string `json:"arguments"`
+// }
 
 func TestCreateexchange(t *testing.T) {
 	// 1. 准备请求数据
-	requestData := ExchangeRequest{
-		ChannelName:  "test_channel",
-		ExchangeName: "test_exchange",
-		ExchangeType: "direct", // direct, fanout, topic, headers
+	requestData := types.ExchangeRequest{
+		ChannelName:  "v_bdx_channel",
+		ExchangeName: "v_bdx_exchange",
+		ExchangeType: "topic", // direct, fanout, topic, headers
 		Durable:      true,
 		Internal:     false,
 		AutoDelete:   false,
 		NoWait:       false,
 		Arguments: map[string]string{
-			"x-delayed-type": "direct",
+			"x-delayed-type": "topic",
 		},
 	}
 
+	// 消息队列
+	queueRequest := types.QueueRequest{
+		QueueName:   "vbdx_queue_count",
+		ChannelName: requestData.ChannelName,
+	}
+
+	queueBindRequest := types.QueueBindRequest{
+		ChannelName:  requestData.ChannelName,
+		QueueName:    queueRequest.QueueName,
+		ExchangeName: requestData.ExchangeName,
+		RoutingKey:   "v-bdx.#",
+	}
+
 	// 2. 将数据转换为JSON
 	jsonData, err := json.Marshal(requestData)
 	if err != nil {
@@ -46,8 +61,47 @@ func TestCreateexchange(t *testing.T) {
 	// 3. 发送HTTP POST请求
 	url := "http://localhost:9090/api/rabbitmq/exchange/create"
 
+	POST(jsonData, url)
+
+	//建立队列
+	CreateQueue(queueRequest)
+
+	//绑定
+	CreateQueueBind(queueBindRequest)
+}
+
+func CreateQueue(queueRequest types.QueueRequest) {
+
+	// 2. 将数据转换为JSON
+	jsonData, err := json.Marshal(queueRequest)
+	if err != nil {
+		fmt.Printf("JSON编码失败: %v\n", err)
+		return
+	}
+	url := "http://localhost:9090/api/rabbitmq/queue/create"
+
+	POST(jsonData, url)
+
+}
+
+func CreateQueueBind(queueBindRequest types.QueueBindRequest) {
+
+	// 2. 将数据转换为JSON
+	jsonData, err := json.Marshal(queueBindRequest)
+	if err != nil {
+		fmt.Printf("JSON编码失败: %v\n", err)
+		return
+	}
+	url := "http://localhost:9090/api/rabbitmq/queue/bind"
+
+	POST(jsonData, url)
+
+}
+
+func POST(jsonData []byte, url string) {
+
 	// 如果服务端需要认证，添加JWT token
-	token := "your-jwt-token-here" // 替换为实际的JWT token
+	token := "123" // 替换为实际的JWT token
 
 	req, err := http.NewRequest("POST", url, bytes.NewBuffer(jsonData))
 	if err != nil {

test/my_send_message_test.go

@@ -0,0 +1,98 @@
+package test
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"testing"
+	"time"
+
+	"git.x2erp.com/qdy/go-base/types"
+)
+
+func TestSendMessage(t *testing.T) {
+
+	queryRequest := getQueyRequest()
+
+	messageRequest := types.MessageRequest{
+		ChannelName:  "v_bdx_channel",
+		ExchangeName: "v_bdx_exchange",
+		RoutingKey:   "v-bdx.count",
+		Message:      queryRequest,
+		ContentType:  "json",
+	}
+
+	// 2. 将数据转换为JSON
+	jsonData, err := json.Marshal(messageRequest)
+	if err != nil {
+		fmt.Printf("JSON编码失败: %v\n", err)
+		return
+	}
+
+	// 3. 发送HTTP POST请求
+	url := "http://localhost:9090/api/rabbitmq/message/send"
+
+	POSTMessage(jsonData, url)
+
+}
+
+func getQueyRequest() types.QueryRequest {
+	sql := `SELECT * FROM (SELECT a.*, ROWNUM rn FROM (SELECT CLOTHING_ID, CLOTHING_NAME FROM X6_STOCK_DEV.A3_CLOTHING ORDER BY CLOTHING_ID) a WHERE ROWNUM <= :1) WHERE rn > :2
+	`
+
+	// // 3个参数
+	params := []interface{}{
+		10,
+		0,
+	}
+
+	return types.QueryRequest{
+		SQL:              sql,
+		PositionalParams: params,
+		WriterHeader:     false,
+	}
+}
+
+func POSTMessage(jsonData []byte, url string) {
+
+	// 如果服务端需要认证，添加JWT token
+	token := "123" // 替换为实际的JWT token
+
+	req, err := http.NewRequest("POST", url, bytes.NewBuffer(jsonData))
+	if err != nil {
+		fmt.Printf("创建请求失败: %v\n", err)
+		return
+	}
+
+	// 添加请求头
+	req.Header.Set("Content-Type", "application/json")
+	if token != "" {
+		req.Header.Set("Authorization", "Bearer "+token)
+	}
+
+	// 4. 发送请求并获取响应
+	client := &http.Client{
+		Timeout: 10 * time.Second,
+	}
+
+	resp, err := client.Do(req)
+	if err != nil {
+		fmt.Printf("请求失败: %v\n", err)
+		return
+	}
+	defer resp.Body.Close()
+
+	// 5. 读取响应
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		fmt.Printf("读取响应失败: %v\n", err)
+		return
+	}
+
+	// 6. 输出结果
+	fmt.Printf("状态码: %d\n", resp.StatusCode)
+	fmt.Printf("响应头: %v\n", resp.Header)
+	fmt.Printf("响应体: %s\n", string(body))
+}