add avro client register

Author: Iachimovschi Danu

.env

@@ -41,3 +41,7 @@ RABBIT_MESSENGER_TRANSPORT_DSN=amqp://guest:guest@rabbitmq:5672/%2f
 # DATABASE_URL="mysql://app:!ChangeMe!@127.0.0.1:3306/app?serverVersion=8.0.32&charset=utf8mb4"
 DATABASE_URL="postgresql://user:password@db:5432/app_db?serverVersion=16&charset=utf8"
 ###< doctrine/doctrine-bundle ###
+
+# SCHEMA SCHEMA REGISTRY URL FOR AVRO SCHEMA REGISTRY
+# SCHEMA_REGISTRY_URL=http://localhost:8081
+SCHEMA_REGISTRY_URL=schema-registry:8081

.env.dev

@@ -1,4 +1,26 @@
-
 ###> symfony/framework-bundle ###
-APP_SECRET=ad170ccb23f26ba4ec3cdd58cbbbbd9f
+APP_ENV=dev
+APP_DEBUG=true
+APP_SECRET=dev-secret-key
 ###< symfony/framework-bundle ###
+
+###> symfony/messenger ###
+# In development, you might want to use a local Kafka or disable messaging
+MESSENGER_TRANSPORT_DSN=kafka://localhost:9092
+KAFKA_BROKERS=localhost:9092
+
+# If you are using RabbitMQ locally for messaging:
+RABBIT_MESSENGER_TRANSPORT_DSN=amqp://guest:guest@localhost:5672/%2f
+###< symfony/messenger ###
+
+###> doctrine/doctrine-bundle ###
+# In development, it's common to use SQLite or PostgreSQL with a dev-specific database
+DATABASE_URL="postgresql://user:password@localhost:5432/app_db_dev?serverVersion=16&charset=utf8"
+# Alternatively, for SQLite:
+# DATABASE_URL="sqlite:///%kernel.project_dir%/var/data_dev.db"
+###< doctrine/doctrine-bundle ###
+
+###> Avro Schema Registry ###
+# Use a local schema registry for development
+SCHEMA_REGISTRY_URL=schema-registry:8081
+###< Avro Schema Registry ###

config/packages/schema_registry.yaml

@@ -0,0 +1,4 @@
+parameters:
+    schema_registry.url: '%env(SCHEMA_REGISTRY_URL)%'
+    schema_registry.register_missing_schemas: 0
+    schema_registry.register_missing_subjects: 1

config/services.yaml

@@ -1,24 +1,25 @@
-# This file is the entry point to configure your own services.
-# Files in the packages/ subdirectory configure your dependencies.
-
-# Put parameters here that don't need to change on each machine where the app is deployed
-# https://symfony.com/doc/current/best_practices.html#use-parameters-for-application-configuration
 parameters:
 
+
+imports:
+    - { resource: 'packages/schema_registry.yaml' }
+
 services:
-    # default configuration for services in *this* file
     _defaults:
-        autowire: true      # Automatically injects dependencies in your services.
-        autoconfigure: true # Automatically registers your services as commands, event subscribers, etc.
+        autowire: true
+        autoconfigure: true
 
-    # makes classes in src/ available to be used as services
-    # this creates a service per class whose id is the fully-qualified class name
     App\:
         resource: '../src/'
         exclude:
             - '../src/DependencyInjection/'
             - '../src/Entity/'
             - '../src/Kernel.php'
 
-    # add more service definitions when explicit configuration is needed
-    # please note that last definitions always *replace* previous ones
+    App\Infrastructure\Avro\SchemaRegistryClientInterface: '@App\Infrastructure\Avro\ConfluenceSchemaRegistryClient'
+    App\Infrastructure\Avro\ConfluenceSchemaRegistryClient:
+        arguments:
+            $schemaRegistryService: '@App\Infrastructure\Avro\SchemaRegistryService'
+        public: true
+        shared: true
+

src/Infrastructure/Avro/ConfluenceSchemaRegistryClient.php

@@ -0,0 +1,72 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Infrastructure\Avro;
+
+use App\Infrastructure\Avro\Interfaces\SchemaRegistryClientInterface;
+use FlixTech\AvroSerializer\Objects\RecordSerializer;
+use FlixTech\SchemaRegistryApi\Registry;
+use FlixTech\SchemaRegistryApi\Registry\Cache\AvroObjectCacheAdapter;
+use FlixTech\SchemaRegistryApi\Registry\CachedRegistry;
+use FlixTech\SchemaRegistryApi\Registry\PromisingRegistry;
+use GuzzleHttp\Client;
+use Jobcloud\Kafka\Message\Decoder\AvroDecoder;
+use Jobcloud\Kafka\Message\Decoder\AvroDecoderInterface;
+use Jobcloud\Kafka\Message\Encoder\AvroEncoder;
+use Jobcloud\Kafka\Message\Encoder\AvroEncoderInterface;
+use Jobcloud\Kafka\Message\Registry\AvroSchemaRegistry;
+use Jobcloud\Kafka\Message\Registry\AvroSchemaRegistryInterface;
+
+class ConfluenceSchemaRegistryClient implements SchemaRegistryClientInterface
+{
+    private readonly Registry $cachedRegistry;
+
+    private readonly AvroSchemaRegistryInterface $schemaRegistryClient;
+
+    private readonly RecordSerializer $recordSerializer;
+
+    public function __construct(SchemaRegistryService $schemaRegistryService)
+    {
+        $this->cachedRegistry = new CachedRegistry(
+            new PromisingRegistry(
+                new Client(['base_uri' => $schemaRegistryService->getUrl()])
+            ),
+            new AvroObjectCacheAdapter()
+        );
+        $this->schemaRegistryClient = new AvroSchemaRegistry($this->cachedRegistry);
+
+        $this->recordSerializer = new RecordSerializer(
+            $this->cachedRegistry,
+            [
+                RecordSerializer::OPTION_REGISTER_MISSING_SCHEMAS => $schemaRegistryService->getRegisterMissingSchemas(),
+                RecordSerializer::OPTION_REGISTER_MISSING_SUBJECTS => $schemaRegistryService->getRegisterMissingSubjects(),
+            ]
+        );
+    }
+
+    public function getRegistry(): Registry
+    {
+        return $this->cachedRegistry;
+    }
+
+    public function getSchemaRegistryClient(): AvroSchemaRegistryInterface
+    {
+        return $this->schemaRegistryClient;
+    }
+
+    public function getRecordSerializer(): RecordSerializer
+    {
+        return $this->recordSerializer;
+    }
+
+    public function getEncoder(): AvroEncoderInterface
+    {
+        return new AvroEncoder($this->schemaRegistryClient, $this->recordSerializer);
+    }
+
+    public function getDecoder(): AvroDecoderInterface
+    {
+        return new AvroDecoder($this->schemaRegistryClient, $this->recordSerializer);
+    }
+}

src/Infrastructure/Avro/Interfaces/SchemaRegistryClientInterface.php

@@ -0,0 +1,9 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Infrastructure\Avro\Interfaces;
+
+interface SchemaRegistryClientInterface extends SchemaSerializerInterface, SchemaRegistryInterface
+{
+}

src/Infrastructure/Avro/Interfaces/SchemaRegistryInterface.php

@@ -0,0 +1,15 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Infrastructure\Avro\Interfaces;
+
+use FlixTech\SchemaRegistryApi\Registry;
+use Jobcloud\Kafka\Message\Registry\AvroSchemaRegistryInterface;
+
+interface SchemaRegistryInterface
+{
+    public function getRegistry(): Registry;
+
+    public function getSchemaRegistryClient(): AvroSchemaRegistryInterface;
+}

src/Infrastructure/Avro/Interfaces/SchemaSerializerInterface.php

@@ -0,0 +1,18 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Infrastructure\Avro\Interfaces;
+
+use FlixTech\AvroSerializer\Objects\RecordSerializer;
+use Jobcloud\Kafka\Message\Decoder\AvroDecoderInterface;
+use Jobcloud\Kafka\Message\Encoder\AvroEncoderInterface;
+
+interface SchemaSerializerInterface
+{
+    public function getRecordSerializer(): RecordSerializer;
+
+    public function getEncoder(): AvroEncoderInterface;
+
+    public function getDecoder(): AvroDecoderInterface;
+}

src/Infrastructure/Avro/SchemaRegistryService.php

@@ -0,0 +1,38 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Infrastructure\Avro;
+
+use Symfony\Component\DependencyInjection\ParameterBag\ParameterBagInterface;
+
+readonly class SchemaRegistryService
+{
+    private string $url;
+
+    private bool $register_missing_schemas;
+
+    private bool $register_missing_subjects;
+
+    public function __construct(ParameterBagInterface $params)
+    {
+        $this->url = $params->get('schema_registry.url');
+        $this->register_missing_schemas = (bool) $params->get('schema_registry.register_missing_schemas');
+        $this->register_missing_subjects = (bool) $params->get('schema_registry.register_missing_subjects');
+    }
+
+    public function getUrl(): string
+    {
+        return $this->url;
+    }
+
+    public function getRegisterMissingSchemas(): bool
+    {
+        return $this->register_missing_schemas;
+    }
+
+    public function getRegisterMissingSubjects(): bool
+    {
+        return $this->register_missing_subjects;
+    }
+}

src/Infrastructure/Command/UserConsumerCommand.php

@@ -5,20 +5,15 @@
 namespace App\Infrastructure\Command;
 
 use App\Domain\user\Entity\User;
+use App\Infrastructure\Avro\Interfaces\SchemaRegistryClientInterface;
 use AvroSchema;
 use Doctrine\ORM\EntityManagerInterface;
-use FlixTech\AvroSerializer\Objects\RecordSerializer;
-use FlixTech\SchemaRegistryApi\Registry\Cache\AvroObjectCacheAdapter;
-use FlixTech\SchemaRegistryApi\Registry\CachedRegistry;
-use FlixTech\SchemaRegistryApi\Registry\PromisingRegistry;
-use GuzzleHttp\Client;
 use Jobcloud\Kafka\Consumer\KafkaConsumerBuilder;
 use Jobcloud\Kafka\Exception\KafkaConsumerConsumeException;
 use Jobcloud\Kafka\Exception\KafkaConsumerEndOfPartitionException;
 use Jobcloud\Kafka\Exception\KafkaConsumerTimeoutException;
 use Jobcloud\Kafka\Message\Decoder\AvroDecoder;
-use Jobcloud\Kafka\Message\Encoder\AvroEncoder;
-use Jobcloud\Kafka\Message\Registry\AvroSchemaRegistry;
+use Symfony\Bundle\FrameworkBundle\Controller\AbstractController;
 use Symfony\Component\Console\Attribute\AsCommand;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputArgument;
@@ -32,10 +27,11 @@
 )]
 class UserConsumerCommand extends Command
 {
-    private EntityManagerInterface $entityManager;
-
-    public function __construct(EntityManagerInterface $entityManager, string $name = null)
-    {
+    public function __construct(
+        private EntityManagerInterface $entityManager,
+        protected readonly SchemaRegistryClientInterface $schemaRegistryClient,
+        string $name = null
+    ) {
         $this->entityManager = $entityManager;
         parent::__construct($name);
     }
@@ -49,23 +45,6 @@ protected function execute(InputInterface $input, OutputInterface $output): int
     {
         $io = new SymfonyStyle($input, $output);
 
-        $schemaRegistryClient = new CachedRegistry(
-            new PromisingRegistry(
-                new Client(['base_uri' => 'schema-registry:8081'])
-            ),
-            new AvroObjectCacheAdapter()
-        );
-
-        $registry = new AvroSchemaRegistry($schemaRegistryClient);
-
-        $recordSerializer = new RecordSerializer(
-            $schemaRegistryClient,
-            [
-                RecordSerializer::OPTION_REGISTER_MISSING_SCHEMAS => false,
-                RecordSerializer::OPTION_REGISTER_MISSING_SUBJECTS => true,
-            ]
-        );
-
         $schema = <<<'JSON'
         {
             "type": "record",
@@ -80,15 +59,13 @@ protected function execute(InputInterface $input, OutputInterface $output): int
 
         $avroSchema = AvroSchema::parse($schema);
 
-        $decoder = new AvroDecoder($registry, $recordSerializer);
-
         $consumer = KafkaConsumerBuilder::create()
             ->withAdditionalConfig(
                 [
                     'enable.auto.commit' => false,
                 ]
             )
-            ->withDecoder($decoder)
+            ->withDecoder($this->schemaRegistryClient->getDecoder())
             ->withAdditionalBroker('kafka:9092')
             ->withConsumerGroup('testGroup')
             ->withAdditionalSubscription('users')
@@ -100,7 +77,7 @@ protected function execute(InputInterface $input, OutputInterface $output): int
             try {
                 $message = $consumer->consume();
 
-                $userData =  $recordSerializer->decodeMessage($message->getBody(), $avroSchema);
+                $userData = $this->schemaRegistryClient->getRecordSerializer()->decodeMessage($message->getBody(), $avroSchema);
 
                 $io->success('Message received: '. $userData['name']);