7.x - Implement AdminClient with topic management operations

admin_client.h

@@ -0,0 +1,62 @@
+/*
+  +----------------------------------------------------------------------+
+  | php-rdkafka                                                          |
+  +----------------------------------------------------------------------+
+  | Copyright (c) 2016 Arnaud Le Blanc                                   |
+  +----------------------------------------------------------------------+
+  | This source file is subject to version 3.01 of the PHP license,      |
+  | that is bundled with this package in the file LICENSE, and is        |
+  | available through the world-wide-web at the following url:           |
+  | http://www.php.net/license/3_01.txt                                  |
+  | If you did not receive a copy of the PHP license and are unable to   |
+  | obtain it through the world-wide-web, please send a note to          |
+  | license@php.net so we can mail you a copy immediately.               |
+  +----------------------------------------------------------------------+
+  | Author: Arnaud Le Blanc <arnaud.lb@gmail.com>                        |
+  +----------------------------------------------------------------------+
+*/
+
+#ifndef KAFKA_ADMIN_CLIENT_H
+#define KAFKA_ADMIN_CLIENT_H
+
+typedef struct _kafka_admin_client_object {
+    rd_kafka_t              *rk;
+    kafka_conf_callbacks    cbs;
+    zend_object             std;
+} kafka_admin_client_object;
+
+typedef struct _kafka_admin_options_object {
+    rd_kafka_AdminOptions_t *options;
+    zend_object              std;
+} kafka_admin_options_object;
+
+typedef struct _kafka_new_topic_object {
+    rd_kafka_NewTopic_t     *new_topic;
+    zend_object              std;
+} kafka_new_topic_object;
+
+typedef struct _kafka_delete_topic_object {
+    rd_kafka_DeleteTopic_t  *delete_topic;
+    zend_object              std;
+} kafka_delete_topic_object;
+
+typedef struct _kafka_new_partitions_object {
+    rd_kafka_NewPartitions_t *new_partitions;
+    zend_object               std;
+} kafka_new_partitions_object;
+
+void kafka_admin_client_minit(INIT_FUNC_ARGS);
+
+extern zend_class_entry *ce_kafka_admin_client;
+extern zend_class_entry *ce_kafka_admin_options;
+extern zend_class_entry *ce_kafka_new_topic;
+extern zend_class_entry *ce_kafka_delete_topic;
+extern zend_class_entry *ce_kafka_new_partitions;
+extern zend_class_entry *ce_kafka_topic_result;
+#ifdef HAS_RD_KAFKA_DESCRIBE_TOPICS
+extern zend_class_entry *ce_kafka_node;
+extern zend_class_entry *ce_kafka_topic_partition_info;
+extern zend_class_entry *ce_kafka_topic_description;
+#endif
+
+#endif /* KAFKA_ADMIN_CLIENT_H */

admin_client.stub.php

@@ -0,0 +1,158 @@
+<?php
+
+/**
+ * @generate-class-entries
+ * @generate-function-entries
+ * @generate-legacy-arginfo
+ */
+
+namespace RdKafka\Admin;
+
+class AdminClient
+{
+    private ?callable $error_cb;
+
+    private ?callable $dr_msg_cb;
+
+    public function __construct(\RdKafka\Conf $conf) {}
+
+    /** @tentative-return-type */
+    public function newAdminOptions(int $for_api): AdminOptions {}
+
+    /**
+     * @param NewTopic[] $new_topics
+     * @tentative-return-type
+     */
+    public function createTopics(array $new_topics, ?AdminOptions $options = null): array {}
+
+    /**
+     * @param DeleteTopic[] $delete_topics
+     * @tentative-return-type
+     */
+    public function deleteTopics(array $delete_topics, ?AdminOptions $options = null): array {}
+
+    /**
+     * @param NewPartitions[] $new_partitions
+     * @tentative-return-type
+     */
+    public function createPartitions(array $new_partitions, ?AdminOptions $options = null): array {}
+
+#ifdef HAS_RD_KAFKA_DESCRIBE_TOPICS
+    /**
+     * @param string[] $topics
+     * @tentative-return-type
+     */
+    public function describeTopics(array $topics, ?AdminOptions $options = null): array {}
+#endif
+
+    /**
+     * @param \RdKafka\TopicPartition[] $topic_partitions
+     * @return \RdKafka\TopicPartition[]
+     * @tentative-return-type
+     */
+    public function deleteRecords(array $topic_partitions, ?AdminOptions $options = null): array {}
+}
+
+class AdminOptions
+{
+    /** @tentative-return-type */
+    public function setRequestTimeout(int $timeout_ms): void {}
+
+    /** @tentative-return-type */
+    public function setOperationTimeout(int $timeout_ms): void {}
+
+    /** @tentative-return-type */
+    public function setValidateOnly(bool $validate_only): void {}
+
+    /** @tentative-return-type */
+    public function setBrokerId(int $broker_id): void {}
+
+#ifdef HAS_RD_KAFKA_DESCRIBE_TOPICS
+    /** @tentative-return-type */
+    public function setIncludeAuthorizedOperations(bool $include): void {}
+#endif
+}
+
+class NewTopic
+{
+    public function __construct(string $topic, int $num_partitions, int $replication_factor) {}
+
+    /** @tentative-return-type */
+    public function setReplicaAssignment(int $partition, array $broker_ids): void {}
+
+    /** @tentative-return-type */
+    public function setConfig(string $name, string $value): void {}
+}
+
+class DeleteTopic
+{
+    public function __construct(string $topic) {}
+}
+
+class NewPartitions
+{
+    public function __construct(string $topic, int $new_total_count) {}
+
+    /** @tentative-return-type */
+    public function setReplicaAssignment(int $new_partition_index, array $broker_ids): void {}
+}
+
+class TopicResult
+{
+    public int $error;
+
+    public ?string $error_string;
+
+    public string $name;
+
+    /** @tentative-return-type */
+    public function getError(): int {}
+
+    /** @tentative-return-type */
+    public function getErrorString(): ?string {}
+
+    /** @tentative-return-type */
+    public function getName(): string {}
+}
+
+#ifdef HAS_RD_KAFKA_DESCRIBE_TOPICS
+class Node
+{
+    public int $id;
+
+    public string $host;
+
+    public int $port;
+
+    public ?string $rack;
+}
+
+class TopicPartitionInfo
+{
+    public int $partition;
+
+    public ?Node $leader;
+
+    /** @var Node[] */
+    public array $isr;
+
+    /** @var Node[] */
+    public array $replicas;
+}
+
+class TopicDescription
+{
+    public string $name;
+
+    public ?string $topic_id;
+
+    public bool $is_internal;
+
+    public int $error;
+
+    public ?string $error_string;
+
+    /** @var TopicPartitionInfo[] */
+    public array $partitions;
+}
+#endif