chore(Pubsub): Fix Avro samples (GoogleCloudPlatform#1813)

Author: ajupazhamayil

pubsub/api/src/publish_avro_records.php

@@ -29,7 +29,7 @@
 use AvroStringIO;
 use AvroSchema;
 use AvroIODatumWriter;
-use AvroDataIOWriter;
+use AvroIOBinaryEncoder;
 
 /**
  * Publish a message using an AVRO schema.
@@ -81,14 +81,10 @@ function publish_avro_records($projectId, $topicId, $definitionFile)
         $io = new AvroStringIO();
         $schema = AvroSchema::parse($definition);
         $writer = new AvroIODatumWriter($schema);
-        $dataWriter = new AvroDataIOWriter($io, $writer, $schema);
+        $encoder = new AvroIOBinaryEncoder($io);
+        $writer->write($messageData, $encoder);
 
-        $dataWriter->append($messageData);
-
-        $dataWriter->close();
-
-        // AVRO binary data must be base64-encoded.
-        $encodedMessageData = base64_encode($io->string());
+        $encodedMessageData = $io->string();
     } else {
         // encode as JSON.
         $encodedMessageData = json_encode($messageData);

pubsub/api/src/subscribe_avro_records.php

@@ -31,24 +31,26 @@
  * @param string $projectId
  * @param string $subscriptionId
  */
-function subscribe_avro_records($projectId, $subscriptionId)
+function subscribe_avro_records($projectId, $subscriptionId, $definitionFile)
 {
     $pubsub = new PubSubClient([
         'projectId' => $projectId,
     ]);
 
     $subscription = $pubsub->subscription($subscriptionId);
+    $definition = file_get_contents($definitionFile);
     $messages = $subscription->pull();
 
     foreach ($messages as $message) {
         $decodedMessageData = '';
         $encoding = $message->attribute('googclient_schemaencoding');
         switch ($encoding) {
             case 'BINARY':
-                $ioReader = new \AvroStringIO(base64_decode($message->data()));
-                $dataReader = new \AvroDataIOReader($ioReader, new \AvroIODatumReader());
-
-                $decodedMessageData = json_encode($dataReader->data());
+                $io = new \AvroStringIO($message->data());
+                $schema = \AvroSchema::parse($definition);
+                $reader = new \AvroIODatumReader($schema);
+                $decoder = new \AvroIOBinaryDecoder($io);
+                $decodedMessageData = json_encode($reader->read($decoder));
                 break;
             case 'JSON':
                 $decodedMessageData = $message->data();

pubsub/api/test/SchemaTest.php

@@ -210,6 +210,7 @@ public function testPublishAndSubscribeAvro($encoding)
         $subscribeOutput = $this->runFunctionSnippet('subscribe_avro_records', [
             self::$projectId,
             $subscriptionId,
+            self::AVRO_DEFINITION,
         ]);
 
         $this->assertStringContainsString(