You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
102 lines
2.4 KiB
102 lines
2.4 KiB
package consumers
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"goInOut/config"
|
|
"goInOut/consumers/HTTP_PRPXY"
|
|
"goInOut/dbOperate"
|
|
"goInOut/utils"
|
|
"io"
|
|
"log"
|
|
"net/http"
|
|
)
|
|
|
|
type consumerHttpProxy struct {
|
|
//数据缓存管道
|
|
routes map[string]config.Router
|
|
//具体配置
|
|
Info HTTP_PRPXY.ConfigFile
|
|
InApiServer *dbOperate.ApiServerHelper
|
|
outHttpPost *dbOperate.HttpHelper
|
|
}
|
|
|
|
func (the *consumerHttpProxy) LoadConfigJson(cfgStr string) {
|
|
// 将 JSON 格式的数据解析到结构体中
|
|
err := json.Unmarshal([]byte(cfgStr), &the.Info)
|
|
if err != nil {
|
|
log.Printf("读取配置文件[%s]异常 err=%v", cfgStr, err.Error())
|
|
panic(err)
|
|
}
|
|
}
|
|
|
|
func (the *consumerHttpProxy) Initial(cfg string) error {
|
|
the.routes = map[string]config.Router{}
|
|
the.LoadConfigJson(cfg)
|
|
err := the.InputInitial()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
err = the.OutputInitial()
|
|
return err
|
|
}
|
|
func (the *consumerHttpProxy) InputInitial() error {
|
|
//数据入口
|
|
the.InApiServer = dbOperate.NewApiServer(
|
|
the.Info.IoConfig.In.ApiServer.Port,
|
|
the.Info.IoConfig.In.ApiServer.Routers,
|
|
)
|
|
the.InApiServer.Initial()
|
|
return nil
|
|
}
|
|
func (the *consumerHttpProxy) OutputInitial() error {
|
|
//数据出口
|
|
the.outHttpPost = &dbOperate.HttpHelper{
|
|
Url: the.Info.IoConfig.Out.HttpPost.Url,
|
|
}
|
|
the.outHttpPost.Initial()
|
|
return nil
|
|
}
|
|
func (the *consumerHttpProxy) Work() {
|
|
go func() {
|
|
for _, router := range the.Info.IoConfig.In.ApiServer.Routers {
|
|
the.InApiServer.RouteRegister(router.Router, the.onData)
|
|
the.routes[router.Router] = router
|
|
}
|
|
the.InApiServer.Run()
|
|
|
|
}()
|
|
}
|
|
func (the *consumerHttpProxy) onData(w http.ResponseWriter, r *http.Request) {
|
|
w.Header().Set("Content-Type", "application/json")
|
|
body, err := io.ReadAll(r.Body)
|
|
|
|
log.Printf("收到 %s 请求 %s", r.RequestURI, body)
|
|
|
|
bodyObj := map[string]any{}
|
|
err = json.Unmarshal(body, &bodyObj)
|
|
if err != nil {
|
|
log.Printf("body 解析失败,请检查 %s", err.Error())
|
|
return
|
|
}
|
|
routePath := fmt.Sprintf("%s %s", r.Method, r.RequestURI)
|
|
idTemplate := the.routes[routePath]
|
|
|
|
id, err := utils.TextTemplateMatch(bodyObj, idTemplate.IdTemplate)
|
|
if err != nil {
|
|
log.Printf("解析id 失败,请检查 %s", err.Error())
|
|
}
|
|
params := map[string]string{
|
|
"id": id,
|
|
}
|
|
resp := ""
|
|
//沿用请求路由
|
|
newUrl := the.outHttpPost.Url + r.URL.String()
|
|
resp, err = the.outHttpPost.HttpPostWithParams(newUrl, string(body), params)
|
|
if err != nil {
|
|
return
|
|
}
|
|
log.Printf("应答: %s", resp)
|
|
defer fmt.Fprintf(w, resp)
|
|
|
|
}
|
|
|