Code Examples

1import { ReliableQueue } from '@aplanka/reliable-queue';
2
3interface FileUploadData {
4  id: string;
5  fileName: string;
6  fileSize: number;
7  uploadUrl: string;
8  metadata?: {
9    userId: string;
10    folder: string;
11    tags?: string[];
12  };
13}
14
15// Create a file upload queue
16const uploadQueue = new ReliableQueue<FileUploadData>({
17  maxRetries: 3,
18  retryDelay: 1000,
19  exponentialBackoff: true,
20  persistent: true,
21  storageKey: 'file-upload-queue',
22  concurrency: 2, // Upload 2 files concurrently
23});
24
25// Set up the processor
26uploadQueue.setProcessor(async (data) => {
27  console.log('Uploading file:', data.fileName);
28  
29  const formData = new FormData();
30  formData.append('fileName', data.fileName);
31  formData.append('fileSize', data.fileSize.toString());
32  
33  if (data.metadata) {
34    formData.append('metadata', JSON.stringify(data.metadata));
35  }
36
37  const response = await fetch(data.uploadUrl, {
38    method: 'POST',
39    body: formData,
40  });
41
42  if (!response.ok) {
43    throw new Error(`Upload failed: ${response.status}`);
44  }
45
46  console.log('File uploaded successfully:', data.fileName);
47});
48
49// Add event handlers
50uploadQueue.on('taskCompleted', (task) => {
51  console.log('✅ File uploaded:', task.data.fileName);
52});
53
54uploadQueue.on('taskFailed', (task, error) => {
55  console.error('❌ Upload failed:', task.data.fileName, error.message);
56});
57
58// Usage
59export function addFileUpload(fileData: Omit<FileUploadData, 'id'>) {
60  const id = `upload-${Date.now()}-${Math.random().toString(36).substr(2, 9)}`;
61  return uploadQueue.add({ ...fileData, id });
62}

1import React, { useState } from 'react';
2import { useReliableQueue } from '@aplanka/reliable-queue';
3
4interface ProcessingTask {
5  id: string;
6  type: 'email' | 'report' | 'export' | 'analysis';
7  data: string;
8  priority?: number;
9}
10
11export function DataProcessingExample() {
12  const [taskType, setTaskType] = useState<ProcessingTask['type']>('email');
13  const [taskData, setTaskData] = useState('');
14
15  const {
16    tasks,
17    stats,
18    addTask,
19    retryTask,
20    removeTask,
21    clearFailed,
22    retryAllTasks,
23  } = useReliableQueue<ProcessingTask>({
24    queueName: 'data-processing',
25    maxRetries: 3,
26    retryDelay: 1000,
27    exponentialBackoff: true,
28    persistent: true,
29    concurrency: 2, // Process 2 tasks concurrently
30    processor: async (data) => {
31      console.log('Processing task:', data.type, data.id);
32      
33      // Simulate different processing times
34      const processingTime = {
35        email: 500,
36        report: 2000,
37        export: 1500,
38        analysis: 3000,
39      }[data.type];
40
41      // Simulate API call that might fail
42      if (Math.random() < 0.2) {
43        throw new Error(`${data.type} processing failed`);
44      }
45
46      await new Promise(resolve => setTimeout(resolve, processingTime));
47      console.log(`${data.type} task completed:`, data.id);
48    },
49  });
50
51  const handleAddTask = () => {
52    if (!taskData.trim()) return;
53
54    const taskId = `${taskType}-${Date.now()}-${Math.random().toString(36).substr(2, 9)}`;
55    
56    addTask({
57      id: taskId,
58      type: taskType,
59      data: taskData.trim(),
60      priority: taskType === 'email' ? 10 : taskType === 'report' ? 5 : 1,
61    });
62
63    setTaskData('');
64  };
65
66  return (
67    <div style={{ padding: '20px' }}>
68      <h2>Data Processing Queue</h2>
69      
70      {/* Queue Statistics */}
71      <div style={{ background: '#f5f5f5', padding: '10px', marginBottom: '20px' }}>
72        <h3>Queue Statistics</h3>
73        <div>
74          <span>Total: {stats.total}</span>
75          <span>Pending: {stats.pending}</span>
76          <span>Processing: {stats.processing}</span>
77          <span>Completed: {stats.completed}</span>
78          <span>Failed: {stats.failed}</span>
79        </div>
80      </div>
81
82      {/* Task Input */}
83      <div style={{ marginBottom: '20px' }}>
84        <select
85          value={taskType}
86          onChange={(e) => setTaskType(e.target.value as ProcessingTask['type'])}
87        >
88          <option value="email">Email Processing</option>
89          <option value="report">Report Generation</option>
90          <option value="export">Data Export</option>
91          <option value="analysis">Data Analysis</option>
92        </select>
93        
94        <input
95          type="text"
96          value={taskData}
97          onChange={(e) => setTaskData(e.target.value)}
98          placeholder="Enter task data..."
99          onKeyPress={(e) => e.key === 'Enter' && handleAddTask()}
100        />
101        
102        <button onClick={handleAddTask}>
103          Add {taskType} Task
104        </button>
105      </div>
106
107      {/* Task List */}
108      <div>
109        <h3>Processing Queue</h3>
110        {tasks.length === 0 ? (
111          <p>No tasks in queue</p>
112        ) : (
113          tasks.map((task) => (
114            <div key={task.id} style={{ border: '1px solid #ddd', padding: '10px', margin: '5px' }}>
115              <div>
116                <strong>{task.data.data}</strong>
117                <span> - {task.status}</span>
118                {task.retryCount > 0 && <span> (Retries: {task.retryCount})</span>}
119              </div>
120              
121              {task.status === 'failed' && (
122                <button onClick={() => retryTask(task.id)}>Retry</button>
123              )}
124              
125              {task.status !== 'processing' && (
126                <button onClick={() => removeTask(task.id)}>Remove</button>
127              )}
128            </div>
129          ))
130        )}
131      </div>
132    </div>
133  );
134}

1import { ReliableQueue, QueueManager } from '@aplanka/reliable-queue';
2
3interface EmailTask {
4  to: string;
5  subject: string;
6  body: string;
7  priority: 'low' | 'normal' | 'high' | 'urgent';
8  templateId?: string;
9  attachments?: string[];
10}
11
12// Create specialized email queue
13const emailQueue = new ReliableQueue<EmailTask>({
14  maxRetries: 5, // Important emails should retry more
15  retryDelay: 2000,
16  exponentialBackoff: true,
17  maxRetryDelay: 60000, // Max 1 minute delay
18  concurrency: 3, // Send 3 emails concurrently
19  persistent: true,
20});
21
22emailQueue.setProcessor(async (emailData) => {
23  console.log(`Sending email to ${emailData.to}`);
24  
25  // Simulate email API call
26  const response = await fetch('/api/send-email', {
27    method: 'POST',
28    headers: { 'Content-Type': 'application/json' },
29    body: JSON.stringify(emailData),
30  });
31  
32  if (!response.ok) {
33    const error = await response.text();
34    throw new Error(`Email send failed: ${error}`);
35  }
36  
37  return response.json();
38});
39
40// Add priority-based processing
41emailQueue.on('taskCompleted', (task) => {
42  console.log(`✅ Email sent to ${task.data.to}`);
43  // Log to analytics, update user notification status, etc.
44});
45
46// Usage examples
47export function sendWelcomeEmail(userEmail: string, userName: string) {
48  return emailQueue.add({
49    to: userEmail,
50    subject: 'Welcome to our platform!',
51    body: `Hello ${userName}, welcome to our platform!`,
52    priority: 'high',
53    templateId: 'welcome-email',
54  }, { priority: 10 }); // High priority
55}
56
57export function sendPasswordReset(userEmail: string, resetToken: string) {
58  return emailQueue.add({
59    to: userEmail,
60    subject: 'Password Reset Request',
61    body: `Click here to reset: /reset?token=${resetToken}`,
62    priority: 'urgent',
63    templateId: 'password-reset',
64  }, { priority: 100 }); // Highest priority
65}
66
67export function sendNewsletter(subscribers: string[], content: string) {
68  subscribers.forEach(email => {
69    emailQueue.add({
70      to: email,
71      subject: 'Weekly Newsletter',
72      body: content,
73      priority: 'low',
74      templateId: 'newsletter',
75    }, { priority: 1 }); // Low priority
76  });
77}

1import { ReliableQueue } from '@aplanka/reliable-queue';
2
3interface BatchJob {
4  id: string;
5  type: 'image-resize' | 'data-export' | 'report-generation';
6  batchSize: number;
7  items: any[];
8  progress?: {
9    processed: number;
10    total: number;
11    currentItem?: string;
12  };
13}
14
15const batchQueue = new ReliableQueue<BatchJob>({
16  maxRetries: 2,
17  retryDelay: 5000,
18  exponentialBackoff: true,
19  concurrency: 1, // Process batches one at a time
20  persistent: true,
21});
22
23batchQueue.setProcessor(async (job) => {
24  console.log(`Starting batch job: ${job.type}`);
25  
26  const results = [];
27  
28  for (let i = 0; i < job.items.length; i++) {
29    const item = job.items[i];
30    
31    try {
32      // Update progress (this would typically be saved to database)
33      const progress = {
34        processed: i,
35        total: job.items.length,
36        currentItem: item.id || item.name || `Item ${i + 1}`,
37      };
38      
39      console.log(`Processing ${progress.currentItem} (${progress.processed + 1}/${progress.total})`);
40      
41      // Process individual item based on job type
42      let result;
43      switch (job.type) {
44        case 'image-resize':
45          result = await resizeImage(item);
46          break;
47        case 'data-export':
48          result = await exportData(item);
49          break;
50        case 'report-generation':
51          result = await generateReport(item);
52          break;
53        default:
54          throw new Error(`Unknown job type: ${job.type}`);
55      }
56      
57      results.push({ item: item.id, result, status: 'success' });
58      
59    } catch (error) {
60      console.error(`Failed to process item ${item.id}:`, error);
61      results.push({ item: item.id, error: error.message, status: 'failed' });
62    }
63  }
64  
65  console.log(`Batch job completed: ${job.type}`);
66  return {
67    jobId: job.id,
68    results,
69    summary: {
70      total: results.length,
71      successful: results.filter(r => r.status === 'success').length,
72      failed: results.filter(r => r.status === 'failed').length,
73    }
74  };
75});
76
77// Helper functions (implement based on your needs)
78async function resizeImage(image: any) {
79  // Simulate image processing
80  await new Promise(resolve => setTimeout(resolve, 1000));
81  return { url: `/resized/${image.id}.jpg`, width: 800, height: 600 };
82}
83
84async function exportData(data: any) {
85  // Simulate data export
86  await new Promise(resolve => setTimeout(resolve, 500));
87  return { file: `/exports/${data.id}.csv`, size: 1024 };
88}
89
90async function generateReport(reportConfig: any) {
91  // Simulate report generation
92  await new Promise(resolve => setTimeout(resolve, 2000));
93  return { file: `/reports/${reportConfig.id}.pdf`, pages: 10 };
94}
95
96// Usage
97export function processBatchImages(images: any[]) {
98  return batchQueue.add({
99    id: `batch-${Date.now()}`,
100    type: 'image-resize',
101    batchSize: images.length,
102    items: images,
103  });
104}