koco/avro-regy

Symfony Messenger Avro Schema Registry Bundle

v1.2.0 2022-10-27 14:44 UTC

Requires

• php: ^7.4|8.*
• flix-tech/avro-serde-php: ^2.0
• symfony/framework-bundle: ^5.4||^6.0
• symfony/messenger: ^5.4||^6.0

Requires (Dev)

• friendsofphp/php-cs-fixer: ^3.0
• matthiasnoback/symfony-config-test: ^4.2
• phpstan/phpstan: ^0.12.52
• symfony/phpunit-bridge: ^5.4||^6.0

Suggests

• koco/messenger-kafka: Out of the box Kafka transport

MIT

• queue
• symfony
• bundle
• transport
• Messenger
• kafka

README

This bundle provides a simple way to integrate Symfony Messenger with the Confluent Schema Registry.

Installation

Applications that use Symfony Flex

Open a command console, enter your project directory and execute:

$ composer require koco/avro-regy

Applications that don't use Symfony Flex

After adding the composer requirement, enable the bundle by adding it to the list of registered bundles in the config/bundles.php file of your project:

return [
    // ...
    Koco\AvroRegy\AvroRegyBundle::class => ['all' => true],
];

Configuration

Create config/avro_regy.yaml and configure similar to the following example:

avro_regy:
  base_uri: '%env(SCHEMA_REGISTRY_URL)%'
  file_naming_strategy: subject
  options:
    register_missing_schemas: true
    register_missing_subjects: true
  serializers:
    catalogue:
      schema_dir: '%kernel.project_dir%/src/Catalogue/Domain/Model/Event/Avro/'
    orders:
      schema_dir: '%kernel.project_dir%/src/Orders/Domain/Model/Event/Avro/'

Each serializer that you configure, will be available as it's own service. The service id is avro_regy.serializer.<name>. In config/messenger.yaml it can be called like so:

framework:
    messenger:
        transports:
            catalogue:
                dsn: '%env(MESSENGER_TRANSPORT_CATALOGUE_KAFKA_DSN)%'
                serializer: avro_regy.serializer.catalogue
                options:
                    ...

Then, write your serializer like so:

namespace App\Catalogue\Infrastructure\Messenger\Serializer;

use App\Catalogue\Domain\Model\Event\ProductCreated;
use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;

class ProductCreatedSerializer implements SerializerInterface
{
    public function decode(array $encodedEnvelope): Envelope
    {
        $record = $encodedEnvelope['body'];

        return new Envelope(new ProductCreated(
            $record['id'],
            $record['name'],
            $record['description'],
        ));
    }

    public function encode(Envelope $envelope): array
    {
        /** @var ProductCreated $event */
        $event = $envelope->getMessage();
        
        return [
            'key' => $event->getId(),
            'headers' => [],
            'body' => [
                'id' => $event->getId(),
                'name' => $event->getName(),
                'description' => $event->getDescription(),
            ],
        ];
    }
}

And tag it:

App\Catalogue\Infrastructure\Messenger\Serializer\ProductCreatedSerializer:
  tags:
    - {
      name: 'avro_regy.serializer.catalogue',
      qualified_name: 'com.Example.Catalogue.ProductCreated',
      class_name: 'App\Catalogue\Domain\Model\Event\ProductCreated',
      key_subject: 'catalogue.product_created-key',
      value_subject: 'catalogue.product_created-value'
    }