simplify workers

Author: muniter

README.md

@@ -7,7 +7,6 @@ A TypeScript queue system inspired by Yii2-Queue architecture, providing a clean
 - **Driver-based architecture**: Swap between DB, SQS, and File drivers seamlessly
 - **Event-based jobs**: Register handlers for job types without complex classes
 - **Type-safe API**: Full TypeScript support with driver-specific option validation
-- **Worker isolation**: Run jobs in separate processes for better stability
 - **Multiple backends**: Database, Amazon SQS, and File storage drivers
 - **Event system**: Hook into queue lifecycle events
 
@@ -219,9 +218,9 @@ await worker.start(true, 3); // repeat=true, timeout=3 seconds
 // Process once then exit
 await worker.start(false);
 
-// With isolation (jobs run in child processes)
-const isolatedWorker = new Worker(queue, { isolate: true });
-await isolatedWorker.start();
+// With custom timeout
+const worker = new Worker(queue, { timeout: 5 });
+await worker.start();
 ```
 
 ## Event Handling
@@ -298,9 +297,6 @@ npm run queue:worker -- --driver sqs --queue-url https://sqs.us-east-1.amazonaws
 # File driver
 npm run queue:worker -- --driver file --path ./queue-data
 
-# Isolated mode
-npm run queue:worker -- --isolate
-
 # Run once and exit
 npm run queue:worker -- --no-repeat
 

src/cli/worker.ts

@@ -6,7 +6,6 @@ import { SqsQueue } from '../drivers/sqs.ts';
 
 interface WorkerConfig {
   driver: 'db' | 'sqs';
-  isolate?: boolean;
   repeat?: boolean;
   timeout?: number;
   dbAdapter?: any;
@@ -18,7 +17,6 @@ function parseArgs(): WorkerConfig {
   const args = process.argv.slice(2);
   const config: WorkerConfig = {
     driver: 'db',
-    isolate: false,
     repeat: true,
     timeout: 3
   };
@@ -30,9 +28,6 @@ function parseArgs(): WorkerConfig {
       case '--driver':
         config.driver = args[++i] as 'db' | 'sqs';
         break;
-      case '--isolate':
-        config.isolate = true;
-        break;
       case '--no-repeat':
         config.repeat = false;
         break;
@@ -60,14 +55,13 @@ Usage: node worker.js [options]
 
 Options:
   --driver <type>     Queue driver: 'db' or 'sqs' (default: db)
-  --isolate          Run jobs in isolated child processes
   --no-repeat        Run once and exit (default: run continuously)
   --timeout <sec>    Polling timeout in seconds (default: 3)
   --queue-url <url>  SQS queue URL (required for SQS driver)
   --help             Show this help message
 
 Examples:
-  node worker.js --driver db --isolate
+  node worker.js --driver db
   node worker.js --driver sqs --queue-url https://sqs.us-east-1.amazonaws.com/123/test
   node worker.js --no-repeat --timeout 10
   `);
@@ -102,12 +96,10 @@ async function main(): Promise<void> {
   }
 
   const worker = new Worker(queue, {
-    isolate: config.isolate,
     timeout: config.timeout
   });
 
   console.log(`Starting worker with ${config.driver} driver...`);
-  console.log(`Isolate: ${config.isolate}`);
   console.log(`Repeat: ${config.repeat}`);
   console.log(`Timeout: ${config.timeout}s`);
 

src/worker/worker-child.ts

@@ -1,53 +1,18 @@
-import { spawn, ChildProcess } from 'child_process';
-import path from 'path';
 import { Queue } from '../core/queue.ts';
-import type { QueueMessage } from '../interfaces/job.ts';
 
 export interface WorkerOptions {
-  isolate?: boolean;
   timeout?: number;
-  childScriptPath?: string;
 }
 
 export class Worker {
   constructor(
     private queue: Queue,
     private options: WorkerOptions = {}
-  ) {
-    if (this.options.isolate) {
-      this.setupIsolatedHandler();
-    }
-  }
+  ) {}
 
   async start(repeat: boolean = true, timeout: number = 3): Promise<void> {
-    await this.queue.run(repeat, timeout);
-  }
-
-  private setupIsolatedHandler(): void {
-    const originalHandleMessage = this.queue['handleMessage'].bind(this.queue);
-    
-    this.queue['handleMessage'] = async (message: QueueMessage): Promise<boolean> => {
-      return new Promise<boolean>((resolve) => {
-        const childScriptPath = this.options.childScriptPath || 
-          path.resolve(__dirname, 'worker-child.js');
-        
-        const child: ChildProcess = spawn(process.execPath, [childScriptPath], {
-          stdio: ['pipe', 'inherit', 'inherit'],
-          timeout: (message.meta.ttr || 300) * 1000
-        });
-
-        child.stdin!.end(message.payload);
-        
-        child.on('close', (code) => {
-          resolve(code === 0);
-        });
-
-        child.on('error', (error) => {
-          console.error('Child process error:', error);
-          resolve(false);
-        });
-      });
-    };
+    const actualTimeout = this.options.timeout ?? timeout;
+    await this.queue.run(repeat, actualTimeout);
   }
 }
 

src/worker/worker.ts

@@ -63,22 +63,6 @@ describe('Worker', () => {
       expect(processedJobs.sort()).toEqual(['job0', 'job1', 'job2', 'job3', 'job4']);
     });
 
-    it('should handle worker with isolation', async () => {
-      const processedJobs: string[] = [];
-      
-      queue.onJob('simple-job', async (payload) => {
-        processedJobs.push(payload.data);
-      });
-
-      await queue.addJob('simple-job', { payload: { data: 'isolated test' } });
-
-      const worker = new Worker(queue, { isolate: true });
-      await worker.start(false, 1);
-
-      // Note: In isolation mode, the job may not execute the registered handler
-      // This test mainly verifies that the worker doesn't crash
-      expect(true).toBe(true); // Worker didn't crash
-    });
 
     it('should handle worker start and stop', async () => {
       const worker = new Worker(queue);
@@ -88,8 +72,8 @@ describe('Worker', () => {
     });
 
     it('should support different worker options', async () => {
-      const worker1 = new Worker(queue, { isolate: false });
-      const worker2 = new Worker(queue, { isolate: true, timeout: 5 });
+      const worker1 = new Worker(queue);
+      const worker2 = new Worker(queue, { timeout: 5 });
 
       expect(worker1).toBeInstanceOf(Worker);
       expect(worker2).toBeInstanceOf(Worker);