package control import ( "bytes" "context" "encoding/json" "fmt" "net/http" "strings" "time" cmpp "github.com/bigwhite/gocmpp" ) const defaultConnectTimeout = 5 * time.Second type DialFunc func(context.Context, ConnectChannelCommand) error type ConnectChannelCommand struct { SchemaVersion string `json:"schemaVersion"` MessageType string `json:"messageType"` TraceID string `json:"traceId"` ChannelID string `json:"channelId"` ConnectionID string `json:"connectionId"` Reason string `json:"reason"` DesiredConnections int `json:"desiredConnections"` Channel ChannelConfig `json:"channel"` } type ChannelConfig struct { Code string `json:"code"` Name string `json:"name"` GatewayHost string `json:"gatewayHost"` GatewayPort int `json:"gatewayPort"` Account string `json:"account"` PasswordCipher string `json:"passwordCipher"` SrcID string `json:"srcId"` CMPPVersion string `json:"cmppVersion"` RateLimitPerSecond int `json:"rateLimitPerSecond"` } type ConnectionStateCallback struct { ChannelID string `json:"channelId"` ConnectionID string `json:"connectionId"` Status string `json:"status"` DesiredConnections int `json:"desiredConnections"` CurrentConnections int `json:"currentConnections"` LastConnectedAt string `json:"lastConnectedAt,omitempty"` LastDisconnectedAt string `json:"lastDisconnectedAt,omitempty"` LastHeartbeatAt string `json:"lastHeartbeatAt,omitempty"` ReconnectCount int `json:"reconnectCount,omitempty"` LastError string `json:"lastError,omitempty"` } type Server struct { APIBaseURL string HTTPClient *http.Client Dial DialFunc } func Register(mux *http.ServeMux, server Server) { if server.HTTPClient == nil { server.HTTPClient = &http.Client{Timeout: 10 * time.Second} } if server.Dial == nil { server.Dial = DialCMPP } mux.HandleFunc("/connections/connect", server.handleConnectChannel) } func (s Server) handleConnectChannel(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } var command ConnectChannelCommand if err := json.NewDecoder(r.Body).Decode(&command); err != nil { http.Error(w, fmt.Sprintf("invalid connect command: %v", err), http.StatusBadRequest) return } if err := validateConnectChannelCommand(command); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } status := ConnectionStateCallback{ ChannelID: command.ChannelID, ConnectionID: command.ConnectionID, DesiredConnections: desiredConnections(command.DesiredConnections), } if err := s.Dial(r.Context(), command); err != nil { status.Status = "failed" status.CurrentConnections = 0 status.LastDisconnectedAt = time.Now().UTC().Format(time.RFC3339Nano) status.LastError = err.Error() } else { now := time.Now().UTC().Format(time.RFC3339Nano) status.Status = "connected" status.CurrentConnections = status.DesiredConnections status.LastConnectedAt = now status.LastHeartbeatAt = now } if err := s.postConnectionState(r.Context(), status); err != nil { http.Error(w, fmt.Sprintf("failed to callback api: %v", err), http.StatusBadGateway) return } w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(status) } func DialCMPP(ctx context.Context, command ConnectChannelCommand) error { ctx, cancel := context.WithTimeout(ctx, defaultConnectTimeout) defer cancel() version := cmpp.V30 if strings.HasPrefix(command.Channel.CMPPVersion, "2") { version = cmpp.V20 } client := cmpp.NewClient(version) defer client.Disconnect() done := make(chan error, 1) go func() { addr := fmt.Sprintf("%s:%d", command.Channel.GatewayHost, command.Channel.GatewayPort) done <- client.Connect(addr, command.Channel.Account, command.Channel.PasswordCipher, defaultConnectTimeout) }() select { case <-ctx.Done(): return ctx.Err() case err := <-done: return err } } func (s Server) postConnectionState(ctx context.Context, state ConnectionStateCallback) error { apiBaseURL := strings.TrimRight(s.APIBaseURL, "/") if apiBaseURL == "" { apiBaseURL = "http://127.0.0.1:3000/api" } payload, err := json.Marshal(state) if err != nil { return err } req, err := http.NewRequestWithContext(ctx, http.MethodPost, apiBaseURL+"/admin/gateway/connections", bytes.NewReader(payload)) if err != nil { return err } req.Header.Set("Content-Type", "application/json") resp, err := s.HTTPClient.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 validateConnectChannelCommand(command ConnectChannelCommand) error { if command.MessageType != "ConnectChannel" { return fmt.Errorf("unsupported messageType %q", command.MessageType) } if command.ChannelID == "" || command.ConnectionID == "" { return fmt.Errorf("channelId and connectionId are required") } if command.Channel.GatewayHost == "" || command.Channel.GatewayPort <= 0 { return fmt.Errorf("gatewayHost and gatewayPort are required") } if command.Channel.Account == "" || command.Channel.PasswordCipher == "" { return fmt.Errorf("account and passwordCipher are required") } return nil } func desiredConnections(value int) int { if value > 0 { return value } return 1 }