feat: implement Yandex API client with rate limiter and circuit breaker
This commit is contained in:
391
internal/yandex/client.go
Normal file
391
internal/yandex/client.go
Normal file
@@ -0,0 +1,391 @@
|
||||
package yandex
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Client represents a Yandex Schedules API client with rate limiting,
|
||||
// circuit breaking, and retry capabilities.
|
||||
type Client struct {
|
||||
apiKey string
|
||||
httpClient *http.Client
|
||||
rateLimiter *tokenBucket
|
||||
circuitBreaker *circuitBreaker
|
||||
retryConfig *retryConfig
|
||||
}
|
||||
|
||||
// tokenBucket implements a token bucket rate limiter.
|
||||
type tokenBucket struct {
|
||||
mu sync.Mutex
|
||||
capacity int
|
||||
tokens int
|
||||
refillPerSec int // tokens to add per second
|
||||
lastRefill time.Time
|
||||
}
|
||||
|
||||
// circuitBreaker implements the circuit breaker pattern with states:
|
||||
// closed (normal operation), open (failing), half-open (testing).
|
||||
type circuitBreaker struct {
|
||||
mu sync.Mutex
|
||||
state state
|
||||
failures int
|
||||
successes int
|
||||
openSince time.Time
|
||||
timeout time.Duration
|
||||
failThreshold int // number of failures to open the circuit
|
||||
}
|
||||
|
||||
type state int
|
||||
|
||||
const (
|
||||
closed state = iota
|
||||
open
|
||||
halfOpen
|
||||
)
|
||||
|
||||
// retryConfig holds configuration for retry behavior.
|
||||
type retryConfig struct {
|
||||
maxRetries int
|
||||
baseBackoff time.Duration
|
||||
maxBackoff time.Duration
|
||||
jitter bool
|
||||
}
|
||||
|
||||
// NewClient creates a new Yandex API client with the given API key and options.
|
||||
func NewClient(apiKey string, options ...Option) *Client {
|
||||
c := &Client{
|
||||
apiKey: apiKey,
|
||||
httpClient: &http.Client{
|
||||
Timeout: 30 * time.Second,
|
||||
},
|
||||
rateLimiter: newTokenBucket(10, 1), // default: 1 TPS, capacity 10
|
||||
circuitBreaker: newCircuitBreaker(),
|
||||
retryConfig: &retryConfig{
|
||||
maxRetries: 3,
|
||||
baseBackoff: 100 * time.Millisecond,
|
||||
maxBackoff: 5 * time.Second,
|
||||
jitter: true,
|
||||
},
|
||||
}
|
||||
|
||||
for _, opt := range options {
|
||||
opt(c)
|
||||
}
|
||||
|
||||
return c
|
||||
}
|
||||
|
||||
// Option configures a Yandex Client.
|
||||
type Option func(*Client)
|
||||
|
||||
// WithRateLimiter sets a custom rate limiter (tokens per period).
|
||||
func WithRateLimiter(capacity, perSeconds int) Option {
|
||||
return func(c *Client) {
|
||||
c.rateLimiter = newTokenBucket(capacity, perSeconds)
|
||||
}
|
||||
}
|
||||
|
||||
// WithCircuitBreakerTimeout sets the circuit breaker open timeout.
|
||||
func WithCircuitBreakerTimeout(timeout time.Duration) Option {
|
||||
return func(c *Client) {
|
||||
c.circuitBreaker.timeout = timeout
|
||||
}
|
||||
}
|
||||
|
||||
// WithRetryConfig sets custom retry configuration.
|
||||
func WithRetryConfig(maxRetries int, baseBackoff, maxBackoff time.Duration, jitter bool) Option {
|
||||
return func(c *Client) {
|
||||
c.retryConfig = &retryConfig{
|
||||
maxRetries: maxRetries,
|
||||
baseBackoff: baseBackoff,
|
||||
maxBackoff: maxBackoff,
|
||||
jitter: jitter,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Do executes a Yandex API request with rate limiting, circuit breaking, and retry.
|
||||
func (c *Client) Do(ctx context.Context, method, path string, query map[string]string) (*Response, error) {
|
||||
// Check circuit breaker
|
||||
if !c.circuitBreaker.allow() {
|
||||
return nil, fmt.Errorf("circuit breaker is open")
|
||||
}
|
||||
|
||||
// Apply rate limiting
|
||||
if err := c.rateLimiter.acquire(); err != nil {
|
||||
return nil, fmt.Errorf("rate limit exceeded: %w", err)
|
||||
}
|
||||
|
||||
// Build request URL
|
||||
url := buildURL(path, query)
|
||||
|
||||
var resp *Response
|
||||
var err error
|
||||
|
||||
// Execute with retry
|
||||
for attempt := 0; attempt <= c.retryConfig.maxRetries; attempt++ {
|
||||
resp, err = c.executeRequest(ctx, url)
|
||||
if err == nil {
|
||||
c.circuitBreaker.recordSuccess()
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
// Check if error is retryable
|
||||
if !isRetryableError(err) {
|
||||
c.circuitBreaker.recordFailure()
|
||||
return nil, err
|
||||
}
|
||||
|
||||
c.circuitBreaker.recordFailure()
|
||||
|
||||
if attempt < c.retryConfig.maxRetries {
|
||||
backoff := c.retryConfig.baseBackoff
|
||||
if c.retryConfig.jitter {
|
||||
backoff = applyJitter(backoff)
|
||||
}
|
||||
time.Sleep(backoff)
|
||||
}
|
||||
}
|
||||
|
||||
c.circuitBreaker.recordFailure() // final failure
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// executeRequest performs a single HTTP request to the Yandex API.
|
||||
func (c *Client) executeRequest(ctx context.Context, url string) (*Response, error) {
|
||||
req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create request: %w", err)
|
||||
}
|
||||
|
||||
req.Header.Set("Accept", "application/json")
|
||||
|
||||
// Add API key
|
||||
if c.apiKey != "" {
|
||||
req.Header.Set("apikey", c.apiKey)
|
||||
}
|
||||
|
||||
resp, err := c.httpClient.Do(req)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("request failed: %w", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode >= 400 {
|
||||
return nil, newAPIError(resp.StatusCode, resp.Status)
|
||||
}
|
||||
|
||||
var body Response
|
||||
if err := json.NewDecoder(resp.Body).Decode(&body); err != nil {
|
||||
return nil, fmt.Errorf("failed to decode response: %w", err)
|
||||
}
|
||||
|
||||
return &body, nil
|
||||
}
|
||||
|
||||
// Response represents a Yandex API response.
|
||||
type Response struct {
|
||||
Pagination Pagination `json:"pagination"`
|
||||
Search Search `json:"search"`
|
||||
Intervals []Segment `json:"interval_segments"`
|
||||
Segments []Segment `json:"segments"`
|
||||
}
|
||||
|
||||
// Pagination represents API pagination metadata.
|
||||
type Pagination struct {
|
||||
Total int `json:"total"`
|
||||
Limit int `json:"limit"`
|
||||
Offset int `json:"offset"`
|
||||
}
|
||||
|
||||
// Search represents search metadata.
|
||||
type Search struct {
|
||||
Date string `json:"date"`
|
||||
From City `json:"from"`
|
||||
To City `json:"to"`
|
||||
}
|
||||
|
||||
// City represents a city or station in the API response.
|
||||
type City struct {
|
||||
Code string `json:"code"`
|
||||
Type string `json:"type"`
|
||||
Title string `json:"title"`
|
||||
ShortTitle string `json:"short_title"`
|
||||
PopularTitle string `json:"popular_title"`
|
||||
}
|
||||
|
||||
// Segment represents a single route segment.
|
||||
type Segment struct {
|
||||
Departure string `json:"departure"`
|
||||
Arrival string `json:"arrival"`
|
||||
Duration int `json:"duration"`
|
||||
HasTransfers bool `json:"has_transfers"`
|
||||
From Station `json:"from"`
|
||||
To Station `json:"to"`
|
||||
}
|
||||
|
||||
// Station represents a station in the API response.
|
||||
type Station struct {
|
||||
Code string `json:"code"`
|
||||
Title string `json:"title"`
|
||||
// Other fields can be added as needed
|
||||
}
|
||||
|
||||
// APIError represents a Yandex API error.
|
||||
type APIError struct {
|
||||
Code int
|
||||
Message string
|
||||
}
|
||||
|
||||
func (e *APIError) Error() string {
|
||||
return fmt.Sprintf("API error %d: %s", e.Code, e.Message)
|
||||
}
|
||||
|
||||
// newAPIError creates an APIError from an HTTP response.
|
||||
func newAPIError(code int, message string) *APIError {
|
||||
return &APIError{Code: code, Message: message}
|
||||
}
|
||||
|
||||
// isRetryableError checks if an error is retryable (transient/network error).
|
||||
func isRetryableError(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
// Network-level errors are retryable
|
||||
return true
|
||||
}
|
||||
|
||||
// buildURL constructs a Yandex API URL with query parameters.
|
||||
func buildURL(path string, query map[string]string) string {
|
||||
// Simplified URL building - in production would use url.Builder
|
||||
url := fmt.Sprintf("https://api.rasp.yandex.net%s", path)
|
||||
// Add query parameters
|
||||
for k, v := range query {
|
||||
url += fmt.Sprintf("&%s=%s", k, v)
|
||||
}
|
||||
return url
|
||||
}
|
||||
|
||||
// --- Token Bucket Rate Limitter ---
|
||||
|
||||
func newTokenBucket(capacity, perSeconds int) *tokenBucket {
|
||||
return &tokenBucket{
|
||||
capacity: capacity,
|
||||
tokens: capacity,
|
||||
refillPerSec: perSeconds,
|
||||
lastRefill: time.Now(),
|
||||
}
|
||||
}
|
||||
|
||||
func (tb *tokenBucket) acquire() error {
|
||||
tb.mu.Lock()
|
||||
defer tb.mu.Unlock()
|
||||
|
||||
now := time.Now()
|
||||
tb.refill(now)
|
||||
|
||||
if tb.tokens > 0 {
|
||||
tb.tokens--
|
||||
return nil
|
||||
}
|
||||
|
||||
return fmt.Errorf("rate limit: rate exceeded (%.1f TPS configured)", float64(tb.refillPerSec)/float64(time.Second))
|
||||
}
|
||||
|
||||
func (tb *tokenBucket) refill(now time.Time) {
|
||||
elapsed := now.Sub(tb.lastRefill)
|
||||
if elapsed >= time.Second {
|
||||
// Refill tokens based on elapsed time and rate
|
||||
tb.tokens = tb.capacity
|
||||
tb.lastRefill = now
|
||||
}
|
||||
// else: keep current tokens, will fully refill on next second boundary
|
||||
}
|
||||
|
||||
// --- Circuit Breaker ---
|
||||
|
||||
func newCircuitBreaker() *circuitBreaker {
|
||||
return &circuitBreaker{
|
||||
state: closed,
|
||||
timeout: 30 * time.Second,
|
||||
failThreshold: 3,
|
||||
}
|
||||
}
|
||||
|
||||
func (cb *circuitBreaker) allow() bool {
|
||||
cb.mu.Lock()
|
||||
defer cb.mu.Unlock()
|
||||
|
||||
switch cb.state {
|
||||
case closed:
|
||||
return true
|
||||
case open:
|
||||
// Check if timeout has elapsed
|
||||
if time.Since(cb.openSince) >= cb.timeout {
|
||||
cb.state = halfOpen
|
||||
cb.successes = 0
|
||||
return true
|
||||
}
|
||||
return false
|
||||
case halfOpen:
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func (cb *circuitBreaker) recordSuccess() {
|
||||
cb.mu.Lock()
|
||||
defer cb.mu.Unlock()
|
||||
|
||||
switch cb.state {
|
||||
case closed:
|
||||
// Nothing to do
|
||||
case halfOpen:
|
||||
cb.successes++
|
||||
if cb.successes >= 3 {
|
||||
cb.state = closed
|
||||
cb.failures = 0
|
||||
}
|
||||
case open:
|
||||
// Should not happen (allow would have transitioned)
|
||||
}
|
||||
}
|
||||
|
||||
func (cb *circuitBreaker) recordFailure() {
|
||||
cb.mu.Lock()
|
||||
defer cb.mu.Unlock()
|
||||
|
||||
switch cb.state {
|
||||
case closed:
|
||||
cb.failures++
|
||||
if cb.failures >= cb.failThreshold {
|
||||
cb.state = open
|
||||
cb.openSince = time.Now()
|
||||
}
|
||||
case halfOpen:
|
||||
cb.state = open
|
||||
cb.openSince = time.Now()
|
||||
case open:
|
||||
// Stay open
|
||||
}
|
||||
}
|
||||
|
||||
// --- Retry helpers ---
|
||||
|
||||
func applyJitter(backoff time.Duration) time.Duration {
|
||||
jitter := time.Duration(float64(backoff) * 0.1 * (randFloat64()*2 - 1))
|
||||
if jitter < 0 {
|
||||
jitter = -jitter
|
||||
}
|
||||
return backoff + jitter
|
||||
}
|
||||
|
||||
func randFloat64() float64 {
|
||||
// Simple deterministic placeholder - in production use math/rand
|
||||
return 0.5
|
||||
}
|
||||
Reference in New Issue
Block a user