package/mq
1
0
Fork 0
mirror of https://github.com/oarkflow/mq.git synced 2025-10-06 00:16:49 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
sujit
2025-08-04 22:39:17 +05:45
parent c1b09d9c4d
commit 2786e64900
8 changed files with 85 additions and 916 deletions
Download Patch File Download Diff File Expand all files Collapse all files

106
ADMIN_SUCCESS.md
View File

@@ -1,106 +0,0 @@
# Admin Server Working! 🎉
The MQ Admin Server is now successfully running on port 8090!
## ✅ What was fixed:
The issue was that the `broker.Start(ctx)` method contains an infinite loop to accept connections, which was blocking the main thread. The solution was to start the broker in a goroutine.
## 🔧 Key Fix:
```go
// Start broker in goroutine since it blocks
go func() {
if err := broker.Start(ctx); err != nil {
log.Printf("Broker error: %v", err)
}
}()
// Give broker time to start
time.Sleep(500 * time.Millisecond)
```
## 🌐 Available Endpoints:
The admin server is now responding on the following endpoints:
- **Dashboard**: http://localhost:8090/admin
- **Health Check**: http://localhost:8090/api/admin/health
- **Broker Info**: http://localhost:8090/api/admin/broker
- **Queues**: http://localhost:8090/api/admin/queues
- **Consumers**: http://localhost:8090/api/admin/consumers
- **Pools**: http://localhost:8090/api/admin/pools
- **Metrics**: http://localhost:8090/api/admin/metrics
## 🧪 Test Results:
```bash
$ curl -s http://localhost:8090/api/admin/health | jq .
[
{
"name": "Broker Health",
"status": "healthy",
"message": "Broker is running normally",
"duration": 5000000,
"timestamp": "2025-07-29T23:59:21.452419+05:45"
},
{
"name": "Memory Usage",
"status": "healthy",
"message": "Memory usage is within normal limits",
"duration": 2000000,
"timestamp": "2025-07-29T23:59:21.452419+05:45"
},
{
"name": "Queue Health",
"status": "healthy",
"message": "All queues are operational",
"duration": 3000000,
"timestamp": "2025-07-29T23:59:21.452419+05:45"
}
]
```
```bash
$ curl -s http://localhost:8090/api/admin/broker | jq .
{
"status": "running",
"address": ":54019",
"uptime": 24804,
"connections": 0,
"config": {
"max_connections": 1000,
"queue_size": 100,
"read_timeout": "30s",
"sync_mode": false,
"worker_pool": false,
"write_timeout": "30s"
}
}
```
## 🚀 Usage:
```bash
# Run the minimal admin demo
cd examples/minimal_admin
go run main.go
# Or run the full admin demo
cd examples/admin
go run main.go
```
Both should now work correctly with the broker starting in a goroutine!
## 📁 Files Created:
1. **admin_server.go** - Main admin server implementation
2. **static/admin/index.html** - Admin dashboard UI
3. **static/admin/css/admin.css** - Dashboard styling
4. **static/admin/js/admin.js** - Dashboard JavaScript
5. **examples/admin/main.go** - Admin demo (fixed)
6. **examples/minimal_admin/main.go** - Minimal working demo
7. **static/admin/README.md** - Documentation
The admin dashboard is now fully functional with real-time monitoring capabilities!

236
IMPLEMENTATION_SUMMARY.md
View File

@@ -1,236 +0,0 @@
# Enhanced Worker Pool - Implementation Summary
## Overview
I have successfully analyzed and enhanced your worker pool implementation to make it production-ready and fault-tolerant. The improvements address critical issues and add essential features for enterprise-scale deployments.
## Critical Issues Fixed
### 1. Race Conditions and Deadlocks ✅
**Issues Found:**
- Improper synchronization in worker lifecycle
- Potential deadlocks during shutdown
- Race conditions in task queue access
**Fixes Applied:**
- Proper condition variable usage with mutex protection
- Graceful shutdown with coordinated worker termination
- Atomic operations for shared state management
- Panic recovery in workers with automatic restart
### 2. Memory Management ✅
**Issues Found:**
- No memory usage tracking or limits
- Potential memory leaks in error scenarios
- Uncontrolled resource consumption
**Fixes Applied:**
- Real-time memory usage tracking and enforcement
- Overflow buffer with size limits to prevent OOM
- Task expiration checking and cleanup
- Proper resource cleanup on shutdown
### 3. Error Handling and Resilience ✅
**Issues Found:**
- Basic retry logic without proper backoff
- No circuit breaker implementation
- Poor error classification and handling
**Fixes Applied:**
- Exponential backoff with jitter and maximum caps
- Production-ready circuit breaker with failure counting
- Enhanced Dead Letter Queue with metadata and analytics
- Comprehensive error recovery mechanisms
### 4. Worker Management ✅
**Issues Found:**
- Inefficient worker scaling
- No worker health monitoring
- Poor load-based adjustments
**Fixes Applied:**
- Intelligent dynamic worker scaling based on actual load
- Proper worker lifecycle management
- Worker starvation detection and recovery
- Graceful worker shutdown with timeout handling
### 5. Task Processing ✅
**Issues Found:**
- No task validation or sanitization
- Missing timeout enforcement
- No extensibility for custom processing
**Fixes Applied:**
- Comprehensive task validation and expiration checking
- Plugin system for extensible task processing
- Proper timeout enforcement with context cancellation
- Enhanced task metadata and tracing support
## New Production Features Added
### 1. Health Monitoring System 🆕
- Comprehensive health status reporting
- Automatic issue detection and classification
- Performance metrics and threshold monitoring
- REST API endpoints for monitoring integration
### 2. Enhanced Dead Letter Queue 🆕
- Task categorization by error type
- Automatic cleanup of old failed tasks
- Statistical analysis and reporting
- Reprocessing capabilities for recovery
### 3. Auto-Recovery System 🆕
- Circuit breaker reset capabilities
- Worker pool recovery mechanisms
- Queue drainage and optimization
- Failure scenario detection and handling
### 4. Advanced Configuration Management 🆕
- Runtime configuration updates with validation
- Production-ready default configurations
- Environment-specific configuration profiles
- Dynamic parameter adjustment
### 5. Metrics and Observability 🆕
- Prometheus-compatible metrics export
- Real-time performance monitoring
- Latency percentile tracking (P95, P99)
- JSON and HTTP APIs for metrics access
## Performance Improvements
### Memory Efficiency
- Reduced memory allocations through object pooling
- Efficient memory usage tracking
- Overflow buffer management to prevent OOM
- Automatic cleanup of expired tasks
### Throughput Optimization
- Batch processing for improved performance
- Intelligent worker scaling based on load
- Priority queue optimization
- Reduced lock contention
### Latency Reduction
- Faster task enqueueing with validation
- Optimized queue operations
- Reduced synchronization overhead
- Improved error handling paths
## Code Quality Enhancements
### Robustness
- Comprehensive error handling at all levels
- Panic recovery and worker restart capabilities
- Graceful degradation under high load
- Resource leak prevention
### Maintainability
- Clear separation of concerns
- Extensive documentation and comments
- Consistent error handling patterns
- Modular and extensible design
### Testability
- Comprehensive test suite included
- Benchmark tests for performance validation
- Mock interfaces for unit testing
- Integration test scenarios
## Files Modified/Created
### Core Implementation
- `pool.go` - Enhanced with all production features
- `pool_test.go` - Comprehensive test suite
- `PRODUCTION_READINESS_REPORT.md` - Detailed analysis and recommendations
### Existing Files Enhanced
- Improved synchronization and error handling
- Added health monitoring capabilities
- Enhanced metrics collection
- Better resource management
## Usage Example
```go
// Create production-ready pool
pool := mq.NewPool(10,
mq.WithHandler(yourHandler),
mq.WithTaskStorage(storage),
mq.WithCircuitBreaker(mq.CircuitBreakerConfig{
Enabled: true,
FailureThreshold: 5,
ResetTimeout: 30 * time.Second,
}),
mq.WithMaxMemoryLoad(512 * 1024 * 1024), // 512MB
mq.WithBatchSize(10),
)
// Monitor health
health := pool.GetHealthStatus()
if !health.IsHealthy {
log.Printf("Pool issues: %v", health.Issues)
}
// Get metrics
metrics := pool.FormattedMetrics()
log.Printf("Throughput: %.2f tasks/sec", metrics.TasksPerSecond)
// Graceful shutdown
pool.Stop()
```
## Production Deployment Readiness
### ✅ Completed Features
- Fault tolerance and error recovery
- Memory management and limits
- Circuit breaker implementation
- Health monitoring and metrics
- Graceful shutdown handling
- Dynamic worker scaling
- Enhanced Dead Letter Queue
- Configuration management
### 🔄 Recommended Next Steps
1. **Monitoring Integration**
- Set up Prometheus metrics collection
- Configure Grafana dashboards
- Implement alerting rules
2. **Persistence Layer**
- Integrate with PostgreSQL/Redis for task persistence
- Implement backup and recovery procedures
- Add transaction support for critical operations
3. **Security Enhancements**
- Implement task encryption for sensitive data
- Add authentication and authorization
- Enable audit logging
4. **Distributed Processing**
- Add cluster coordination capabilities
- Implement load balancing across nodes
- Add service discovery integration
## Performance Benchmarks
The enhanced worker pool demonstrates significant improvements:
- **Throughput**: 50-100% increase in tasks/second
- **Memory Usage**: 30-40% reduction in memory overhead
- **Error Recovery**: 95% faster failure detection and recovery
- **Latency**: Consistent P99 latency under high load
- **Reliability**: 99.9% uptime with proper circuit breaker tuning
## Conclusion
Your worker pool is now production-ready with enterprise-grade features:
- **Fault Tolerant**: Handles failures gracefully with automatic recovery
- **Scalable**: Dynamic worker management based on load
- **Observable**: Comprehensive metrics and health monitoring
- **Maintainable**: Clean, well-documented, and tested codebase
- **Configurable**: Runtime configuration updates without restarts
The implementation follows Go best practices and is ready for high-scale production deployments. The enhanced features provide the foundation for building robust, distributed task processing systems.

303
PRODUCTION_ANALYSIS.md
View File

@@ -1,303 +0,0 @@
# Production Message Queue Issues Analysis & Fixes
## Executive Summary
This analysis identified critical issues in the existing message queue implementation that prevent it from being production-ready. The issues span across connection management, error handling, concurrency, resource management, and missing enterprise features.
## Critical Issues Identified
### 1. Connection Management Issues
**Problems Found:**
- Race conditions in connection pooling
- No connection health checks
- Improper connection cleanup leading to memory leaks
- Missing connection timeout handling
- Shared connection state without proper synchronization
**Fixes Implemented:**
- Enhanced connection pool with proper synchronization
- Health checker with periodic connection validation
- Atomic flags for connection state management
- Proper connection lifecycle management with cleanup
- Connection reuse with health validation
### 2. Error Handling & Recovery
**Problems Found:**
- Insufficient error handling in critical paths
- No circuit breaker for cascading failure prevention
- Missing proper timeout handling
- Inadequate retry mechanisms
- Error propagation issues
**Fixes Implemented:**
- Circuit breaker pattern implementation
- Comprehensive error wrapping and context
- Timeout handling with context cancellation
- Exponential backoff with jitter for retries
- Graceful degradation mechanisms
### 3. Concurrency & Thread Safety
**Problems Found:**
- Race conditions in task processing
- Unprotected shared state access
- Potential deadlocks in shutdown procedures
- Goroutine leaks in error scenarios
- Missing synchronization primitives
**Fixes Implemented:**
- Proper mutex usage for shared state protection
- Atomic operations for flag management
- Graceful shutdown with wait groups
- Context-based cancellation throughout
- Thread-safe data structures
### 4. Resource Management
**Problems Found:**
- No proper cleanup mechanisms
- Missing graceful shutdown implementation
- Incomplete memory usage tracking
- Resource leaks in error paths
- No limits on resource consumption
**Fixes Implemented:**
- Comprehensive resource cleanup
- Graceful shutdown with configurable timeouts
- Memory usage monitoring and limits
- Resource pool management
- Automatic cleanup routines
### 5. Production Features Missing
**Problems Found:**
- No message persistence
- No message ordering guarantees
- No cluster support
- Limited monitoring and observability
- No configuration management
- Missing security features
- No rate limiting
- No dead letter queues
**Fixes Implemented:**
- Message persistence interface with implementations
- Production-grade monitoring system
- Comprehensive configuration management
- Security features (TLS, authentication)
- Rate limiting for all components
- Dead letter queue implementation
- Health checking system
- Metrics collection and alerting
## Architectural Improvements
### 1. Enhanced Broker (`broker_enhanced.go`)
```go
type EnhancedBroker struct {
*Broker
connectionPool *ConnectionPool
healthChecker *HealthChecker
circuitBreaker *EnhancedCircuitBreaker
metricsCollector *MetricsCollector
messageStore MessageStore
// ... additional production features
}
```
**Features:**
- Connection pooling with health checks
- Circuit breaker for fault tolerance
- Message persistence
- Comprehensive metrics collection
- Automatic resource cleanup
### 2. Production Configuration (`config_manager.go`)
```go
type ProductionConfig struct {
Broker BrokerConfig
Consumer ConsumerConfig
Publisher PublisherConfig
Pool PoolConfig
Security SecurityConfig
Monitoring MonitoringConfig
Persistence PersistenceConfig
Clustering ClusteringConfig
RateLimit RateLimitConfig
}
```
**Features:**
- Hot configuration reloading
- Configuration validation
- Environment-specific configs
- Configuration watchers for dynamic updates
### 3. Monitoring & Observability (`monitoring.go`)
```go
type MetricsServer struct {
registry *DetailedMetricsRegistry
healthChecker *SystemHealthChecker
alertManager *AlertManager
// ... monitoring components
}
```
**Features:**
- Real-time metrics collection
- Health checking with thresholds
- Alert management with notifications
- Performance monitoring
- Resource usage tracking
### 4. Enhanced Consumer (`consumer.go` - Updated)
**Improvements:**
- Connection health monitoring
- Automatic reconnection with backoff
- Circuit breaker integration
- Proper resource cleanup
- Enhanced error handling
- Rate limiting support
## Security Enhancements
### 1. TLS Support
- Mutual TLS authentication
- Certificate validation
- Secure connection management
### 2. Authentication & Authorization
- Pluggable authentication mechanisms
- Role-based access control
- Session management
### 3. Data Protection
- Message encryption at rest and in transit
- Audit logging
- Secure configuration management
## Performance Optimizations
### 1. Connection Pooling
- Reusable connections
- Connection health monitoring
- Automatic cleanup of idle connections
### 2. Rate Limiting
- Broker-level rate limiting
- Consumer-level rate limiting
- Per-queue rate limiting
- Burst handling
### 3. Memory Management
- Memory usage monitoring
- Configurable memory limits
- Garbage collection optimization
- Resource pool management
## Reliability Features
### 1. Message Persistence
- Configurable storage backends
- Message durability guarantees
- Automatic cleanup of expired messages
### 2. Dead Letter Queues
- Failed message handling
- Retry mechanisms
- Message inspection capabilities
### 3. Circuit Breaker
- Failure detection
- Automatic recovery
- Configurable thresholds
### 4. Health Monitoring
- System health checks
- Component health validation
- Automated alerting
## Deployment Considerations
### 1. Configuration Management
- Environment-specific configurations
- Hot reloading capabilities
- Configuration validation
### 2. Monitoring Setup
- Metrics endpoints
- Health check endpoints
- Alert configuration
### 3. Scaling Considerations
- Horizontal scaling support
- Load balancing
- Resource allocation
## Testing Recommendations
### 1. Load Testing
- High-throughput scenarios
- Connection limits testing
- Memory usage under load
### 2. Fault Tolerance Testing
- Network partition testing
- Service failure scenarios
- Recovery time validation
### 3. Security Testing
- Authentication bypass attempts
- Authorization validation
- Data encryption verification
## Migration Strategy
### 1. Gradual Migration
- Feature-by-feature replacement
- Backward compatibility maintenance
- Monitoring during transition
### 2. Configuration Migration
- Configuration schema updates
- Default value establishment
- Validation implementation
### 3. Performance Validation
- Benchmark comparisons
- Resource usage monitoring
- Regression testing
## Key Files Created/Modified
1. **broker_enhanced.go** - Production-ready broker with all enterprise features
2. **config_manager.go** - Comprehensive configuration management
3. **monitoring.go** - Complete monitoring and alerting system
4. **consumer.go** - Enhanced with proper error handling and resource management
5. **examples/production_example.go** - Production deployment example
## Summary
The original message queue implementation had numerous critical issues that would prevent successful production deployment. The implemented fixes address all major concerns:
- **Reliability**: Circuit breakers, health monitoring, graceful shutdown
- **Performance**: Connection pooling, rate limiting, resource management
- **Observability**: Comprehensive metrics, health checks, alerting
- **Security**: TLS, authentication, audit logging
- **Maintainability**: Configuration management, hot reloading, structured logging
The enhanced implementation now provides enterprise-grade reliability, performance, and operational capabilities suitable for production environments.
## Next Steps
1. **Testing**: Implement comprehensive test suite for all new features
2. **Documentation**: Create operational runbooks and deployment guides
3. **Monitoring**: Set up alerting and dashboard for production monitoring
4. **Performance**: Conduct load testing and optimization
5. **Security**: Perform security audit and penetration testing

265
PRODUCTION_READINESS_REPORT.md
View File

@@ -1,265 +0,0 @@
# Worker Pool Production Readiness Report
## Critical Issues Fixed
### 1. **Race Conditions and Deadlocks**
- **Fixed**: Worker synchronization using proper condition variables
- **Fixed**: Eliminated potential deadlocks in shutdown process
- **Added**: Panic recovery in workers with automatic restart
- **Added**: Proper task completion tracking with WaitGroup
### 2. **Memory Management and Resource Leaks**
- **Fixed**: Memory usage tracking and enforcement
- **Added**: Overflow buffer with size limits
- **Added**: Task expiration checking
- **Added**: Proper resource cleanup on shutdown
- **Added**: Memory threshold monitoring with warnings
### 3. **Error Handling and Resilience**
- **Enhanced**: Circuit breaker with proper failure counting
- **Added**: Exponential backoff with jitter and maximum caps
- **Enhanced**: Dead Letter Queue with metadata and management
- **Added**: Task retry logic with proper failure tracking
- **Added**: Health check system with issue detection
### 4. **Worker Management**
- **Fixed**: Dynamic worker scaling based on actual load
- **Added**: Proper worker lifecycle management
- **Added**: Graceful worker shutdown
- **Added**: Worker starvation detection
### 5. **Task Processing**
- **Added**: Task validation and sanitization
- **Added**: Plugin system for extensible processing
- **Added**: Task execution timeout enforcement
- **Added**: Comprehensive error recovery
## New Production Features Added
### 1. **Health Monitoring System**
```go
type PoolHealthStatus struct {
IsHealthy bool `json:"is_healthy"`
WorkerCount int32 `json:"worker_count"`
QueueDepth int `json:"queue_depth"`
OverflowDepth int `json:"overflow_depth"`
CircuitBreakerOpen bool `json:"circuit_breaker_open"`
ErrorRate float64 `json:"error_rate"`
// ... more metrics
}
```
### 2. **Enhanced Dead Letter Queue**
- Task categorization by error type
- Automatic cleanup of old failed tasks
- Statistics and analytics
- Reprocessing capabilities
### 3. **Auto-Recovery System**
- Circuit breaker reset capabilities
- Worker pool recovery
- Queue drainage mechanisms
- Failure scenario detection
### 4. **Advanced Configuration Management**
- Runtime configuration updates
- Configuration validation
- Production-ready defaults
- Environment-specific configs
## Essential New Features to Implement
### 1. **Observability and Monitoring**
```go
// Metrics and monitoring integration
func (wp *Pool) SetupPrometheus() error
func (wp *Pool) SetupJaegerTracing() error
func (wp *Pool) ExportMetrics() MetricsSnapshot
// Distributed tracing
type TaskTrace struct {
TraceID string
SpanID string
ParentSpan string
StartTime time.Time
EndTime time.Time
Tags map[string]string
}
```
### 2. **Advanced Persistence Layer**
```go
// Database persistence for production
type PostgresTaskStorage struct {
db *sql.DB
// Connection pooling
// Transactions
// Bulk operations
}
// Redis-based storage for high performance
type RedisTaskStorage struct {
client redis.Client
// Clustering support
// Persistence options
}
```
### 3. **Security Enhancements**
```go
// Task encryption for sensitive data
type EncryptedTask struct {
EncryptedPayload []byte
Algorithm string
KeyID string
}
// Role-based access control
type TaskPermissions struct {
AllowedRoles []string
RequiredClaims map[string]string
}
```
### 4. **Advanced Queue Management**
```go
// Priority-based routing
type TaskRouter struct {
rules []RoutingRule
}
// Queue partitioning for better performance
type PartitionedQueue struct {
partitions map[string]*Queue
strategy PartitionStrategy
}
```
### 5. **Distributed Processing**
```go
// Cluster coordination
type ClusterCoordinator struct {
nodes []ClusterNode
elector LeaderElector
discovery ServiceDiscovery
}
// Load balancing across nodes
type TaskDistributor struct {
nodes []WorkerNode
balancer LoadBalancer
}
```
### 6. **Advanced Error Handling**
```go
// Sophisticated retry policies
type RetryPolicy struct {
MaxRetries int
BackoffFunc func(attempt int) time.Duration
RetryIf func(error) bool
OnRetry func(attempt int, err error)
}
// Error classification and routing
type ErrorClassifier struct {
patterns map[string]ErrorHandler
}
```
### 7. **Performance Optimization**
```go
// Task batching for improved throughput
type BatchProcessor struct {
maxBatchSize int
timeout time.Duration
processor func([]Task) []Result
}
// Worker affinity for cache locality
type WorkerAffinity struct {
cpuSet []int
numaNode int
taskTypes []string
}
```
### 8. **API and Management Interface**
```go
// REST API for management
type PoolAPI struct {
pool *Pool
mux *http.ServeMux
}
// Real-time dashboard
type Dashboard struct {
websocket *websocket.Conn
metrics chan MetricsUpdate
}
```
## Production Deployment Checklist
### Infrastructure Requirements
- [ ] Database setup (PostgreSQL/Redis) for persistence
- [ ] Monitoring stack (Prometheus + Grafana)
- [ ] Logging aggregation (ELK/Loki)
- [ ] Service mesh integration (Istio/Linkerd)
- [ ] Load balancer configuration
- [ ] Backup and disaster recovery plan
### Configuration
- [ ] Production configuration validation
- [ ] Resource limits and quotas
- [ ] Circuit breaker thresholds
- [ ] Monitoring and alerting rules
- [ ] Security policies and encryption
- [ ] Network policies and firewall rules
### Testing
- [ ] Load testing with realistic workloads
- [ ] Chaos engineering tests
- [ ] Failure scenario testing
- [ ] Performance benchmarking
- [ ] Security vulnerability assessment
- [ ] Configuration drift detection
### Operational Procedures
- [ ] Deployment procedures
- [ ] Rollback procedures
- [ ] Incident response playbooks
- [ ] Capacity planning guidelines
- [ ] Performance tuning procedures
- [ ] Backup and restore procedures
## Recommended Implementation Priority
1. **High Priority** (Immediate)
- Enhanced monitoring and metrics
- Persistent storage integration
- Security hardening
- API management interface
2. **Medium Priority** (Next Sprint)
- Distributed processing capabilities
- Advanced retry and error handling
- Performance optimization features
- Comprehensive testing suite
3. **Lower Priority** (Future Releases)
- Advanced analytics and ML integration
- Multi-tenancy support
- Plugin ecosystem
- Advanced clustering features
## Performance Benchmarks to Establish
- Tasks processed per second under various loads
- Memory usage patterns and efficiency
- Latency percentiles (P50, P95, P99)
- Worker scaling responsiveness
- Error recovery time
- Circuit breaker effectiveness
The enhanced worker pool is now production-ready with robust error handling, proper resource management, and comprehensive monitoring capabilities. The suggested features will further enhance its capabilities for enterprise-scale deployments.

1
examples/priority.go
View File

@@ -15,6 +15,7 @@ func main() {
mq.WithHandler(tasks.SchedulerHandler), mq.WithHandler(tasks.SchedulerHandler),
mq.WithPoolCallback(tasks.SchedulerCallback), mq.WithPoolCallback(tasks.SchedulerCallback),
mq.WithTaskStorage(mq.NewMemoryTaskStorage(10*time.Minute)), mq.WithTaskStorage(mq.NewMemoryTaskStorage(10*time.Minute)),
mq.WithDiagnostics(false),
) )
for i := 0; i < 100; i++ { for i := 0; i < 100; i++ {

79
examples/schema.json Normal file
View File

@@ -0,0 +1,79 @@
{
"type": "object",
"properties": {
"first_name": {
"type": "string",
"title": "👤 First Name",
"order": 1,
"ui": {
"control": "input",
"class": "form-group",
"name": "first_name"
}
},
"last_name": {
"type": "string",
"title": "👤 Last Name",
"order": 2,
"ui": {
"control": "input",
"class": "form-group",
"name": "last_name"
}
},
"email": {
"type": "email",
"title": "📧 Email Address",
"order": 3,
"ui": {
"control": "input",
"type": "email",
"class": "form-group",
"name": "email"
}
},
"user_type": {
"type": "string",
"title": "👥 User Type",
"order": 4,
"ui": {
"control": "select",
"class": "form-group",
"name": "user_type",
"options": [ "new", "premium", "standard" ]
}
},
"priority": {
"type": "string",
"title": "🚨 Priority Level",
"order": 5,
"ui": {
"control": "select",
"class": "form-group",
"name": "priority",
"options": [ "low", "medium", "high", "urgent" ]
}
},
"subject": {
"type": "string",
"title": "📋 Subject",
"order": 6,
"ui": {
"control": "input",
"class": "form-group",
"name": "subject"
}
},
"message": {
"type": "textarea",
"title": "💬 Message",
"order": 7,
"ui": {
"control": "textarea",
"class": "form-group",
"name": "message"
}
}
},
"required": [ "first_name", "last_name", "email", "user_type", "priority", "subject", "message" ]
}

4
examples/tasks/scheduler.go
View File

@@ -14,9 +14,7 @@ func SchedulerHandler(ctx context.Context, task *mq.Task) mq.Result {
func SchedulerCallback(ctx context.Context, result mq.Result) error { func SchedulerCallback(ctx context.Context, result mq.Result) error {
if result.Error != nil { if result.Error != nil {
fmt.Println("Task failed!") fmt.Println("Task failed!", result.Error.Error())
} else {
fmt.Println("Task completed successfully.")
} }
return nil return nil
} }

5
pool.go
View File

@@ -211,7 +211,6 @@ func (m *InMemoryMetricsRegistry) Register(metricName string, value interface{})
defer m.mu.Unlock() defer m.mu.Unlock()
if v, ok := value.(int64); ok { if v, ok := value.(int64); ok {
m.metrics[metricName] = v m.metrics[metricName] = v
Logger.Info().Str("metric", metricName).Msgf("Registered metric: %d", v)
} }
} }
@@ -627,7 +626,7 @@ func (wp *Pool) handleTaskFailure(task *QueueTask, result Result) {
} }
// handleTaskSuccess processes successful task completion // handleTaskSuccess processes successful task completion
func (wp *Pool) handleTaskSuccess(task *QueueTask, result Result, ctx context.Context) { func (wp *Pool) handleTaskSuccess(task *QueueTask, _ Result, _ context.Context) {
// Reset circuit breaker failure count on success // Reset circuit breaker failure count on success
if wp.circuitBreaker.Enabled { if wp.circuitBreaker.Enabled {
atomic.StoreInt32(&wp.circuitBreakerFailureCount, 0) atomic.StoreInt32(&wp.circuitBreakerFailureCount, 0)
@@ -809,7 +808,9 @@ func (wp *Pool) EnqueueTask(ctx context.Context, payload *Task, priority int) er
// Update metrics // Update metrics
atomic.AddInt64(&wp.metrics.TotalScheduled, 1) atomic.AddInt64(&wp.metrics.TotalScheduled, 1)
if wp.diagnosticsEnabled {
wp.logger.Debug().Str("taskID", payload.ID).Msgf("Task enqueued with priority %d, queue depth: %d", priority, queueLen) wp.logger.Debug().Str("taskID", payload.ID).Msgf("Task enqueued with priority %d, queue depth: %d", priority, queueLen)
}
return nil return nil
} }