Merge pull request #24 from kvukelic/env_relocation_2022

Changes in response to UOF production environment relocation

api/api.go

@@ -15,8 +15,9 @@ import (
 )
 
 const (
-	stagingServer    = "stgapi.betradar.com"
-	productionServer = "api.betradar.com"
+	stagingServer          = "stgapi.betradar.com"
+	productionServer       = "api.betradar.com"
+	productionServerGlobal = "global.api.betradar.com"
 )
 
 var RequestTimeout = 32 * time.Second
@@ -37,6 +38,8 @@ func Dial(ctx context.Context, env uof.Environment, token string) (*API, error)
 		return Staging(ctx, token)
 	case uof.Production:
 		return Production(ctx, token)
+	case uof.ProductionGlobal:
+		return ProductionGlobal(ctx, token)
 	default:
 		return nil, uof.Notice("queue dial", fmt.Errorf("unknown environment %d", env))
 	}
@@ -64,6 +67,17 @@ func Production(exitSig context.Context, token string) (*API, error) {
 	return a, a.Ping()
 }
 
+// Production connects to the production system
+func ProductionGlobal(exitSig context.Context, token string) (*API, error) {
+	a := &API{
+		server:  productionServerGlobal,
+		token:   token,
+		exitSig: exitSig,
+		client:  client(),
+	}
+	return a, a.Ping()
+}
+
 func client() *retryablehttp.Client {
 	c := retryablehttp.NewClient()
 	c.Logger = nil

cmd/client/main.go

@@ -96,7 +96,7 @@ func logMessages(in <-chan *uof.Message) error {
 func logMessage(m *uof.Message) {
 	switch m.Type {
 	case uof.MessageTypeConnection:
-		fmt.Printf("%-25s status: %s\n", m.Type, m.Connection.Status)
+		fmt.Printf("%-25s status: %s, server: %s, local: %s, network: %s, tls: %s\n", m.Type, m.Connection.Status, m.Connection.ServerName, m.Connection.LocalAddr, m.Connection.Network, m.Connection.TLSVersionToString())
 	case uof.MessageTypeFixture:
 		fmt.Printf("%-25s lang: %s, urn: %s raw: %d\n", m.Type, m.Lang, m.Fixture.URN, len(m.Raw))
 	case uof.MessageTypeMarkets:

enum.go

@@ -489,4 +489,5 @@ const (
 	Production Environment = iota
 	Staging
 	Replay
+	ProductionGlobal
 )

internal.go

@@ -1,8 +1,28 @@
 package uof
 
+import "crypto/tls"
+
 type Connection struct {
-	Status    ConnectionStatus `json:"status"`
-	Timestamp int              `json:"timestamp,omitempty"`
+	Status     ConnectionStatus `json:"status"`
+	Timestamp  int              `json:"timestamp,omitempty"`
+	ServerName string           `json:"servername,omitempty"`
+	LocalAddr  string           `json:"localaddr,omitempty"`
+	Network    string           `json:"network,omitempty"`
+	TLSVersion uint16           `json:"tlsversion,omitempty"`
+}
+
+func (c Connection) TLSVersionToString() string {
+	switch c.TLSVersion {
+	case tls.VersionTLS10:
+		return "1.0"
+	case tls.VersionTLS11:
+		return "1.1"
+	case tls.VersionTLS12:
+		return "1.2"
+	case tls.VersionTLS13:
+		return "1.3"
+	}
+	return "unknown"
 }
 
 type ConnectionStatus int8

message.go

@@ -256,7 +256,7 @@ func NewPlayerMessage(lang Lang, player *Player, requestedAt int) *Message {
 	}
 }
 
-func NewConnnectionMessage(status ConnectionStatus) *Message {
+func NewSimpleConnnectionMessage(status ConnectionStatus) *Message {
 	ts := uniqTimestamp()
 	return &Message{
 		Header: Header{
@@ -273,6 +273,27 @@ func NewConnnectionMessage(status ConnectionStatus) *Message {
 	}
 }
 
+func NewDetailedConnnectionMessage(status ConnectionStatus, serverName, localAddr, network string, tlsVersion uint16) *Message {
+	ts := uniqTimestamp()
+	return &Message{
+		Header: Header{
+			Type:       MessageTypeConnection,
+			Scope:      MessageScopeSystem,
+			ReceivedAt: ts,
+		},
+		Body: Body{
+			Connection: &Connection{
+				Status:     status,
+				Timestamp:  ts,
+				ServerName: serverName,
+				LocalAddr:  localAddr,
+				Network:    network,
+				TLSVersion: tlsVersion,
+			},
+		},
+	}
+}
+
 func NewProducersChangeMessage(pc ProducersChange) *Message {
 	return &Message{
 		Header: Header{

message_test.go

@@ -245,7 +245,7 @@ func TestUIDWithLang(t *testing.T) {
 }
 
 func TestNewMessage(t *testing.T) {
-	m := NewConnnectionMessage(ConnectionStatusUp)
+	m := NewSimpleConnnectionMessage(ConnectionStatusUp)
 	assert.True(t, m.Is(MessageTypeConnection))
 
 	m = NewPlayerMessage(LangEN, nil, 0)

pipe/fixture_test.go

@@ -42,7 +42,7 @@ func TestFixturePipe(t *testing.T) {
 	out, _ := f(in)
 
 	// this type of message is passing through
-	m := uof.NewConnnectionMessage(uof.ConnectionStatusUp)
+	m := uof.NewSimpleConnnectionMessage(uof.ConnectionStatusUp)
 	in <- m
 	om := <-out
 	assert.Equal(t, m, om)

pipe/market_test.go

@@ -34,7 +34,7 @@ func TestMarketsPipe(t *testing.T) {
 	out, _ := ms(in)
 
 	// this type of message is passing through
-	m := uof.NewConnnectionMessage(uof.ConnectionStatusUp)
+	m := uof.NewSimpleConnnectionMessage(uof.ConnectionStatusUp)
 	in <- m
 	om := <-out
 	assert.Equal(t, m, om)

pipe/player_test.go

@@ -30,7 +30,7 @@ func TestPlayerPipe(t *testing.T) {
 	out, _ := p(in)
 
 	// this type of message is passing through
-	m := uof.NewConnnectionMessage(uof.ConnectionStatusUp)
+	m := uof.NewSimpleConnnectionMessage(uof.ConnectionStatusUp)
 	in <- m
 	om := <-out
 	assert.Equal(t, m, om)

pipe/recovery_test.go

@@ -125,7 +125,7 @@ func TestRecoveryRequests(t *testing.T) {
 	go r.loop(in, out, errc)
 
 	// 1. connection status triggers recovery requests
-	in <- uof.NewConnnectionMessage(uof.ConnectionStatusUp)
+	in <- uof.NewSimpleConnnectionMessage(uof.ConnectionStatusUp)
 	recoveryRequestPrematch := <-m.calls
 	recoveryRequestLive := <-m.calls
 	if recoveryRequestPrematch.producer == uof.ProducerLiveOdds {

queue/queue.go

@@ -15,11 +15,12 @@ import (
 )
 
 const (
-	replayServer     = "replaymq.betradar.com:5671"
-	stagingServer    = "stgmq.betradar.com:5671"
-	productionServer = "mq.betradar.com:5671"
-	queueExchange    = "unifiedfeed"
-	bindingKeyAll    = "#"
+	replayServer           = "replaymq.betradar.com:5671"
+	stagingServer          = "stgmq.betradar.com:5671"
+	productionServer       = "mq.betradar.com:5671"
+	productionServerGlobal = "global.mq.betradar.com:5671"
+	queueExchange          = "unifiedfeed"
+	bindingKeyAll          = "#"
 )
 
 // Dial connects to the queue chosen by environment
@@ -31,6 +32,8 @@ func Dial(ctx context.Context, env uof.Environment, bookmakerID, token string) (
 		return DialStaging(ctx, bookmakerID, token)
 	case uof.Production:
 		return DialProduction(ctx, bookmakerID, token)
+	case uof.ProductionGlobal:
+		return DialProductionGlobal(ctx, bookmakerID, token)
 	default:
 		return nil, uof.Notice("queue dial", fmt.Errorf("unknown environment %d", env))
 	}
@@ -41,6 +44,11 @@ func DialProduction(ctx context.Context, bookmakerID, token string) (*Connection
 	return dial(ctx, productionServer, bookmakerID, token)
 }
 
+// Dial connects to the production queue
+func DialProductionGlobal(ctx context.Context, bookmakerID, token string) (*Connection, error) {
+	return dial(ctx, productionServerGlobal, bookmakerID, token)
+}
+
 // DialStaging connects to the staging queue
 func DialStaging(ctx context.Context, bookmakerID, token string) (*Connection, error) {
 	return dial(ctx, stagingServer, bookmakerID, token)
@@ -55,6 +63,14 @@ type Connection struct {
 	msgs   <-chan amqp.Delivery
 	errs   <-chan *amqp.Error
 	reDial func() (*Connection, error)
+	info   ConnectionInfo
+}
+
+type ConnectionInfo struct {
+	server     string
+	local      string
+	network    string
+	tlsVersion uint16
 }
 
 func (c *Connection) Listen() (<-chan *uof.Message, <-chan error) {
@@ -154,6 +170,12 @@ func dial(ctx context.Context, server, bookmakerID, token string) (*Connection,
 		reDial: func() (*Connection, error) {
 			return dial(ctx, server, bookmakerID, token)
 		},
+		info: ConnectionInfo{
+			server:     server,
+			local:      conn.LocalAddr().String(),
+			network:    conn.LocalAddr().Network(),
+			tlsVersion: conn.ConnectionState().Version,
+		},
 	}
 
 	go func() {

queue/reconnect.go

@@ -44,12 +44,14 @@ func WithReconnect(ctx context.Context, conn *Connection) func() (<-chan *uof.Me
 			defer close(out)
 			defer close(errc)
 			for {
-				out <- uof.NewConnnectionMessage(uof.ConnectionStatusUp) // signal connect
+				// signal connect
+				out <- uof.NewDetailedConnnectionMessage(uof.ConnectionStatusUp, conn.info.server, conn.info.local, conn.info.network, conn.info.tlsVersion)
 				conn.drain(out, errc)
 				if done() {
 					return
 				}
-				out <- uof.NewConnnectionMessage(uof.ConnectionStatusDown) // signal connection lost
+				// signal connection lost
+				out <- uof.NewSimpleConnnectionMessage(uof.ConnectionStatusDown)
 				if err := withBackoff(ctx, reconnect, maxInterval, maxElapsedTime); err != nil {
 					return
 				}

sdk/sdk.go

@@ -24,7 +24,6 @@ type Config struct {
 	Stages        []pipe.InnerStage
 	Replay        func(*api.ReplayAPI) error
 	Env           uof.Environment
-	Staging       bool
 	Languages     []uof.Lang
 	ErrorListener ErrorListenerFunc
 }
@@ -128,11 +127,17 @@ func Languages(langs []uof.Lang) Option {
 	}
 }
 
+// Global forces use of global production environment.
+func Global() Option {
+	return func(c *Config) {
+		c.Env = uof.ProductionGlobal
+	}
+}
+
 // Staging forces use of staging environment instead of production.
 func Staging() Option {
 	return func(c *Config) {
 		c.Env = uof.Staging
-		c.Staging = true
 	}
 }