Completion for available topic in kafka file

CHANGELOG.md

@@ -8,6 +8,7 @@ All notable changes to `Tools for Apache Kafka®` are documented in this file.
 - Declare key/value formats for PRODUCER in kafka file. See [#113](https://github.com/jlandersen/vscode-kafka/issues/113).
 - Completion support for property names and values of CONSUMER and PRODUCER blocks. See [#146](https://github.com/jlandersen/vscode-kafka/issues/146).
 - Completion support for fakerJS PRODUCER key and value. See [#152](https://github.com/jlandersen/vscode-kafka/issues/152).
+- Completion support for available topics for CONSUMER and PRODUCER blocks. See [#150](https://github.com/jlandersen/vscode-kafka/issues/150).
 
 ### Changed
 - Renamed extension as `Tools for Apache Kafka®`

docs/Consuming.md

@@ -85,6 +85,10 @@ Completion is available for
 
 ![Property value completion](assets/kafka-file-consumer-property-value-completion.png)
 
+ * topic:
+
+![Topic completion](assets/kafka-file-consumer-topic-completion.png)
+
 ### Start Consumer command
 
 ![Start Consumer from command palette](assets/start-consumer-from-command.png)

docs/Producing.md

@@ -59,10 +59,14 @@ Completion is available for
 
 ![Property value completion](assets/kafka-file-producer-property-value-completion.png)
 
- * randomized content (see following section)
+ * randomized content (see following section):
 
 ![FakerJS completion](assets/kafka-file-producer-fakerjs-completion.png)
 
+ * topic:
+
+![Topic completion](assets/kafka-file-producer-topic-completion.png)
+
 ## Randomized content
 
 Record content can be randomized by injecting mustache-like placeholders of [faker.js properties](https://github.com/Marak/faker.js#api-methods), like ``{{name.lastName}}`` or ``{{random.number}}``. Some randomized properties can be localized via the `kafka.producers.fakerjs.locale` setting.

src/explorer/kafkaExplorer.ts

@@ -4,7 +4,7 @@ import { ClientAccessor } from "../client";
 import { WorkspaceSettings, ClusterSettings } from "../settings";
 import { NodeBase } from "./models/nodeBase";
 import { TreeView } from "vscode";
-import { KafkaModel } from "./models/kafka";
+import { KafkaModel, KafkaModelProvider } from "./models/kafka";
 import { ClusterItem } from "./models/cluster";
 import { EOL } from 'os';
 import { TopicItem } from "./models/topics";
@@ -17,7 +17,7 @@ const TREEVIEW_ID = 'kafkaExplorer';
 /**
  * Kafka explorer to show in a tree clusters, topics.
  */
-export class KafkaExplorer implements vscode.Disposable, vscode.TreeDataProvider<NodeBase> {
+export class KafkaExplorer implements KafkaModelProvider, vscode.Disposable, vscode.TreeDataProvider<NodeBase> {
 
     private onDidChangeTreeDataEvent: vscode.EventEmitter<NodeBase | undefined>
         = new vscode.EventEmitter<NodeBase | undefined>();
@@ -63,10 +63,7 @@ export class KafkaExplorer implements vscode.Disposable, vscode.TreeDataProvider
 
     async getChildren(element?: NodeBase): Promise<NodeBase[]> {
         if (!element) {
-            if (!this.root) {
-                this.root = new KafkaModel(this.clusterSettings, this.clientAccessor);
-            }
-            element = this.root;
+            element = this.getDataModel();
         }
         return element.getChildren();
     }
@@ -180,4 +177,17 @@ export class KafkaExplorer implements vscode.Disposable, vscode.TreeDataProvider
         }
     }
 
+
+    /**
+     * Returns the kafka data model.
+     *
+     * @returns  the kafka data model.
+     */
+    public getDataModel(): KafkaModel {
+        if (!this.root) {
+            this.root = new KafkaModel(this.clusterSettings, this.clientAccessor);
+        }
+        return this.root;
+    }
+
 }

src/explorer/models/cluster.ts

@@ -61,14 +61,22 @@ export class ClusterItem extends NodeBase implements Disposable {
     }
 
     async findTopictemByName(topicName: string): Promise<NodeBase | TopicItem | undefined> {
-        const topics = (await this.getTopicGroupItem()).getChildren();
-        return topics
-            .then(t =>
-                t.find(child => (<TopicItem>child).topic.id === topicName));
+        const topics = await this.getTopics();
+        return topics.find(child => (<TopicItem>child).topic.id === topicName);
+    }
+
+    /**
+     * Returns the topics of the cluster.
+     * @returns the topics of the cluster.
+     */
+    async getTopics() {
+        return (await this.getTopicGroupItem()).getChildren();
     }
 
     private async getTopicGroupItem(): Promise<NodeBase> {
         return (await this.getChildren())[TOPIC_INDEX];
     }
 
 }
+
+

src/explorer/models/kafka.ts

@@ -4,6 +4,10 @@ import { ClusterSettings } from "../../settings";
 import { ClusterItem } from "./cluster";
 import { NodeBase } from "./nodeBase";
 
+export interface KafkaModelProvider {
+    getDataModel(): KafkaModel;
+}
+
 export class KafkaModel extends NodeBase implements Disposable {
 
     public contextValue = "";
@@ -33,10 +37,8 @@ export class KafkaModel extends NodeBase implements Disposable {
             );
     }
 
-    async findClusterItemById(clusterId: string): Promise<NodeBase | ClusterItem | undefined> {
-        return this.getChildren()
-            .then(clusters =>
-                clusters.find(child => (<ClusterItem>child).cluster.id === clusterId)
-            );
+    async findClusterItemById(clusterId: string): Promise<ClusterItem | undefined> {
+        const clusters = await this.getChildren();
+        return <ClusterItem>clusters.find(child => (<ClusterItem>child).cluster.id === clusterId);
     }
 }

src/extension.ts

@@ -138,7 +138,7 @@ export function activate(context: vscode.ExtensionContext): KafkaExtensionPartic
 
     // .kafka file related
     context.subscriptions.push(
-        startLanguageClient(clusterSettings, producerCollection, consumerCollection, context)
+        startLanguageClient(clusterSettings, producerCollection, consumerCollection, explorer, context)
     );
 
     context.subscriptions.push(

src/kafka-file/kafkaFileClient.ts

@@ -5,13 +5,16 @@ import { ClusterSettings } from "../settings/clusters";
 
 import { getLanguageModelCache, LanguageModelCache } from './languageModelCache';
 import { KafkaFileDocument } from "./languageservice/parser/kafkaFileParser";
-import { ConsumerLaunchStateProvider, getLanguageService, LanguageService, ProducerLaunchStateProvider, SelectedClusterProvider } from "./languageservice/kafkaFileLanguageService";
+import { ConsumerLaunchStateProvider, getLanguageService, LanguageService, ProducerLaunchStateProvider, SelectedClusterProvider, TopicDetail, TopicProvider } from "./languageservice/kafkaFileLanguageService";
 import { runSafeAsync } from "./utils/runner";
+import { TopicItem } from "../explorer";
+import { KafkaModelProvider } from "../explorer/models/kafka";
 
 export function startLanguageClient(
     clusterSettings: ClusterSettings,
     producerCollection: ProducerCollection,
     consumerCollection: ConsumerCollection,
+    modelProvider: KafkaModelProvider,
     context: vscode.ExtensionContext
 ): vscode.Disposable {
 
@@ -20,7 +23,7 @@ export function startLanguageClient(
     const kafkaFileDocuments = getLanguageModelCache<KafkaFileDocument>(10, 60, document => languageService.parseKafkaFileDocument(document));
 
     // Create the Kafka file language service.
-    const languageService = createLanguageService(clusterSettings, producerCollection, consumerCollection);
+    const languageService = createLanguageService(clusterSettings, producerCollection, consumerCollection, modelProvider);
 
     // Open / Close document
     context.subscriptions.push(vscode.workspace.onDidOpenTextDocument(e => {
@@ -74,7 +77,7 @@ export function startLanguageClient(
     };
 }
 
-function createLanguageService(clusterSettings: ClusterSettings, producerCollection: ProducerCollection, consumerCollection: ConsumerCollection): LanguageService {
+function createLanguageService(clusterSettings: ClusterSettings, producerCollection: ProducerCollection, consumerCollection: ConsumerCollection, modelProvider: KafkaModelProvider): LanguageService {
     const producerLaunchStateProvider = {
         getProducerLaunchState(uri: vscode.Uri): ProducerLaunchState {
             const producer = producerCollection.get(uri);
@@ -90,18 +93,29 @@ function createLanguageService(clusterSettings: ClusterSettings, producerCollect
     } as ConsumerLaunchStateProvider;
 
     const selectedClusterProvider = {
-
         getSelectedCluster() {
             const selected = clusterSettings.selected;
             return {
                 clusterId: selected?.id,
                 clusterName: selected?.name,
             };
         }
-
     } as SelectedClusterProvider;
 
-    return getLanguageService(producerLaunchStateProvider, consumerLaunchStateProvider, selectedClusterProvider);
+    const topicProvider = {
+        async getTopics(clusterId: string): Promise<TopicDetail[]> {
+            // Retrieve the proper cluster item from the explorer
+            const model = modelProvider.getDataModel();
+            const cluster = await model.findClusterItemById(clusterId);
+            if (!cluster) {
+                return [];
+            }
+            // Returns topics from the cluster
+            return (await cluster.getTopics()).map(child => (<TopicItem>child).topic);
+        }
+    } as TopicProvider;
+
+    return getLanguageService(producerLaunchStateProvider, consumerLaunchStateProvider, selectedClusterProvider, topicProvider);
 }
 
 class AbstractKafkaFileFeature {

src/kafka-file/languageservice/kafkaFileLanguageService.ts

@@ -26,6 +26,19 @@ export interface SelectedClusterProvider {
     getSelectedCluster(): { clusterId?: string, clusterName?: string };
 }
 
+export interface TopicDetail {
+    id: string;
+    partitionCount: number;
+    replicationFactor: number;
+}
+
+/**
+ * Provider API which gets topics from  given cluster id.
+ */
+export interface TopicProvider {
+    getTopics(clusterid: string): Promise<TopicDetail[]>;
+}
+
 /**
  * Kafka language service API.
  *
@@ -52,12 +65,12 @@ export interface LanguageService {
 
     /**
      * Returns the completion result for the given text document and parsed AST at given position.
-     * 
+     *
      * @param document the text document.
      * @param kafkaFileDocument the parsed AST.
      * @param position the position where the completion was triggered.
      */
-    doComplete(document: TextDocument, kafkaFileDocument: KafkaFileDocument, position: Position): CompletionList | undefined
+    doComplete(document: TextDocument, kafkaFileDocument: KafkaFileDocument, position: Position): Promise<CompletionList | undefined>
 }
 
 /**
@@ -66,11 +79,12 @@ export interface LanguageService {
  * @param producerLaunchStateProvider the provider which gets the state for a given producer.
  * @param consumerLaunchStateProvider the provider which gets the state for a given consumer.
  * @param selectedClusterProvider the provider which gets the selected cluster id and name.
+ * @param topicProvider the provider which returns topics from a given cluster id.
  */
-export function getLanguageService(producerLaunchStateProvider: ProducerLaunchStateProvider, consumerLaunchStateProvider: ConsumerLaunchStateProvider, selectedClusterProvider: SelectedClusterProvider): LanguageService {
+export function getLanguageService(producerLaunchStateProvider: ProducerLaunchStateProvider, consumerLaunchStateProvider: ConsumerLaunchStateProvider, selectedClusterProvider: SelectedClusterProvider, topicProvider: TopicProvider): LanguageService {
 
     const kafkaFileCodeLenses = new KafkaFileCodeLenses(producerLaunchStateProvider, consumerLaunchStateProvider, selectedClusterProvider);
-    const kafkaFileCompletion = new KafkaFileCompletion();
+    const kafkaFileCompletion = new KafkaFileCompletion(selectedClusterProvider, topicProvider);
     return {
         parseKafkaFileDocument: (document: TextDocument) => parseKafkaFile(document),
         getCodeLenses: kafkaFileCodeLenses.getCodeLenses.bind(kafkaFileCodeLenses),

src/kafka-file/languageservice/services/codeLensProvider.ts

@@ -5,6 +5,9 @@ import { LaunchConsumerCommand, ProduceRecordCommand, ProduceRecordCommandHandle
 import { ProducerLaunchStateProvider, ConsumerLaunchStateProvider, SelectedClusterProvider } from "../kafkaFileLanguageService";
 import { Block, BlockType, ConsumerBlock, KafkaFileDocument, ProducerBlock } from "../parser/kafkaFileParser";
 
+/**
+ * Kafka file codeLens support.
+ */
 export class KafkaFileCodeLenses {
 
     constructor(private producerLaunchStateProvider: ProducerLaunchStateProvider, private consumerLaunchStateProvider: ConsumerLaunchStateProvider, private selectedClusterProvider: SelectedClusterProvider) {