common

Documentation

Index

• type RateLimiter
  • func NewRateLimiter(config RateLimiterConfig) (*RateLimiter, error)
  • func (rl *RateLimiter) Allow(sender string) bool
• type RateLimiterConfig
• type RequestCache
  • func NewRequestCache[T any](timeout time.Duration, maxCacheSize uint32) RequestCache[T]
• type ResponseProcessor

Types

type RateLimiter

type RateLimiter struct {
	// contains filtered or unexported fields
}

Wrapper around Go's rate.Limiter that supports both global and a per-sender rate limiting.

func NewRateLimiter

func NewRateLimiter(config RateLimiterConfig) (*RateLimiter, error)

func (*RateLimiter) Allow

func (rl *RateLimiter) Allow(sender string) bool

type RateLimiterConfig

type RateLimiterConfig struct {
	GlobalRPS      float64 `json:"globalRPS"`
	GlobalBurst    int     `json:"globalBurst"`
	PerSenderRPS   float64 `json:"perSenderRPS"`
	PerSenderBurst int     `json:"perSenderBurst"`
}

type RequestCache

type RequestCache[T any] interface {
	NewRequest(request *api.Message, callbackCh chan<- handlers.UserCallbackPayload, responseData *T) error
	ProcessResponse(response *api.Message, process ResponseProcessor[T]) error
}

RequestCache is used to store pending requests and collect incoming responses as they arrive. It is parameterized by responseData, which is a service-specific type storing all data needed to aggregate responses. Client needs to implement a ResponseProcessor, which is called for every response (see below). Additionally, each request has a timeout, after which the netry will be removed from the cache and an error sent to the callback channel. All methods are thread-safe.

func NewRequestCache

func NewRequestCache[T any](timeout time.Duration, maxCacheSize uint32) RequestCache[T]

type ResponseProcessor

type ResponseProcessor[T any] func(response *api.Message, state *T) (aggregated *handlers.UserCallbackPayload, newState *T, err error)

If aggregated != nil then the aggregated response is ready and the entry will be deleted from RequestCache. Otherwise, state will be updated to newState and the entry will remain in cache, awaiting more responses from nodes.