feat(PubSub): Add Schema Revision Support (#6648)

PubSub/src/Connection/ConnectionInterface.php

@@ -189,4 +189,19 @@ public function validateSchema(array $args);
      * @param array $args
      */
     public function validateMessage(array $args);
+
+    /**
+     * @param array $args
+     */
+    public function listRevisions(array $args);
+
+    /**
+     * @param array $args
+     */
+    public function commitSchema(array $args);
+
+    /**
+     * @param array $args
+     */
+    public function deleteRevision(array $args);
 }

PubSub/src/Connection/Grpc.php

@@ -761,4 +761,48 @@ protected function getSchemaClient()
         $this->schemaClient = $this->constructGapic(SchemaServiceClient::class, $this->clientConfig);
         return $this->schemaClient;
     }
+
+    /**
+     * Retrieve schema revisions
+     *
+     * @param array $args
+     * @return array
+     */
+    public function listRevisions(array $args)
+    {
+        return $this->send([$this->getSchemaClient(), 'listSchemaRevisions'], [
+            $this->pluck('name', $args),
+            $args,
+        ]);
+    }
+
+    /**
+     * Create schema revisions
+     *
+     * @param array $args
+     * @return array
+     */
+    public function commitSchema(array $args)
+    {
+        return $this->send([$this->getSchemaClient(), 'commitSchema'], [
+            $this->pluck('name', $args),
+            new Schema($this->pluck('schema', $args)),
+            $args,
+        ]);
+    }
+
+    /**
+     * Delete schema revision
+     *
+     * @param array $args
+     * @return array
+     */
+    public function deleteRevision(array $args)
+    {
+        return $this->send([$this->getSchemaClient(), 'deleteSchemaRevision'], [
+            $this->pluck('name', $args),
+            null,
+            $args,
+        ]);
+    }
 }

PubSub/src/Connection/Rest.php

@@ -348,4 +348,28 @@ public function validateMessage(array $args)
     {
         return $this->send('schemas', 'validateMessage', $args);
     }
+
+    /**
+     * @param array $args
+     */
+    public function listRevisions(array $args)
+    {
+        return $this->send('schemas', 'listRevisions', $args);
+    }
+
+    /**
+     * @param array $args
+     */
+    public function commitSchema(array $args)
+    {
+        return $this->send('schemas', 'commit', $args);
+    }
+
+    /**
+     * @param array $args
+     */
+    public function deleteRevision(array $args)
+    {
+        return $this->send('schemas', 'deleteRevision', $args);
+    }
 }

PubSub/src/Schema.php

@@ -197,4 +197,77 @@ public function exists(array $options = [])
             return false;
         }
     }
+
+    /**
+     * Get list  schema revisions.
+     *
+     * Example:
+     * ```
+     * $revisions = $schema->listRevisions();
+     * foreach ($revisions['schemas'] as $revision) {
+     *     echo $revisions['definition'];
+     * }
+     * ```
+     * @see https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.schemas/listRevisions List revisions
+     * @param array $options [optional] Configuration Options
+     * @return array
+     */
+    public function listRevisions(array $options = [])
+    {
+        return $this->connection->listRevisions([
+                'name' => $this->name,
+            ] + $options);
+    }
+
+    /**
+     * Commit schema revision.
+     *
+     * Example:
+     * ```
+     * $definition = file_get_contents('my-schema.txt');
+     * $revision = $schema->commit($definition, 'AVRO);
+     *
+     * @see https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.schemas/commit Commit Schema revision.
+     * ```
+     *
+     * @param string $definition The definition of the schema. This should
+     *     contain a string representing the full definition of the schema that
+     *     is a valid schema definition of the type specified in `type`. See
+     *     [Schema](https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.schemas#Schema)
+     *     for details.
+     * @param string $type The schema type. Allowed values are `AVRO` and `PROTOCOL_BUFFER`.
+     * @param array $options [optional] Configuration options
+     * @return array revision created
+     */
+    public function commit($definition, $type, array $options = [])
+    {
+        return $this->connection->commitSchema([
+                'schema' => [
+                    'definition' => $definition,
+                    'type' => $type,
+                ],
+                'name' => $this->name
+            ] + $options);
+    }
+
+    /**
+     * Commit schema revision.
+     *
+     * Example:
+     * ```
+     * $definition = file_get_contents('my-schema.txt');
+     * $revision = $schema->commit($definition, 'AVRO);
+     *
+     * @see https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.schemas/deleteRevision Delete revision.
+     * ```
+     *
+     * @param string $revisionId The revisionId
+     * @return array deleted revision
+     */
+    public function deleteRevision($revisionId)
+    {
+        return $this->connection->deleteRevision([
+            'name' => $this->name .'@' . $revisionId
+        ]);
+    }
 }

PubSub/tests/Unit/Connection/RestTest.php

@@ -128,6 +128,9 @@ public function methodProvider()
             ['deleteSchema'],
             ['validateSchema'],
             ['validateMessage'],
+            ['listRevisions'],
+            ['commitSchema'],
+            ['deleteRevision'],
         ];
     }
 }

PubSub/tests/Unit/SchemaTest.php

@@ -116,4 +116,48 @@ public function testExistsReturnsFalse()
 
         $this->assertFalse($this->schema->exists());
     }
+
+    public function testlistRevisions()
+    {
+        $this->connection
+            ->listRevisions(['name' => self::NAME,])
+            ->shouldBeCalledOnce()
+            ->willReturn(['foo' => 'bar']);
+
+        $this->schema->___setProperty('connection', $this->connection->reveal());
+
+        $this->assertEquals('bar', $this->schema->listRevisions()['foo']);
+    }
+
+    public function testCommit()
+    {
+        $this->connection
+            ->commitSchema(
+                [
+                    'name' => self::NAME,
+                    'schema' => [
+                        'definition' => 'test',
+                        'type' => 'AVRO',
+                    ],
+                ]
+            )
+            ->shouldBeCalledOnce()
+            ->willReturn(['foo' => 'bar']);
+
+        $this->schema->___setProperty('connection', $this->connection->reveal());
+
+        $this->assertEquals('bar', $this->schema->commit('test', 'AVRO')['foo']);
+    }
+
+    public function testDeleteRevision()
+    {
+        $this->connection
+            ->deleteRevision(['name' => self::NAME . '@1234567'])
+            ->shouldBeCalledOnce()
+            ->willReturn(['foo' => 'bar']);
+
+        $this->schema->___setProperty('connection', $this->connection->reveal());
+
+        $this->assertEquals('bar', $this->schema->deleteRevision('1234567')['foo']);
+    }
 }