120 lines
3.2 KiB
Go
120 lines
3.2 KiB
Go
package upstream
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log"
|
|
"net"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"testing"
|
|
"time"
|
|
|
|
"cmpp-platform/gateway/internal/queue"
|
|
|
|
cmpp "github.com/bigwhite/gocmpp"
|
|
)
|
|
|
|
func TestFailedSupplierConnectionReconnectsWhenEndpointRecovers(t *testing.T) {
|
|
address := reserveSupplierAddress(t)
|
|
states := make(chan ConnectionState, 16)
|
|
api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
var state ConnectionState
|
|
if err := json.NewDecoder(r.Body).Decode(&state); err != nil {
|
|
http.Error(w, err.Error(), http.StatusBadRequest)
|
|
return
|
|
}
|
|
states <- state
|
|
w.WriteHeader(http.StatusNoContent)
|
|
}))
|
|
defer api.Close()
|
|
|
|
host, portText, err := net.SplitHostPort(address)
|
|
if err != nil {
|
|
t.Fatalf("split address: %v", err)
|
|
}
|
|
var port int
|
|
if _, err := fmt.Sscanf(portText, "%d", &port); err != nil {
|
|
t.Fatalf("parse port: %v", err)
|
|
}
|
|
manager := &Manager{APIBaseURL: api.URL}
|
|
command := queue.ConnectChannelCommand{
|
|
SchemaVersion: queue.SchemaVersion,
|
|
MessageType: queue.MessageTypeConnectChannel,
|
|
ChannelID: "channel-reconnect",
|
|
ConnectionID: "channel-reconnect:primary",
|
|
DesiredConnections: 1,
|
|
Channel: queue.ConnectChannelConfig{
|
|
GatewayHost: host,
|
|
GatewayPort: port,
|
|
Account: "sp",
|
|
PasswordCipher: "secret",
|
|
CMPPVersion: "3.0",
|
|
HeartbeatIntervalSeconds: 1,
|
|
HeartbeatMissThreshold: 3,
|
|
},
|
|
}
|
|
initial, err := manager.ConnectChannel(context.Background(), command)
|
|
if err != nil {
|
|
t.Fatalf("initial connect command: %v", err)
|
|
}
|
|
if initial.Status != "failed" {
|
|
t.Fatalf("initial state = %s, want failed", initial.Status)
|
|
}
|
|
|
|
go func() {
|
|
_ = cmpp.ListenAndServe(address, cmpp.V30, time.Hour, 3, nil,
|
|
cmpp.HandlerFunc(func(response *cmpp.Response, packet *cmpp.Packet, _ *log.Logger) (bool, error) {
|
|
if _, ok := packet.Packer.(*cmpp.CmppConnReqPkt); ok {
|
|
response.Packer.(*cmpp.Cmpp3ConnRspPkt).Version = 0x30
|
|
}
|
|
return false, nil
|
|
}),
|
|
)
|
|
}()
|
|
time.Sleep(100 * time.Millisecond)
|
|
|
|
manager.mu.Lock()
|
|
pool := manager.conns[command.ChannelID]
|
|
manager.mu.Unlock()
|
|
pool.mu.Lock()
|
|
pool.nextReconnectAt = time.Now()
|
|
pool.mu.Unlock()
|
|
pool.signalReconnect()
|
|
|
|
deadline := time.After(4 * time.Second)
|
|
connected := false
|
|
for {
|
|
select {
|
|
case state := <-states:
|
|
if state.Status == "connected" && state.CurrentConnections == 1 {
|
|
connected = true
|
|
}
|
|
if connected && state.Status == "heartbeat" && state.LastHeartbeatAt != "" {
|
|
_, _ = manager.DisconnectChannel(context.Background(), queue.DisconnectChannelCommand{
|
|
MessageType: queue.MessageTypeDisconnectChannel,
|
|
ChannelID: command.ChannelID,
|
|
ConnectionID: command.ConnectionID,
|
|
})
|
|
return
|
|
}
|
|
case <-deadline:
|
|
t.Fatal("timed out waiting for automatic supplier reconnection and active heartbeat")
|
|
}
|
|
}
|
|
}
|
|
|
|
func reserveSupplierAddress(t *testing.T) string {
|
|
t.Helper()
|
|
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
|
if err != nil {
|
|
t.Fatalf("reserve address: %v", err)
|
|
}
|
|
address := listener.Addr().String()
|
|
if err := listener.Close(); err != nil {
|
|
t.Fatalf("close reserved listener: %v", err)
|
|
}
|
|
return address
|
|
}
|