PHP code example of stietotalwin / stw-sdk-php-redis-queue

    
        
<?php
require_once('vendor/autoload.php');

/* Start to develop here. Best regards https://php-download.com/ */

    

stietotalwin / stw-sdk-php-redis-queue example snippets

namespace Config;

use StieTotalWin\RedisQueue\Config\RedisQueue as BaseRedisQueue;

class RedisQueue extends BaseRedisQueue
{
    // Redis connection settings
    public $scheme = 'tcp';              // or 'tls' for secure connections
    public $host = '127.0.0.1';
    public $port = 6379;
    public $user = null;                 // for cloud Redis services (Upstash, etc.)
    public $password = null;
    public $database = 0;                // Optional for cloud services, set to null
    public $parameters = [               // Additional Predis parameters
        'timeout' => 30,
        'read_write_timeout' => 30,
    ];
    
    // Queue settings
    public $defaultQueue = 'default';
    public $maxAttempts = 3;
    public $retryDelay = 60;
    public $jobTtl = 86400;
    public $cleanupInterval = 3600;
}

namespace Config;

use StieTotalWin\RedisQueue\Config\RedisQueue as BaseRedisQueue;

class RedisQueue extends BaseRedisQueue
{
    public $scheme = 'tls';
    public $host = 'your-redis-host.upstash.io';
    public $port = 6379;
    public $user = 'default';              // Cloud services use 'user' instead of 'username'
    public $password = 'your-redis-password';
    public $database = null;               // Optional for cloud services
    public $parameters = [
        'timeout' => 30,
        'read_write_timeout' => 30,
    ];
}

use StieTotalWin\RedisQueue\Publisher;

// Initialize publisher with config
$config = config('RedisQueue');
$publisher = new Publisher($config, 'default');

// Publish a simple job (data as JSON string)
$jobId = $publisher->publish('send_email', json_encode([
    'to' => '[email protected]',
    'subject' => 'Welcome!',
    'message' => 'Thank you for joining us.'
]));

// Publish a delayed job (process after 5 minutes)
$jobId = $publisher->publish('send_newsletter', json_encode([
    'template' => 'monthly',
    'users' => [1, 2, 3]
]), 300);

// Publish to a specific queue
$jobId = $publisher->setQueue('high_priority')
    ->publish('urgent_task', json_encode(['data' => 'important']));

// Bulk publish (data will be automatically JSON encoded if it's an array)
$jobs = [
    ['type' => 'send_email', 'data' => ['to' => '[email protected]']],
    ['type' => 'send_email', 'data' => ['to' => '[email protected]'], 'delay' => 60],
];
$jobIds = $publisher->publishBulk($jobs, 'email_queue');

use StieTotalWin\RedisQueue\Consumer;

// Initialize consumer with same config
$config = config('RedisQueue');
$consumer = new Consumer($config);

// Consume jobs from a queue
while (true) {
    $job = $consumer->consume('default');
    
    if ($job) {
        try {
            // Process the job (data is returned as JSON string)
            $jobData = json_decode($job->getData(), true);
            processJob($job->getType(), $jobData);
            
            // Mark as completed
            $consumer->markCompleted($job->getId(), 'default');
            
        } catch (Exception $e) {
            // Mark as failed with error message
            $consumer->markFailed($job->getId(), 'default', $e->getMessage());
        }
    } else {
        // No jobs available, wait before checking again
        sleep(1);
    }
}

function processJob($type, $data) {
    switch ($type) {
        case 'send_email':
            // Send email logic
            sendEmail($data['to'], $data['subject'], $data['message']);
            break;
            
        case 'send_newsletter':
            // Newsletter logic
            sendNewsletter($data['template'], $data['users']);
            break;
            
        default:
            throw new Exception("Unknown job type: {$type}");
    }
}

// Get queue statistics
$stats = $publisher->getQueueStats('default');
/*
Array:
[
    'queue_name' => 'default',
    'pending_jobs' => 5,
    'processing_jobs' => 2,
    'failed_jobs' => 1,
    'total_job_data' => 8
]
*/

// Get job count
$count = $publisher->getJobCount('default');

// Get all queue names
$queues = $publisher->getQueueNames();

// Clear a queue (removes all jobs)
$publisher->clearQueue('default');

// Get a specific job
$job = $publisher->getJob($jobId, 'default');
if ($job) {
    echo "Job Type: " . $job->getType();
    echo "Job Data: " . json_encode($job->getData());
    echo "Status: " . $job->getStatus();
}

// Delete a specific job
$publisher->deleteJob($jobId, 'default');

// Get stuck jobs (jobs processing for too long)
$stuckJobs = $publisher->getStuckJobs('default', 10); // stuck for 10+ minutes

// Republish stuck jobs back to the queue
$republishedCount = $publisher->republishStuckJobs('default', 10);
echo "Republished {$republishedCount} stuck jobs";

// Get failed jobs
$failedJobs = $consumer->getFailedJobs('default', 10);

foreach ($failedJobs as $job) {
    echo "Failed Job: " . $job->getType();
    echo "Error: " . $job->getError();
    echo "Attempts: " . $job->getAttempts();
}

// Retry failed jobs
$retriedCount = $consumer->retryFailedJobs('default', 5);
echo "Retried {$retriedCount} jobs";

// Get processing jobs
$processingJobs = $consumer->getProcessingJobs('default');

// Clean up expired jobs
$cleanedCount = $consumer->cleanupExpiredJobs('default', 86400);
echo "Cleaned {$cleanedCount} expired jobs";

// Comprehensive cleanup (more thorough)
$cleanupStats = $consumer->comprehensiveCleanup('default', 86400, 100);
echo "Cleanup stats: " . json_encode($cleanupStats);

// Recover lost jobs and get recovery statistics
$recoveryStats = $consumer->recoverLostJobs('default', 3600);
echo "Recovery stats: " . json_encode($recoveryStats);

// Get recovery recommendations
$recoveryInfo = $consumer->getRecoveryStats('default');
if ($recoveryInfo['recovery_recommended']) {
    echo "Recovery is recommended for queue 'default'";
}

// Enable detailed logging for debugging
$consumer->enableDetailedLogging(true);

// Get pending jobs
$pendingJobs = $consumer->getPendingJobs('default', 10);

$job = $consumer->consume('default');

if ($job) {
    echo $job->getId();          // Unique job ID (UUID)
    echo $job->getType();        // Job type
    echo $job->getData();        // Job data (JSON string)
    echo $job->getStatus();      // Current status
    echo $job->getAttempts();    // Number of attempts
    echo $job->getMaxAttempts(); // Maximum allowed attempts
    echo $job->getCreatedAt();   // Creation timestamp
    echo $job->getProcessAt();   // When to process (for delayed jobs)
    echo $job->getUpdatedAt();   // Last update timestamp
    echo $job->getError();       // Error message (if failed)
    
    // Check if job can be retried
    if ($job->canRetry()) {
        echo "Job can be retried";
    }
    
    // Check if job is expired
    if ($job->isExpired()) {
        echo "Job has expired";
    }
}

// Check if recovery is recommended
$recoveryStats = $consumer->getRecoveryStats('default');
if ($recoveryStats['recovery_recommended']) {
    // Run recovery process
    $recovered = $consumer->recoverLostJobs('default');
    echo "Recovered jobs: " . json_encode($recovered);
}

// Performance-optimized cleanup for large queues
$cleanupStats = $consumer->comprehensiveCleanup('default', 86400, 100);
echo "Cleanup completed in {$cleanupStats['execution_time']}ms";

bash
composer