Updates to golang market data handling code

Author: ettec

api/services.go

@@ -1,13 +1,14 @@
-// Code to interact with the apis of standard otp service interfaces
+// Package api contains Code to interact with the apis of standard otp service interfaces
 package api
 
 import (
+	"errors"
 	"fmt"
 	"github.com/ettec/otp-common/api/executionvenue"
 	"google.golang.org/grpc"
 	"k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
-	"log"
+	"log/slog"
 	"strconv"
 	"time"
 )
@@ -19,7 +20,7 @@ func GetOrderRouter(clientSet *kubernetes.Clientset, maxConnectRetrySecs time.Du
 	})
 
 	if err != nil {
-		panic(err)
+		return nil, fmt.Errorf("failed to list order router services: %w", err)
 	}
 
 	var client executionvenue.ExecutionVenueClient
@@ -34,26 +35,26 @@ func GetOrderRouter(clientSet *kubernetes.Clientset, maxConnectRetrySecs time.Du
 		}
 
 		if podPort == 0 {
-			log.Printf("ignoring order router service as it does not have a port named executionvenue, service: %v", service)
+			slog.Info("ignoring order router service as it does not have an api port", "order-router-service", service)
 			continue
 		}
 
 		targetAddress := service.Name + ":" + strconv.Itoa(int(podPort))
 
-		log.Printf("connecting to order router service %v at: %v", service.Name, targetAddress)
+		slog.Info("connecting to order router service", "name", service.Name, "address", targetAddress)
 
 		conn, err := grpc.Dial(targetAddress, grpc.WithInsecure(), grpc.WithBackoffMaxDelay(maxConnectRetrySecs))
 
 		if err != nil {
-			panic(err)
+			return nil, fmt.Errorf("failed to dial order router service %v at %v: %w", service.Name, targetAddress, err)
 		}
 
 		client = executionvenue.NewExecutionVenueClient(conn)
 		break
 	}
 
 	if client == nil {
-		return nil, fmt.Errorf("failed to find order router")
+		return nil, errors.New("failed to find order router")
 	}
 
 	return client, nil

bootstrap/envvars.go

@@ -1,113 +1,132 @@
-// Utility functions to read environment variables and set defaults.
+// Package bootstrap contains utility functions to read environment variables, if a default is not provided and
+// the value of an environment variable is not set then the application will panic.
 package bootstrap
 
 import (
-	"log"
+	"fmt"
+	"log/slog"
 	"os"
 	"strconv"
 )
 
+// GetIntEnvVar reads an environment variable and returns the value as an int, if the environment variable is not set
+// then the application will panic.
 func GetIntEnvVar(key string) int {
 	value, exists := os.LookupEnv(key)
 	if !exists {
-		log.Panicf("missing required env var %v", key)
+		panic(fmt.Sprintf("missing required env var %v", key))
 	}
 
 	var err error
 	result, err := strconv.Atoi(value)
 	if err != nil {
-		log.Panicf("cannot parse %v, error: %v", key, err)
+		panic(fmt.Sprintf("cannot parse %v, error: %v", key, err))
 	}
 
-	log.Printf("%v set to %v", key, value)
+	slog.Info("Environment Variable Set", key, value)
 
 	return result
 }
 
+// GetEnvVar reads an environment variable and returns the value as a string, if the environment variable is not set
+// then the application will panic.
 func GetEnvVar(key string) string {
 	value, exists := os.LookupEnv(key)
 	if !exists {
-		log.Panicf("missing required env var %v", key)
+		panic(fmt.Sprintf("missing required env var %v", key))
 	}
 
-	log.Printf("%v set to %v", key, value)
+	slog.Info("Environment Variable Set", key, value)
 
 	return value
 }
 
+// GetOptionalEnvVar reads an environment variable and returns the value as a string, or the default value if the
+// environment variable is not set.
 func GetOptionalEnvVar(key string, def string) string {
 	strValue, exists := os.LookupEnv(key)
 	result := def
 	if exists {
 		result = strValue
 	}
 
-	log.Printf("%v set to %v", key, result)
+	slog.Info("Environment Variable Set", key, strValue)
 
 	return result
 }
 
+// GetOptionalBoolEnvVar reads an environment variable and returns the value as a bool, or the default value if the
+// environment variable is not set. If the environment variable is set but cannot be parsed as a bool then the
+// application will panic.
 func GetOptionalBoolEnvVar(key string, def bool) bool {
 	strValue, exists := os.LookupEnv(key)
 	result := def
 	if exists {
 		var err error
 		result, err = strconv.ParseBool(strValue)
 		if err != nil {
-			log.Panicf("cannot parse %v, error: %v", key, err)
+			panic(fmt.Sprintf("cannot parse %v, error: %v", key, err))
 		}
 	}
 
-	log.Printf("%v set to %v", key, result)
+	slog.Info("Environment Variable Set", key, strValue)
 
 	return result
 }
 
+// GetBoolEnvVar reads an environment variable and returns the value as a bool, if the environment variable is not set
+// then the application will panic.
 func GetBoolEnvVar(key string) bool {
 	value, exists := os.LookupEnv(key)
 	if !exists {
-		log.Panicf("missing required env var %v", key)
+		panic(fmt.Sprintf("missing required env var %v", key))
 	}
 
 	var err error
 	result, err := strconv.ParseBool(value)
 	if err != nil {
-		log.Panicf("cannot parse %v, error: %v", key, err)
+		panic(fmt.Sprintf("cannot parse %v, error: %v", key, err))
 	}
 
-	log.Printf("%v set to %v", key, value)
+	slog.Info("Environment Variable Set", key, value)
 
 	return result
 }
 
+// GetOptionalIntEnvVar reads an environment variable and returns the value as an int, or the default value if the
+// environment variable is not set. If the environment variable is set but cannot be parsed as an int then the
+// application will panic.
 func GetOptionalIntEnvVar(key string, def int) int {
 	strValue, exists := os.LookupEnv(key)
 	result := def
 	if exists {
 		var err error
 		result, err = strconv.Atoi(strValue)
 		if err != nil {
-			log.Panicf("cannot parse %v, error: %v", key, err)
+			panic(fmt.Sprintf("cannot parse %v, error: %v", key, err))
 		}
 	}
 
-	log.Printf("%v set to %v", key, result)
+	slog.Info("Environment Variable Set", key, strValue)
 
 	return result
 }
 
+// GetOptionalFloatEnvVar reads an environment variable and returns the value as a float64, if the environment variable is not set
+// then the application will panic.  If the environment variable is set but cannot be parsed as a float64 then the
+// application will panic.
 func GetOptionalFloatEnvVar(key string, def float64) float64 {
 	strValue, exists := os.LookupEnv(key)
 	result := def
 	if exists {
 		var err error
 		result, err = strconv.ParseFloat(strValue, 64)
 		if err != nil {
-			log.Panicf("cannot parse %v, error: %v", key, err)
+			panic(fmt.Sprintf("cannot parse %v, error: %v", key, err))
 		}
 	}
 
-	log.Printf("%v set to %v", key, result)
+	slog.Info("Environment Variable Set", key, result)
 
 	return result
-}
+}

go.mod

@@ -8,6 +8,7 @@ require (
 	github.com/prometheus/client_golang v1.7.1
 	github.com/segmentio/kafka-go v0.3.4
 	github.com/shopspring/decimal v0.0.0-20191009025716-f1972eb1d1f5
+	github.com/stretchr/testify v1.4.0
 	google.golang.org/grpc v1.25.1
 	k8s.io/api v0.17.4
 	k8s.io/apimachinery v0.17.4
@@ -27,6 +28,7 @@ require (
 	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.1 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_model v0.2.0 // indirect
 	github.com/prometheus/common v0.10.0 // indirect
 	github.com/prometheus/procfs v0.1.3 // indirect

loadbalancing/loadbalancing.go

@@ -1,4 +1,4 @@
-// Common utility functions used in load balancing.
+// Package loadbalancing contains utility functions used in load balancing.
 package loadbalancing
 
 import (
@@ -17,10 +17,10 @@ type BalancingStatefulPod struct {
 	TargetAddress string
 	Ordinal       int
 	Name          string
-	Mic string
+	Mic           string
 }
 
-func GetBalancingStatefulPod(pod v12.Pod) ( *BalancingStatefulPod,  error) {
+func GetBalancingStatefulPod(pod v12.Pod) (*BalancingStatefulPod, error) {
 	const micLabel = "mic"
 	if _, ok := pod.Labels[micLabel]; !ok {
 		return nil, fmt.Errorf("ignoring stateful pod as it does not have a mic label, pod: %v", pod)
@@ -30,7 +30,7 @@ func GetBalancingStatefulPod(pod v12.Pod) ( *BalancingStatefulPod,  error) {
 
 	targetAddress, err := getStatefulSetMemberAddress(pod)
 	if err != nil {
-		return  nil, fmt.Errorf("failed to get stateful pod address:%v", err)
+		return nil, fmt.Errorf("failed to get stateful pod address:%v", err)
 	}
 
 	ordinal, err := getStatefulSetPodOrdinal(pod)

marketdata/boundedcircularbuffer.go

@@ -0,0 +1,62 @@
+package marketdata
+
+type boundedCircularBuffer[T any] struct {
+	buffer   []T
+	capacity int
+	len      int
+	readPtr  int
+	writePtr int
+}
+
+func newBoundedCircularBuffer[T any](capacity int) *boundedCircularBuffer[T] {
+	b := &boundedCircularBuffer[T]{buffer: make([]T, capacity), capacity: capacity}
+
+	return b
+}
+
+// true if the buffer is not full and the value is added
+func (b *boundedCircularBuffer[T]) addHead(item T) bool {
+
+	if b.len == b.capacity {
+		return false
+	}
+
+	b.buffer[b.writePtr] = item
+	b.len++
+
+	if b.writePtr == b.capacity-1 {
+		b.writePtr = 0
+	} else {
+		b.writePtr++
+	}
+
+	return true
+
+}
+
+func (b *boundedCircularBuffer[T]) getTail() (T, bool) {
+	var result T
+	if b.len == 0 {
+		return result, false
+	}
+
+	return b.buffer[b.readPtr], true
+}
+
+// returns the value and true if a value is available
+func (b *boundedCircularBuffer[T]) removeTail() (T, bool) {
+	var result T
+	if b.len == 0 {
+		return result, false
+	}
+
+	result = b.buffer[b.readPtr]
+	b.len--
+	b.readPtr++
+	if b.readPtr == b.capacity {
+		b.readPtr = 0
+	}
+
+	return result, true
+
+}