concurrent

package
v0.9.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 22, 2026 License: MIT Imports: 6 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	ErrCodeInternalError = -32603
	ErrCodeTimeout       = -32001
	ErrCodeCircuitOpen   = -32002
	ErrCodeRateLimited   = -32003
)

Variables

This section is empty.

Functions

This section is empty.

Types

type BatchConfig 

type BatchConfig struct {
	WindowMs       int
	MaxBatchSize   int
	MaxWaitTime    time.Duration
	EnableBatching bool
}

type BatchProcessor 

type BatchProcessor interface {
	ProcessBatch(ctx context.Context, serverName string, requests []Request) []Response
}

type BatchRequest 

type BatchRequest struct {
	Request  Request
	ResultCh chan *Response
	ErrorCh  chan error
	Arrived  time.Time
}

type Batcher 

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

func NewBatcher 

func NewBatcher(config BatchConfig, logger *slog.Logger) *Batcher

func (*Batcher) AddRequest 

func (b *Batcher) AddRequest(serverName string, req Request, resultCh chan *Response, errorCh chan error) bool

func (*Batcher) Close 

func (b *Batcher) Close()

func (*Batcher) Flush 

func (b *Batcher) Flush(serverName string)

func (*Batcher) GetPendingCount 

func (b *Batcher) GetPendingCount(serverName string) int

type CircuitBreaker 

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

func NewCircuitBreaker 

func NewCircuitBreaker(threshold int, cooldown time.Duration, halfOpenCooldown time.Duration) *CircuitBreaker

func (*CircuitBreaker) Allow 

func (cb *CircuitBreaker) Allow() bool

func (*CircuitBreaker) GetMetrics 

func (cb *CircuitBreaker) GetMetrics() CircuitBreakerMetrics

func (*CircuitBreaker) RecordFailure 

func (cb *CircuitBreaker) RecordFailure()

func (*CircuitBreaker) RecordSuccess 

func (cb *CircuitBreaker) RecordSuccess()

func (*CircuitBreaker) Reset 

func (cb *CircuitBreaker) Reset()

func (*CircuitBreaker) State 

func (cb *CircuitBreaker) State() CircuitState

type CircuitBreakerConfig 

type CircuitBreakerConfig struct {
	Threshold          int
	Cooldown           time.Duration
	HalfOpenMaxSuccess int
}

type CircuitBreakerGroup 

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

func NewCircuitBreakerGroup 

func NewCircuitBreakerGroup() *CircuitBreakerGroup

func (*CircuitBreakerGroup) Get 

func (g *CircuitBreakerGroup) Get(name string) *CircuitBreaker

func (*CircuitBreakerGroup) Register 

func (g *CircuitBreakerGroup) Register(name string, cb *CircuitBreaker)

func (*CircuitBreakerGroup) ResetAll 

func (g *CircuitBreakerGroup) ResetAll()

type CircuitBreakerMetrics 

type CircuitBreakerMetrics struct {
	State          string
	Failures       int32
	Threshold      int
	TotalSuccesses int32
	TotalFailures  int32
	LastFailure    time.Time
}

type CircuitState 

type CircuitState int
const (
	StateClosed CircuitState = iota
	StateOpen
	StateHalfOpen
)

func (CircuitState) String 

func (s CircuitState) String() string

type Clock added in v0.6.0 

type Clock interface {
	Now() time.Time
	Since(time.Time) time.Duration
}

type ConcurrentError 

type ConcurrentError struct {
	Code    int             `json:"code"`
	Message string          `json:"message"`
	Data    json.RawMessage `json:"data,omitempty"`
}

func (*ConcurrentError) Error 

func (e *ConcurrentError) Error() string

type DefaultBatchProcessor 

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

func NewDefaultBatchProcessor 

func NewDefaultBatchProcessor(logger *slog.Logger) *DefaultBatchProcessor

func (*DefaultBatchProcessor) ProcessBatch 

func (p *DefaultBatchProcessor) ProcessBatch(ctx context.Context, serverName string, requests []Request) []Response

type FakeClock added in v0.6.0 

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

func NewFakeClock added in v0.6.0 

func NewFakeClock(t time.Time) *FakeClock

func (*FakeClock) Add added in v0.6.0 

func (fc *FakeClock) Add(d time.Duration)

func (*FakeClock) Now added in v0.6.0 

func (fc *FakeClock) Now() time.Time

func (*FakeClock) Set added in v0.6.0 

func (fc *FakeClock) Set(t time.Time)

func (*FakeClock) Since added in v0.6.0 

func (fc *FakeClock) Since(t time.Time) time.Duration

type MultiServerRateLimiter 

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

func NewMultiServerRateLimiter 

func NewMultiServerRateLimiter(config RateLimiterConfig) *MultiServerRateLimiter

func (*MultiServerRateLimiter) Allow 

func (m *MultiServerRateLimiter) Allow(serverName string) bool

func (*MultiServerRateLimiter) Close added in v0.6.0 

func (m *MultiServerRateLimiter) Close()

func (*MultiServerRateLimiter) GetLimiter 

func (m *MultiServerRateLimiter) GetLimiter(serverName string) *RateLimiter

func (*MultiServerRateLimiter) GetStats 

func (m *MultiServerRateLimiter) GetStats(serverName string) RateLimiterStats

func (*MultiServerRateLimiter) Reset 

func (m *MultiServerRateLimiter) Reset(serverName string)

func (*MultiServerRateLimiter) ResetAll 

func (m *MultiServerRateLimiter) ResetAll()

type PoolConfig 

type PoolConfig struct {
	MaxConcurrent   int
	MaxQueueSize    int
	QueueTimeout    time.Duration
	WorkerCount     int
	BatchWindowMs   int
	RateLimitMax    int
	RateLimitWindow time.Duration
}

type PoolStats 

type PoolStats struct {
	TotalRequests   int64
	ActiveRequests  int64
	QueuedRequests  int64
	FailedRequests  int64
	SuccessRequests int64
	AverageLatency  time.Duration
}

type QueueManager 

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

func NewQueueManager 

func NewQueueManager(maxSize int, timeout time.Duration) *QueueManager

func (*QueueManager) ClearQueue 

func (qm *QueueManager) ClearQueue(serverName string)

func (*QueueManager) Enqueue 

func (qm *QueueManager) Enqueue(serverName string, req Request, resultCh chan *Response, errorCh chan error) error

func (*QueueManager) GetOrCreateQueue 

func (qm *QueueManager) GetOrCreateQueue(serverName string) *RequestQueue

func (*QueueManager) GetOverflowCount 

func (qm *QueueManager) GetOverflowCount() int64

func (*QueueManager) GetQueueSize 

func (qm *QueueManager) GetQueueSize(serverName string) int

type QueuedRequest 

type QueuedRequest struct {
	Request    Request
	ResultCh   chan *Response
	ErrorCh    chan error
	EnqueuedAt time.Time
}

type RateLimiter 

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

func NewRateLimiter 

func NewRateLimiter(max int, window time.Duration) *RateLimiter

func (*RateLimiter) Allow 

func (rl *RateLimiter) Allow() bool

func (*RateLimiter) Close added in v0.6.0 

func (rl *RateLimiter) Close()

func (*RateLimiter) GetBlockedCount 

func (rl *RateLimiter) GetBlockedCount() int64

func (*RateLimiter) GetUsage 

func (rl *RateLimiter) GetUsage() (current int, max int)

func (*RateLimiter) Reset 

func (rl *RateLimiter) Reset()

type RateLimiterConfig 

type RateLimiterConfig struct {
	MaxRequests int
	Window      time.Duration
}

type RateLimiterStats 

type RateLimiterStats struct {
	ServerName      string
	CurrentRequests int
	MaxRequests     int
	BlockedCount    int64
}

type RealClock added in v0.6.0 

type RealClock struct{}

func (RealClock) Now added in v0.6.0 

func (RealClock) Now() time.Time

func (RealClock) Since added in v0.6.0 

func (RealClock) Since(t time.Time) time.Duration

type Request 

type Request struct {
	Method     string          `json:"method"`
	Params     json.RawMessage `json:"params,omitempty"`
	ID         interface{}     `json:"id"`
	ServerName string          `json:"server_name"`
	Timeout    time.Duration
	ResultCh   chan *Response
	ErrorCh    chan error
}

type RequestHandler 

type RequestHandler interface {
	Handle(ctx context.Context, req *Request) (*Response, error)
}

type RequestQueue 

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

type Response 

type Response struct {
	Result json.RawMessage  `json:"result,omitempty"`
	Error  *ConcurrentError `json:"error,omitempty"`
	ID     interface{}      `json:"id"`
}

type ServerHandle 

type ServerHandle struct {
	Name           string
	State          ServerState
	CurrentLoad    int32
	MaxConcurrent  int
	RequestCount   int64
	ErrorCount     int64
	AvgLatencyMs   float64
	LastRequestAt  time.Time
	RestartCount   int
	CurrentBackoff time.Duration
}

type ServerState 

type ServerState string
const (
	StateIdle     ServerState = "idle"
	StateRunning  ServerState = "running"
	StateBusy     ServerState = "busy"
	StateStopping ServerState = "stopping"
	StateStopped  ServerState = "stopped"
	StateError    ServerState = "error"
)

type StdioPool 

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

func NewStdioPool 

func NewStdioPool(config PoolConfig, logger *slog.Logger) *StdioPool

func (*StdioPool) Close 

func (p *StdioPool) Close() error

func (*StdioPool) GetPoolStats 

func (p *StdioPool) GetPoolStats() PoolStats

func (*StdioPool) GetServerStats 

func (p *StdioPool) GetServerStats(serverName string) (*ServerHandle, error)

func (*StdioPool) RegisterServer 

func (p *StdioPool) RegisterServer(name string, maxConcurrent int)

func (*StdioPool) SendRequest 

func (p *StdioPool) SendRequest(ctx context.Context, serverName string, req *Request) (*Response, error)

func (*StdioPool) ServerCount 

func (p *StdioPool) ServerCount() int

type WorkerPool 

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

func NewWorkerPool 

func NewWorkerPool(workers, queueSize int, logger *slog.Logger) *WorkerPool

func (*WorkerPool) GetActiveWorkers 

func (p *WorkerPool) GetActiveWorkers() int

func (*WorkerPool) GetQueueCapacity 

func (p *WorkerPool) GetQueueCapacity() int

func (*WorkerPool) Metrics 

func (p *WorkerPool) Metrics() WorkerPoolMetrics

func (*WorkerPool) QueueSize 

func (p *WorkerPool) QueueSize() int

func (*WorkerPool) Shutdown 

func (p *WorkerPool) Shutdown()

func (*WorkerPool) Submit 

func (p *WorkerPool) Submit(req Request, resultCh chan *Response, errorCh chan error) error

type WorkerPoolMetrics 

type WorkerPoolMetrics struct {
	SubmittedTasks  int64
	CompletedTasks  int64
	FailedTasks     int64
	QueuedTasks     int64
	RejectedTasks   int64
	AverageWaitTime time.Duration
}

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.