PHP code example of onlyphp / simple-queue

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

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

    

onlyphp / simple-queue example snippets

use OnlyPHP\SimpleQueue\JobProcessor;

// PDO MySQL
$pdo = new PDO('mysql:host=localhost;dbname=your_database', 'username', 'password');
$queue = new JobProcessor($pdo);

// PDO MSSQL
$pdoMssql = new PDO('sqlsrv:Server=localhost;Database=your_database', 'username', 'password');
$queueMssql = new JobProcessor($pdoMssql);

// PDO Oracle
$pdoOracle = new PDO('oci:dbname=your_database', 'username', 'password');
$queueOracle = new JobProcessor($pdoOracle);

// CodeIgniter 3
$queue = new JobProcessor($this->db);

// CodeIgniter 4
$db = \Config\Database::connect();
$queue = new JobProcessor($db);

// Laravel
$queue = new JobProcessor(DB::connection());

// Dispatch a closure
$queue->job(function() {
    echo "Processing job...";
})->dispatch();

// Dispatch a function
$queue->job('process_data', ['param1', 'param2'])->setIncludePathFile('/path/to/file.php')->dispatch();

// Dispatch a class method
$queue->job([new YourClass(), 'methodName'], $params)->dispatch();

// Execute job immediately without queueing (foreground processed) using closure/callable
$result = $queue->job($callable)->dispatchNow();

class EmailService {
    public function sendBulkEmails(array $recipients, string $template) {
        foreach ($recipients as $recipient) {
            // Process each email
        }
    }
}

// Queue the email job
$emailService = new EmailService();
$recipients = ['[email protected]', '[email protected]'];

$queue->job([$emailService, 'sendBulkEmails'], $recipients)
    ->setName('bulk-email-campaign')
    ->setPriority('high')
    ->setMaxRetries(3)
    ->setRetryDelay(60)
    ->dispatch();

class FileProcessor {
    public function processLargeFile(string $filePath) {
        // Process large file
    }
}

// Queue file processing
$processor = new FileProcessor();
$queue->job([$processor, 'processLargeFile'], $filePath)
    ->setName('large-file-processing')
    ->setTimeout(3600) // 1 hour timeout
    ->dispatch();

class OrderProcessor {
    public function processOrder(int $orderId, array $items) {
        // Process order logic
    }
}

// Queue order processing
$processor = new OrderProcessor();
$order = [
    'id' => 1234,
    'items' => ['item1', 'item2']
];

$queue->job([$processor, 'processOrder'], $order)
    ->setName("process-order-{$order['id']}")
    ->setPriority('urgent')
    ->setMaxRetries(5)
    ->setRetryDelay(30)
    ->dispatch();

class DataExporter {
    public function exportToCSV(string $query, string $filename) {
        // Export logic
    }
}

// Queue export job
$exporter = new DataExporter();
$queue->job([$exporter, 'exportToCSV'], $params)
    ->setName('monthly-report-export')
    ->setPriority('normal')
    ->setTimeout(7200) // 2 hours
    ->dispatch();

class ImageProcessor {
    public function processImage(string $path, array $options) {
        // Image processing logic
    }
}

// Queue image processing
$processor = new ImageProcessor();
$options = [
    'resize' => true,
    'width' => 800,
    'height' => 600,
    'optimize' => true
];

$queue->job([$processor, 'processImage'], $options)
    ->setName('image-processing')
    ->setPriority('low')
    ->setMaxRetries(2)
    ->dispatch();

function exportCsvData(string $filePath) {
    // Process to export data
}

// Queue file export processing
$queue->job('exportCsvData', $filePath)
    ->setIncludePathFile('/path/to/csv_helpers.php') // Include the file before called the function
    ->dispatch();

class MyInvokableClass {
    public function __invoke($param1, $param2) {
        // Invokable class logic
        echo "Processing invokable class with params: $param1, $param2";
    }
}

$invokable = new MyInvokableClass();
$processor->job($invokable, ['param1', 'param2'])
    ->dispatch();

$config = [
    'process_check_interval' => 1000000,  // Worker check interval (microseconds)
    'worker_timeout' => 3600,            // Worker timeout (seconds)
    'max_workers' => 1,                  // Maximum number of concurrent workers
    'lock_dir' => '/tmp'                 // Directory for worker lock files
];

$queue = new JobProcessor($connection, $config);

$processor = new JobProcessor($connection, $config);

// Get queue statistics
$stats = $processor->getJobStats();
print_r($stats);
/* Output:
[
    'total_jobs' => 100,
    'pending_jobs' => 10,
    'processing_jobs' => 5,
    'completed_jobs' => 80,
    'failed_jobs' => 5,
    'avg_processing_time' => 45.5
]
*/

// Get specific job status
$jobUuId = $processor->job($callable)->dispatch();
$status = $processor->getJobStatus($jobUuId);

// Retry failed jobs
$processor->retryAllFailed();

// Clear old failed jobs
$processor->clearFailedJobs(30); // Clear jobs older than 30 days