External Tasks

import { Client } from '@processcube/client';
 
const client = new Client('http://localhost:8000');
 
async function setupExternalTaskWorker() {
  // Topic, auf das der Worker hört
  const externalTaskTopic = 'SendEmail';
 
  console.log(`Erstelle External Task Worker für Topic: ${externalTaskTopic}`);
 
  // External Task Worker registrieren
  const worker = await client.subscribeToExternalTask(
    externalTaskTopic,
    handleExternalTask
  );
 
  // Worker starten
  worker.start();
 
  console.log('External Task Worker gestartet');
}
 
// Handler-Funktion für External Tasks
async function handleExternalTask(externalTask): Promise<object> {
  console.log('Verarbeite External Task:', externalTask.id);
 
  // Task-Parameter auslesen
  const recipient = externalTask.payload.recipient;
  const subject = externalTask.payload.subject;
  const body = externalTask.payload.body;
 
  // E-Mail versenden (Pseudo-Code)
  await sendEmail(recipient, subject, body);
 
  // Ergebnis zurückgeben
  return {
    success: true,
    sentAt: new Date().toISOString(),
  };
}
 
setupExternalTaskWorker().catch(console.error);

const workerConfig = {
  lockDuration: 60000,        // Task-Lock-Dauer in ms (60 Sekunden)
  maxTasks: 25,               // Max. Anzahl Tasks pro Poll
  longpollingTimeout: 2000,   // Timeout für Long-Polling in ms
};
 
const worker = await client.subscribeToExternalTask(
  'ProcessOrders',
  handleOrderTask,
  workerConfig
);

// Worker erstellen
const worker = await client.subscribeToExternalTask(
  'ProcessPayment',
  handlePayment
);
 
// Worker starten
worker.start();
console.log('Worker läuft...');
 
// Worker stoppen (z.B. bei Shutdown)
process.on('SIGINT', () => {
  console.log('Stoppe Worker...');
  worker.stop();
  process.exit(0);
});

async function handleExternalTask(externalTask): Promise<object> {
  try {
    // Task-Logik
    await processTask(externalTask.payload);
 
    return { success: true };
  } catch (error) {
    // Task als fehlgeschlagen markieren
    throw new Error(`Task fehlgeschlagen: ${error.message}`);
  }
}

async function handleExternalTask(externalTask): Promise<object> {
  const amount = externalTask.payload.amount;
 
  if (amount > 10000) {
    // BPMN-Fehler werfen (für Error Boundary Events)
    throw {
      code: 'AMOUNT_TOO_HIGH',
      message: 'Betrag überschreitet Limit',
    };
  }
 
  return { approved: true };
}

// Email Worker
const emailWorker = await client.subscribeToExternalTask(
  'SendEmail',
  handleEmail
);
emailWorker.start();
 
// Payment Worker
const paymentWorker = await client.subscribeToExternalTask(
  'ProcessPayment',
  handlePayment
);
paymentWorker.start();
 
// Order Worker
const orderWorker = await client.subscribeToExternalTask(
  'ProcessOrder',
  handleOrder,
  { maxTasks: 50 }
);
orderWorker.start();
 
console.log('Alle Worker gestartet');

async function handleExternalTask(externalTask): Promise<object> {
  const taskId = externalTask.id;
 
  // Prüfen, ob Task bereits verarbeitet wurde
  if (await isAlreadyProcessed(taskId)) {
    console.log(`Task ${taskId} wurde bereits verarbeitet`);
    return { skipped: true };
  }
 
  // Task verarbeiten
  const result = await processTask(externalTask.payload);
 
  // Als verarbeitet markieren
  await markAsProcessed(taskId);
 
  return result;
}

async function handleExternalTask(externalTask): Promise<object> {
  const startTime = Date.now();
 
  console.log(`[${externalTask.id}] Start - Topic: ${externalTask.topic}`);
 
  try {
    const result = await processTask(externalTask.payload);
 
    const duration = Date.now() - startTime;
    console.log(`[${externalTask.id}] Erfolg - Dauer: ${duration}ms`);
 
    return result;
  } catch (error) {
    const duration = Date.now() - startTime;
    console.error(`[${externalTask.id}] Fehler nach ${duration}ms:`, error);
 
    throw error;
  }
}

let isShuttingDown = false;
 
async function handleExternalTask(externalTask): Promise<object> {
  if (isShuttingDown) {
    throw new Error('Server wird heruntergefahren');
  }
 
  // Normal processing...
  return { success: true };
}
 
process.on('SIGTERM', async () => {
  console.log('SIGTERM empfangen, fahre herunter...');
  isShuttingDown = true;
 
  // Warte auf laufende Tasks
  await new Promise((resolve) => setTimeout(resolve, 5000));
 
  // Stoppe Worker
  worker.stop();
 
  process.exit(0);
});

import * as nodemailer from 'nodemailer';
 
const transporter = nodemailer.createTransport({
  host: process.env.SMTP_HOST,
  port: 587,
  auth: {
    user: process.env.SMTP_USER,
    pass: process.env.SMTP_PASS,
  },
});
 
async function handleEmailTask(externalTask): Promise<object> {
  const { recipient, subject, body } = externalTask.payload;
 
  const info = await transporter.sendMail({
    from: '"ProcessCube" <noreply@processcube.io>',
    to: recipient,
    subject: subject,
    html: body,
  });
 
  return {
    messageId: info.messageId,
    success: true,
  };
}
 
const worker = await client.subscribeToExternalTask('SendEmail', handleEmailTask);
worker.start();

import axios from 'axios';
 
async function handleHttpRequestTask(externalTask): Promise<object> {
  const { url, method, data } = externalTask.payload;
 
  const response = await axios({
    method,
    url,
    data,
    timeout: 30000,
  });
 
  return {
    status: response.status,
    data: response.data,
  };
}
 
const worker = await client.subscribeToExternalTask('HttpRequest', handleHttpRequestTask);
worker.start();

import { Pool } from 'pg';
 
const pool = new Pool({
  connectionString: process.env.DATABASE_URL,
});
 
async function handleDatabaseTask(externalTask): Promise<object> {
  const { query, params } = externalTask.payload;
 
  const client = await pool.connect();
 
  try {
    const result = await client.query(query, params);
    return {
      rows: result.rows,
      rowCount: result.rowCount,
    };
  } finally {
    client.release();
  }
}
 
const worker = await client.subscribeToExternalTask('DatabaseQuery', handleDatabaseTask);
worker.start();