268 lines
8.1 KiB
Go
268 lines
8.1 KiB
Go
package app
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"log"
|
|
"net/http"
|
|
"net/url"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"lingniu/vehicle-data-platform/apps/api/internal/config"
|
|
"lingniu/vehicle-data-platform/apps/api/internal/httpx"
|
|
"lingniu/vehicle-data-platform/apps/api/internal/platform"
|
|
"lingniu/vehicle-data-platform/apps/api/internal/static"
|
|
)
|
|
|
|
func NewServer(cfg config.Config) http.Handler {
|
|
var store platform.Store = platform.NewMockStore()
|
|
if cfg.MySQLDSN != "" {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
db, err := platform.OpenSQL(ctx, "mysql", cfg.MySQLDSN)
|
|
if err != nil {
|
|
log.Printf("production mysql store disabled: %v", err)
|
|
} else {
|
|
var tdengine *sql.DB
|
|
if cfg.TDengineDSN != "" {
|
|
tdengine, err = platform.OpenSQL(ctx, cfg.TDengineDriver, cfg.TDengineDSN)
|
|
if err != nil {
|
|
log.Printf("production tdengine store disabled: %v", err)
|
|
} else {
|
|
log.Printf("production tdengine store enabled")
|
|
}
|
|
}
|
|
productionStore := platform.NewProductionStore(db, tdengine, cfg.TDengineDatabase)
|
|
if cfg.RedisAddr != "" {
|
|
productionStore.WithRedisOnlineKeyCounter(platform.NewRedisOnlineKeyCounter(cfg.RedisAddr, cfg.RedisUsername, cfg.RedisPassword, cfg.RedisDB))
|
|
log.Printf("production redis health probe enabled")
|
|
}
|
|
if cfg.CapacityCheckBin != "" {
|
|
productionStore.WithCapacityChecker(platform.NewCapacityCheckCommand(cfg.CapacityCheckBin))
|
|
log.Printf("production capacity-check probe enabled")
|
|
}
|
|
store = productionStore
|
|
log.Printf("production mysql store enabled")
|
|
}
|
|
}
|
|
api := platform.NewHandler(platform.NewServiceWithRuntime(store, platform.RuntimeInfo{
|
|
RequestTimeoutMs: int(cfg.RequestTimeout / time.Millisecond),
|
|
AMapWebJSConfigured: strings.TrimSpace(cfg.AMapWebJSKey) != "",
|
|
AMapSecurityProxyEnabled: strings.TrimSpace(cfg.AMapSecurityCode) != "" && strings.TrimSpace(cfg.AMapServiceHost) != "",
|
|
AMapSecurityCodeExposed: exposedAMapSecurityCode(cfg) != "",
|
|
AMapSecurityServiceHost: strings.TrimSpace(cfg.AMapServiceHost),
|
|
}))
|
|
handler := static.Handler(cfg.StaticDir, api)
|
|
handler = withAppConfig(handler, cfg)
|
|
handler = withAMapSecurityProxy(handler, cfg, defaultAMapProxyUpstreams(), http.DefaultClient)
|
|
return withRequestTimeout(handler, cfg.RequestTimeout)
|
|
}
|
|
|
|
func withAppConfig(next http.Handler, cfg config.Config) http.Handler {
|
|
type appConfig struct {
|
|
AMapWebJSKey string `json:"amapWebJsKey,omitempty"`
|
|
AMapSecurityCode string `json:"amapSecurityJsCode,omitempty"`
|
|
AMapServiceHost string `json:"amapSecurityServiceHost,omitempty"`
|
|
}
|
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
if r.URL.Path != "/app-config.js" {
|
|
next.ServeHTTP(w, r)
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "application/javascript; charset=utf-8")
|
|
w.Header().Set("Cache-Control", "no-store")
|
|
body, err := json.Marshal(appConfig{
|
|
AMapWebJSKey: cfg.AMapWebJSKey,
|
|
AMapSecurityCode: exposedAMapSecurityCode(cfg),
|
|
AMapServiceHost: cfg.AMapServiceHost,
|
|
})
|
|
if err != nil {
|
|
http.Error(w, "failed to render app config", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
_, _ = fmt.Fprintf(w, "window.__LINGNIU_APP_CONFIG__=%s;\n", body)
|
|
})
|
|
}
|
|
|
|
func exposedAMapSecurityCode(cfg config.Config) string {
|
|
if cfg.AMapServiceHost != "" {
|
|
return ""
|
|
}
|
|
return cfg.AMapSecurityCode
|
|
}
|
|
|
|
type amapProxyUpstreams struct {
|
|
RestAPI string
|
|
WebAPI string
|
|
FMap string
|
|
}
|
|
|
|
func defaultAMapProxyUpstreams() amapProxyUpstreams {
|
|
return amapProxyUpstreams{
|
|
RestAPI: "https://restapi.amap.com",
|
|
WebAPI: "https://webapi.amap.com",
|
|
FMap: "https://fmap01.amap.com",
|
|
}
|
|
}
|
|
|
|
func (upstreams amapProxyUpstreams) normalized() amapProxyUpstreams {
|
|
defaults := defaultAMapProxyUpstreams()
|
|
if upstreams.RestAPI == "" {
|
|
upstreams.RestAPI = defaults.RestAPI
|
|
}
|
|
if upstreams.WebAPI == "" {
|
|
upstreams.WebAPI = defaults.WebAPI
|
|
}
|
|
if upstreams.FMap == "" {
|
|
upstreams.FMap = defaults.FMap
|
|
}
|
|
upstreams.RestAPI = strings.TrimRight(upstreams.RestAPI, "/")
|
|
upstreams.WebAPI = strings.TrimRight(upstreams.WebAPI, "/")
|
|
upstreams.FMap = strings.TrimRight(upstreams.FMap, "/")
|
|
return upstreams
|
|
}
|
|
|
|
func withAMapSecurityProxy(next http.Handler, cfg config.Config, upstreams amapProxyUpstreams, client *http.Client) http.Handler {
|
|
serviceHost := strings.TrimRight(cfg.AMapServiceHost, "/")
|
|
securityCode := strings.TrimSpace(cfg.AMapSecurityCode)
|
|
if serviceHost == "" || securityCode == "" {
|
|
return next
|
|
}
|
|
if client == nil {
|
|
client = http.DefaultClient
|
|
}
|
|
upstreams = upstreams.normalized()
|
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
if r.URL.Path != serviceHost && !strings.HasPrefix(r.URL.Path, serviceHost+"/") {
|
|
next.ServeHTTP(w, r)
|
|
return
|
|
}
|
|
target, err := amapProxyURL(upstreams, serviceHost, securityCode, r)
|
|
if err != nil {
|
|
httpx.WriteError(w, http.StatusBadGateway, "AMAP_PROXY_URL_INVALID", "高德地图代理地址无效", err.Error(), requestTraceID(r))
|
|
return
|
|
}
|
|
req, err := http.NewRequestWithContext(r.Context(), r.Method, target, r.Body)
|
|
if err != nil {
|
|
httpx.WriteError(w, http.StatusBadGateway, "AMAP_PROXY_REQUEST_INVALID", "高德地图代理请求无效", err.Error(), requestTraceID(r))
|
|
return
|
|
}
|
|
req.Header = r.Header.Clone()
|
|
req.Host = ""
|
|
resp, err := client.Do(req)
|
|
if err != nil {
|
|
httpx.WriteError(w, http.StatusBadGateway, "AMAP_PROXY_FAILED", "高德地图代理请求失败", err.Error(), requestTraceID(r))
|
|
return
|
|
}
|
|
defer resp.Body.Close()
|
|
for key, values := range resp.Header {
|
|
for _, value := range values {
|
|
w.Header().Add(key, value)
|
|
}
|
|
}
|
|
w.WriteHeader(resp.StatusCode)
|
|
_, _ = io.Copy(w, resp.Body)
|
|
})
|
|
}
|
|
|
|
func amapProxyURL(upstreams amapProxyUpstreams, serviceHost string, securityCode string, r *http.Request) (string, error) {
|
|
targetPath := strings.TrimPrefix(r.URL.Path, serviceHost)
|
|
if targetPath == "" {
|
|
targetPath = "/"
|
|
}
|
|
upstreamBase := upstreams.RestAPI
|
|
if strings.HasPrefix(targetPath, "/v4/map/styles") {
|
|
upstreamBase = upstreams.WebAPI
|
|
} else if strings.HasPrefix(targetPath, "/v3/vectormap") {
|
|
upstreamBase = upstreams.FMap
|
|
}
|
|
base, err := url.Parse(upstreamBase)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
base.Path = strings.TrimRight(base.Path, "/") + targetPath
|
|
query := r.URL.Query()
|
|
query.Set("jscode", securityCode)
|
|
base.RawQuery = query.Encode()
|
|
return base.String(), nil
|
|
}
|
|
|
|
func withRequestTimeout(next http.Handler, timeout time.Duration) http.Handler {
|
|
if timeout <= 0 {
|
|
return next
|
|
}
|
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
ctx, cancel := context.WithTimeout(r.Context(), timeout)
|
|
defer cancel()
|
|
buffered := newBufferedResponseWriter()
|
|
result := make(chan any, 1)
|
|
go func() {
|
|
defer func() {
|
|
result <- recover()
|
|
}()
|
|
next.ServeHTTP(buffered, r.WithContext(ctx))
|
|
}()
|
|
select {
|
|
case panicValue := <-result:
|
|
if panicValue != nil {
|
|
panic(panicValue)
|
|
}
|
|
buffered.writeTo(w)
|
|
case <-ctx.Done():
|
|
httpx.WriteError(w, http.StatusServiceUnavailable, "REQUEST_TIMEOUT", "请求处理超时", "", requestTraceID(r))
|
|
}
|
|
})
|
|
}
|
|
|
|
func requestTraceID(r *http.Request) string {
|
|
if value := r.Header.Get("X-Trace-Id"); value != "" {
|
|
return value
|
|
}
|
|
return "trace-" + time.Now().Format("20060102150405.000000")
|
|
}
|
|
|
|
type bufferedResponseWriter struct {
|
|
mu sync.Mutex
|
|
header http.Header
|
|
status int
|
|
body bytes.Buffer
|
|
}
|
|
|
|
func newBufferedResponseWriter() *bufferedResponseWriter {
|
|
return &bufferedResponseWriter{header: http.Header{}, status: http.StatusOK}
|
|
}
|
|
|
|
func (w *bufferedResponseWriter) Header() http.Header {
|
|
return w.header
|
|
}
|
|
|
|
func (w *bufferedResponseWriter) WriteHeader(statusCode int) {
|
|
w.mu.Lock()
|
|
defer w.mu.Unlock()
|
|
w.status = statusCode
|
|
}
|
|
|
|
func (w *bufferedResponseWriter) Write(body []byte) (int, error) {
|
|
w.mu.Lock()
|
|
defer w.mu.Unlock()
|
|
return w.body.Write(body)
|
|
}
|
|
|
|
func (w *bufferedResponseWriter) writeTo(target http.ResponseWriter) {
|
|
w.mu.Lock()
|
|
defer w.mu.Unlock()
|
|
for key, values := range w.header {
|
|
for _, value := range values {
|
|
target.Header().Add(key, value)
|
|
}
|
|
}
|
|
target.WriteHeader(w.status)
|
|
_, _ = target.Write(w.body.Bytes())
|
|
}
|