example of avro serialize desirialize message

Author: Iachimovschi Danu

src/Infrastructure/Command/UserConsumerCommand.php

@@ -5,7 +5,13 @@
 namespace App\Infrastructure\Command;
 
 use App\Domain\user\Entity\User;
+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;
@@ -40,6 +46,22 @@ 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()
+        );
+
+        $recordSerializer = new RecordSerializer(
+            $schemaRegistryClient,
+            [
+                RecordSerializer::OPTION_REGISTER_MISSING_SCHEMAS => false,
+                RecordSerializer::OPTION_REGISTER_MISSING_SUBJECTS => true,
+            ]
+        );
+
         $consumer = KafkaConsumerBuilder::create()
             ->withAdditionalConfig(
                 [
@@ -53,16 +75,32 @@ protected function execute(InputInterface $input, OutputInterface $output): int
 
         $consumer->subscribe();
 
+        $schema = <<<'JSON'
+        {
+            "type": "record",
+            "name": "User",
+            "fields": [
+                {"name": "name", "type": "string"},
+                {"name": "surname", "type": "string"},
+                {"name": "email", "type": "string"}
+            ]
+        }
+        JSON;
+
+        $avroSchema = AvroSchema::parse($schema);
+
         while (true) {
             try {
                 $message = $consumer->consume();
-                $userData = json_decode($message->getBody(), true);
-                $io->success('Message received: '. $userData['Name']);
+
+                $userData =  $recordSerializer->decodeMessage($message->getBody(), $avroSchema);
+
+                $io->success('Message received: '. $userData['name']);
 
                 $user = new User();
-                $user->setName($userData['Name']);
-                $user->setSurname($userData['Surname']);
-                $user->setEmail($userData['Email']);
+                $user->setName($userData['name']);
+                $user->setSurname($userData['surname']);
+                $user->setEmail($userData['email']);
 
                 $this->entityManager->persist($user);
                 $this->entityManager->flush();

src/Infrastructure/Command/UserProducerCommand.php

@@ -4,7 +4,13 @@
 
 namespace App\Infrastructure\Command;
 
+use AvroSchema;
 use Exception;
+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\Message\KafkaProducerMessage;
 use Jobcloud\Kafka\Producer\KafkaProducerBuilder;
 use League\Csv\Reader;
@@ -24,6 +30,36 @@ 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()
+        );
+
+        $recordSerializer = new RecordSerializer(
+            $schemaRegistryClient,
+            [
+                RecordSerializer::OPTION_REGISTER_MISSING_SCHEMAS => false,
+                RecordSerializer::OPTION_REGISTER_MISSING_SUBJECTS => true,
+            ]
+        );
+
+        // Define Avro Schema
+        $schema = <<<'JSON'
+        {
+            "type": "record",
+            "name": "User",
+            "fields": [
+                {"name": "name", "type": "string"},
+                {"name": "surname", "type": "string"},
+                {"name": "email", "type": "string"}
+            ]
+        }
+        JSON;
+
+        $avroSchema = AvroSchema::parse($schema);
+
         $producer = KafkaProducerBuilder::create()
             ->withAdditionalBroker('kafka:9092')
             ->build();
@@ -35,8 +71,14 @@ protected function execute(InputInterface $input, OutputInterface $output): int
 
         foreach ($records as $record) {
             try {
+                $avroData = $recordSerializer->encodeRecord('User', $avroSchema, [
+                    'name' => $record['Name'],
+                    'surname' => $record['Surname'],
+                    'email' => $record['Email']
+                ]);
+
                 $message = KafkaProducerMessage::create('users', RD_KAFKA_PARTITION_UA)
-                    ->withBody(json_encode($record, JSON_HEX_TAG));
+                    ->withBody($avroData);
 
                 $producer->produce($message);
                 $producer->flush(20000);