RabbitMQ avec Symfony et Celery : Guide Complet

<?php
namespace App\Message;

class ProcessImageMessage
{
    private string $imagePath;
    private string $userId;
    private array $filters;

    public function __construct(string $imagePath, string $userId, array $filters = [])
    {
        $this->imagePath = $imagePath;
        $this->userId = $userId;
        $this->filters = $filters;
    }

    public function getImagePath(): string
    {
        return $this->imagePath;
    }

    public function getUserId(): string
    {
        return $this->userId;
    }

    public function getFilters(): array
    {
        return $this->filters;
    }
}

<?php
namespace App\Service;

use App\Message\ProcessImageMessage;
use Symfony\Component\Messenger\MessageBusInterface;

class MessagePublisher
{
    private MessageBusInterface $messageBus;

    public function __construct(MessageBusInterface $messageBus)
    {
        $this->messageBus = $messageBus;
    }

    public function publishImageProcessing(string $imagePath, string $userId, array $filters): void
    {
        $message = new ProcessImageMessage($imagePath, $userId, $filters);

        // Publication du message dans RabbitMQ
        $this->messageBus->dispatch($message);
    }
}

<?php
namespace App\Controller;

use App\Service\MessagePublisher;
use Symfony\Bundle\FrameworkBundle\Controller\AbstractController;
use Symfony\Component\HttpFoundation\Request;
use Symfony\Component\HttpFoundation\Response;
use Symfony\Component\Routing\Annotation\Route;

class ImageController extends AbstractController
{
    #[Route('/image/process', name: 'image_process', methods: ['POST'])]
    public function processImage(Request $request, MessagePublisher $publisher): Response
    {
        $imagePath = $request->request->get('image_path');
        $filters = $request->request->get('filters', []);
        $userId = $this->getUser()->getId();

        // Publication du message pour traitement asynchrone
        $publisher->publishImageProcessing($imagePath, $userId, $filters);

        return $this->json([
            'status' => 'success',
            'message' => 'Image processing started',
            'job_id' => uniqid() // Dans un vrai projet, utiliser un UUID
        ]);
    }
}

from celery import Celery
import json

# Configuration de Celery
app = Celery('image_processor')
app.conf.update(
    broker_url='amqp://admin:password@localhost:5672//',
    result_backend='rpc://',
    task_serializer='json',
    accept_content=['json'],
    result_serializer='json',
    timezone='Europe/Paris',
    enable_utc=True,
)

# Configuration des queues
app.conf.task_routes = {
    'tasks.process_image': {'queue': 'image_processing'},
    'tasks.analyze_data': {'queue': 'data_analysis'},
}

if __name__ == '__main__':
    app.start()

from celery_app import app
from PIL import Image, ImageFilter
import os
import json

@app.task(name='tasks.process_image')
def process_image(message_data):
    """
    Traite une image selon les filtres spécifiés
    """
    try:
        # Décodage du message Symfony
        data = json.loads(message_data) if isinstance(message_data, str) else message_data

        image_path = data.get('imagePath')
        user_id = data.get('userId')
        filters = data.get('filters', [])

        # Ouverture de l'image
        with Image.open(image_path) as img:
            processed_img = img.copy()

            # Application des filtres
            for filter_name in filters:
                if filter_name == 'blur':
                    processed_img = processed_img.filter(ImageFilter.BLUR)
                elif filter_name == 'sharpen':
                    processed_img = processed_img.filter(ImageFilter.SHARPEN)
                elif filter_name == 'grayscale':
                    processed_img = processed_img.convert('L')

            # Sauvegarde de l'image traitée
            output_path = f"processed_{os.path.basename(image_path)}"
            processed_img.save(output_path)

            # Notification du résultat (webhook, base de données, etc.)
            notify_processing_complete(user_id, output_path)

            return {
                'status': 'success',
                'output_path': output_path,
                'user_id': user_id
            }

    except Exception as e:
        return {
            'status': 'error',
            'error': str(e)
        }

def notify_processing_complete(user_id, output_path):
    """
    Notifie Symfony que le traitement est terminé
    """
    # Ici vous pourriez :
    # - Insérer en base de données
    # - Appeler une API Symfony
    # - Publier un message de retour
    pass

<?php
namespace App\Tests\Service;

use App\Service\MessagePublisher;
use PHPUnit\Framework\TestCase;
use Symfony\Component\Messenger\MessageBusInterface;

class MessagePublisherTest extends TestCase
{
    public function testPublishImageProcessing(): void
    {
        $messageBus = $this->createMock(MessageBusInterface::class);
        $messageBus->expects($this->once())
                  ->method('dispatch');

        $publisher = new MessagePublisher($messageBus);
        $publisher->publishImageProcessing('/path/to/image.jpg', 'user123', ['blur']);

        $this->assertTrue(true); // Le message a été publié
    }
}

import unittest
from unittest.mock import patch, MagicMock
from tasks import process_image

class TestImageProcessing(unittest.TestCase):

    @patch('tasks.Image.open')
    def test_process_image_success(self, mock_image_open):
        # Mock de PIL
        mock_img = MagicMock()
        mock_image_open.return_value.__enter__.return_value = mock_img

        message_data = {
            'imagePath': '/test/image.jpg',
            'userId': 'user123',
            'filters': ['blur']
        }

        result = process_image(message_data)

        self.assertEqual(result['status'], 'success')
        self.assertEqual(result['user_id'], 'user123')

if __name__ == '__main__':
    unittest.main()