106 lines
2.5 KiB
Go
106 lines
2.5 KiB
Go
package upstream
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
func (m *Manager) post(ctx context.Context, path string, payload any) error {
|
|
client := m.HTTPClient
|
|
if client == nil {
|
|
client = &http.Client{Timeout: defaultHTTPTimeout}
|
|
}
|
|
return postJSON(ctx, client, m.APIBaseURL, path, payload)
|
|
}
|
|
|
|
func (p *connectionPool) reportState(ctx context.Context, status string, stateErr error) error {
|
|
if p.reporter == nil {
|
|
return nil
|
|
}
|
|
return p.reporter(ctx, p.snapshotState(status, stateErr))
|
|
}
|
|
|
|
func (p *connectionPool) snapshotState(status string, stateErr error) ConnectionState {
|
|
now := time.Now().UTC().Format(time.RFC3339Nano)
|
|
state := ConnectionState{
|
|
ChannelID: p.channelID,
|
|
ConnectionID: p.connectionID,
|
|
Status: status,
|
|
DesiredConnections: p.config.DesiredConnections,
|
|
CurrentConnections: p.countActiveConnections(),
|
|
}
|
|
p.mu.Lock()
|
|
state.ReconnectCount = p.reconnectCount
|
|
state.LastErrorCategory = p.lastErrorCategory
|
|
if !p.lastReconnectAttemptAt.IsZero() {
|
|
state.LastReconnectAttemptAt = p.lastReconnectAttemptAt.Format(time.RFC3339Nano)
|
|
}
|
|
if !p.nextReconnectAt.IsZero() {
|
|
state.NextReconnectAt = p.nextReconnectAt.Format(time.RFC3339Nano)
|
|
}
|
|
p.mu.Unlock()
|
|
if state.DesiredConnections <= 0 {
|
|
state.DesiredConnections = 1
|
|
}
|
|
switch status {
|
|
case "connected":
|
|
state.LastConnectedAt = now
|
|
state.LastHeartbeatAt = now
|
|
case "heartbeat":
|
|
state.LastHeartbeatAt = now
|
|
case "disconnected", "failed":
|
|
state.LastDisconnectedAt = now
|
|
}
|
|
if stateErr != nil {
|
|
state.LastError = stateErr.Error()
|
|
}
|
|
return state
|
|
}
|
|
|
|
func postJSON(ctx context.Context, client *http.Client, apiBaseURL string, path string, payload any) error {
|
|
if client == nil {
|
|
client = &http.Client{Timeout: defaultHTTPTimeout}
|
|
}
|
|
body, err := json.Marshal(payload)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
base := strings.TrimRight(apiBaseURL, "/")
|
|
if base == "" {
|
|
base = "http://127.0.0.1:3000/api"
|
|
}
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodPost, base+path, bytes.NewReader(body))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
req.Header.Set("Content-Type", "application/json")
|
|
resp, err := client.Do(req)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer resp.Body.Close()
|
|
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
|
return fmt.Errorf("api returned %s", resp.Status)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func defaultString(value string, fallback string) string {
|
|
if value == "" {
|
|
return fallback
|
|
}
|
|
return value
|
|
}
|
|
|
|
func defaultInt(value int, fallback int) int {
|
|
if value == 0 {
|
|
return fallback
|
|
}
|
|
return value
|
|
}
|