go-svc-mqconsumer/handlers/http_handler.go

package handlers

import (
	"bytes"
	"encoding/json"
	"fmt"
	"net/http"
	"time"

	"git.x2erp.com/qdy/go-base/types"
)

// HTTPHandler HTTP处理器实现MessageHandler接口
type HTTPHandler struct {
	client *http.Client
}

// NewHTTPHandler 创建HTTP处理器
func NewHTTPHandler() *HTTPHandler {
	return &HTTPHandler{
		client: &http.Client{
			Timeout: 30 * time.Second,
			Transport: &http.Transport{
				MaxIdleConns:        100,
				MaxIdleConnsPerHost: 100,
				IdleConnTimeout:     90 * time.Second,
			},
		},
	}
}

// Process 处理消息，发送HTTP POST请求
func (h *HTTPHandler) Process(queueRequest *types.QueueRequest, body []byte) types.QueryResult {
	if queueRequest.Url == "" {
		return types.QueryResult{
			Success: false,
			Message: "URL is empty",
		}
	}

	// 准备请求体
	requestBody := map[string]interface{}{
		"queue":     queueRequest.QueueName,
		"data":      string(body),
		"timestamp": time.Now().Format(time.RFC3339),
	}

	jsonBody, err := json.Marshal(requestBody)
	if err != nil {
		return types.QueryResult{
			Success: false,
			Message: fmt.Sprintf("Failed to marshal request: %v", err),
		}
	}

	// 发送HTTP POST请求
	resp, err := h.client.Post(queueRequest.Url, "application/json", bytes.NewBuffer(jsonBody))
	if err != nil {
		return types.QueryResult{
			Success: false,
			Message: fmt.Sprintf("HTTP request failed: %v", err),
		}
	}
	defer resp.Body.Close()

	if resp.StatusCode >= 200 && resp.StatusCode < 300 {
		return types.QueryResult{
			Success: true,
			Message: "Message processed successfully",
		}
	}

	return types.QueryResult{
		Success: false,
		Message: fmt.Sprintf("HTTP error: %s", resp.Status),
	}
}