String Encoding serialization support

Fixes jlandersen#181

Signed-off-by: azerr <[email protected]>

CHANGELOG.md

@@ -8,6 +8,7 @@ All notable changes to `Tools for Apache Kafka®` are documented in this file.
 - Validation for available topics in `.kafka` files. See [#153](https://github.com/jlandersen/vscode-kafka/issues/153).
 - Simplify snippets. See [#180](https://github.com/jlandersen/vscode-kafka/pull/180).
 - Hover support in `.kafka` files. See [#149](https://github.com/jlandersen/vscode-kafka/issues/149).
+- String Encoding serialization support. See [#181](https://github.com/jlandersen/vscode-kafka/issues/181).
 
 ## [0.12.0] - 2021-04-26
 ### Added

docs/Consuming.md

@@ -20,11 +20,6 @@ Once this command is launched, it creates a consumer group (with an auto-generat
 
 In this case, the starting offset can be only be configured via the [kafka.consumers.offset](#kafkaconsumersoffset) preference.
 
-Known limitations:
-
-* UTF-8 encoded keys and values only. If data is encoded differently, it will not be pretty.
-* One consumer group is created per topic (may change in the future to just have one for the extension).
-
 ### Kafka file
 
 Define simple consumers in a `.kafka` file, using the following format:
@@ -58,7 +53,7 @@ The `CONSUMER` block defines:
 The deserializers can have the following value:
 
    * `none`: no deserializer (ignores content).
-   * `string`: similar deserializer to the Kafka Java client [org.apache.kafka.common.serialization.StringDeserializer](https://github.com/apache/kafka/blob/master/clients/src/main/java/org/apache/kafka/common/serialization/StringDeserializer.java) which currently only supports `UTF-8` encoding.
+   * `string`: similar deserializer to the Kafka Java client [org.apache.kafka.common.serialization.StringDeserializer](https://github.com/apache/kafka/blob/master/clients/src/main/java/org/apache/kafka/common/serialization/StringDeserializer.java). By default it supports `UTF-8` encoding, but you can specify the encoding as parameter like this `string(base64)`. The valid encoding values are defined in [Node.js' buffers and character encodings].
    * `double`: similar deserializer to the Kafka Java client [org.apache.kafka.common.serialization.DoubleDeserializer](https://github.com/apache/kafka/blob/master/clients/src/main/java/org/apache/kafka/common/serialization/DoubleDeserializer.java).
    * `float`: similar deserializer to the Kafka Java client [org.apache.kafka.common.serialization.FloatDeserializer](https://github.com/apache/kafka/blob/master/clients/src/main/java/org/apache/kafka/common/serialization/FloatDeserializer.java).
    * `integer`: similar deserializer to the Kafka Java client [org.apache.kafka.common.serialization.IntegerDeserializer](https://github.com/apache/kafka/blob/master/clients/src/main/java/org/apache/kafka/common/serialization/IntegerDeserializer.java).
@@ -85,6 +80,10 @@ Completion is available for
 
 ![Property value completion](assets/kafka-file-consumer-property-value-completion.png)
 
+ * string encoding:
+
+![String encoding completion](assets/kafka-file-consumer-string-encoding-completion.png)
+
  * topic:
 
 ![Topic completion](assets/kafka-file-consumer-topic-completion.png)

docs/Producing.md

@@ -36,7 +36,7 @@ The `PRODUCER` block defines:
 
 The serializers can have the following value:
 
-   * `string`: similar serializer to the Kafka Java client [org.apache.kafka.common.serialization.StringSerializer](https://github.com/apache/kafka/blob/master/clients/src/main/java/org/apache/kafka/common/serialization/StringSerializer.java) which currently only supports `UTF-8` encoding.
+   * `string`: similar serializer to the Kafka Java client [org.apache.kafka.common.serialization.StringSerializer](https://github.com/apache/kafka/blob/master/clients/src/main/java/org/apache/kafka/common/serialization/StringSerializer.java). By default it supports `UTF-8` encoding, but you can specify the encoding as parameter like this `string(base64)`. The valid encoding values are defined in [Node.js' buffers and character encodings].
    * `double`: similar serializer to the Kafka Java client [org.apache.kafka.common.serialization.DoubleSerializer](https://github.com/apache/kafka/blob/master/clients/src/main/java/org/apache/kafka/common/serialization/DoubleSerializer.java).
    * `float`: similar serializer to the Kafka Java client [org.apache.kafka.common.serialization.FloatSerializer](https://github.com/apache/kafka/blob/master/clients/src/main/java/org/apache/kafka/common/serialization/FloatSerializer.java).
    * `integer`: similar serializer to the Kafka Java client [org.apache.kafka.common.serialization.IntegerSerializer](https://github.com/apache/kafka/blob/master/clients/src/main/java/org/apache/kafka/common/serialization/IntegerSerializer.java).
@@ -59,6 +59,10 @@ Completion is available for
 
 ![Property value completion](assets/kafka-file-producer-property-value-completion.png)
 
+ * string encoding:
+
+![String encoding completion](assets/kafka-file-producer-string-encoding-completion.png)
+
  * randomized content (see following section):
 
 ![FakerJS completion](assets/kafka-file-producer-fakerjs-completion.png)

src/client/consumer.ts

@@ -4,15 +4,17 @@ import * as vscode from "vscode";
 import { ClientAccessor } from ".";
 import { getWorkspaceSettings, InitialConsumerOffset, ClusterSettings } from "../settings";
 import { addQueryParameter, Client, ConnectionOptions } from "./client";
-import { deserialize, MessageFormat, SerializationdResult } from "./serialization";
+import { deserialize, MessageFormat, SerializationdResult, SerializationSetting } from "./serialization";
 
 interface ConsumerOptions extends ConnectionOptions {
     consumerGroupId: string;
     topicId: string;
     fromOffset: InitialConsumerOffset | string;
     partitions?: number[];
     messageKeyFormat?: MessageFormat;
+    messageKeyFormatSettings?: SerializationSetting[];
     messageValueFormat?: MessageFormat;
+    messageValueFormatSettings?: SerializationSetting[];
 }
 
 export interface RecordReceivedEvent {
@@ -62,7 +64,7 @@ export class Consumer implements vscode.Disposable {
     public error: any;
 
     constructor(public uri: vscode.Uri, clusterSettings: ClusterSettings, private clientAccessor: ClientAccessor) {
-        const { clusterId, consumerGroupId, topicId, fromOffset, partitions, messageKeyFormat, messageValueFormat } = extractConsumerInfoUri(uri);
+        const { clusterId, consumerGroupId, topicId, fromOffset, partitions, messageKeyFormat, messageKeyFormatSettings, messageValueFormat,messageValueFormatSettings } = extractConsumerInfoUri(uri);
         this.clusterId = clusterId;
         const cluster = clusterSettings.get(clusterId);
 
@@ -81,7 +83,9 @@ export class Consumer implements vscode.Disposable {
                 fromOffset: fromOffset || settings.consumerOffset,
                 partitions: parsePartitions(partitions),
                 messageKeyFormat,
-                messageValueFormat
+                messageKeyFormatSettings,
+                messageValueFormat,
+                messageValueFormatSettings
             };
         }
         catch (e) {
@@ -114,8 +118,8 @@ export class Consumer implements vscode.Disposable {
 
         this.consumer.run({
             eachMessage: async ({ topic, partition, message }) => {
-                message.key = deserialize(message.key, this.options.messageKeyFormat);
-                message.value = deserialize(message.value, this.options.messageValueFormat);
+                message.key = deserialize(message.key, this.options.messageKeyFormat, this.options.messageKeyFormatSettings);
+                message.value = deserialize(message.value, this.options.messageValueFormat, this.options.messageValueFormatSettings);
                 this.onDidReceiveMessageEmitter.fire({
                     uri: this.uri,
                     record: { topic: topic, partition: partition, ...message },
@@ -360,14 +364,18 @@ export interface ConsumerInfoUri {
     fromOffset?: string;
     partitions?: string;
     messageKeyFormat?: MessageFormat;
+    messageKeyFormatSettings?: SerializationSetting[];
     messageValueFormat?: MessageFormat;
+    messageValueFormatSettings?: SerializationSetting[];
 }
 
 const TOPIC_QUERY_PARAMETER = 'topic';
 const FROM_QUERY_PARAMETER = 'from';
 const PARTITIONS_QUERY_PARAMETER = 'partitions';
 const KEY_FORMAT_QUERY_PARAMETER = 'key';
+const KEY_FORMAT_SETTINGS_QUERY_PARAMETER = 'key-settings';
 const VALUE_FORMAT_QUERY_PARAMETER = 'value';
+const VALUE_FORMAT_SETTINGS_QUERY_PARAMETER = 'value-settings';
 
 export function createConsumerUri(info: ConsumerInfoUri): vscode.Uri {
     const path = `kafka:${info.clusterId}/${info.consumerGroupId}`;
@@ -376,7 +384,9 @@ export function createConsumerUri(info: ConsumerInfoUri): vscode.Uri {
     query = addQueryParameter(query, FROM_QUERY_PARAMETER, info.fromOffset);
     query = addQueryParameter(query, PARTITIONS_QUERY_PARAMETER, info.partitions);
     query = addQueryParameter(query, KEY_FORMAT_QUERY_PARAMETER, info.messageKeyFormat);
+    query = addQueryParameter(query, KEY_FORMAT_SETTINGS_QUERY_PARAMETER, info.messageKeyFormatSettings?.map(p => p.value).join(','));
     query = addQueryParameter(query, VALUE_FORMAT_QUERY_PARAMETER, info.messageValueFormat);
+    query = addQueryParameter(query, VALUE_FORMAT_SETTINGS_QUERY_PARAMETER, info.messageValueFormatSettings?.map(p => p.value).join(','));
     return vscode.Uri.parse(path + query);
 }
 
@@ -387,7 +397,9 @@ export function extractConsumerInfoUri(uri: vscode.Uri): ConsumerInfoUri {
     const from = urlParams.get(FROM_QUERY_PARAMETER);
     const partitions = urlParams.get(PARTITIONS_QUERY_PARAMETER);
     const messageKeyFormat = urlParams.get(KEY_FORMAT_QUERY_PARAMETER);
+    const messageKeyFormatSettings = urlParams.get(KEY_FORMAT_SETTINGS_QUERY_PARAMETER);
     const messageValueFormat = urlParams.get(VALUE_FORMAT_QUERY_PARAMETER);
+    const messageValueFormatSettings = urlParams.get(VALUE_FORMAT_SETTINGS_QUERY_PARAMETER);
     const result: ConsumerInfoUri = {
         clusterId,
         consumerGroupId,
@@ -402,9 +414,19 @@ export function extractConsumerInfoUri(uri: vscode.Uri): ConsumerInfoUri {
     if (messageKeyFormat && messageKeyFormat.trim().length > 0) {
         result.messageKeyFormat = messageKeyFormat as MessageFormat;
     }
+    if (messageKeyFormatSettings) {
+        const settings = messageKeyFormatSettings.split(',').
+            map(value => <SerializationSetting>{ value });
+        result.messageKeyFormatSettings = settings;
+    }
     if (messageValueFormat && messageValueFormat.trim().length > 0) {
         result.messageValueFormat = messageValueFormat as MessageFormat;
     }
+    if (messageValueFormatSettings) {
+        const settings = messageValueFormatSettings.split(',').
+            map(value => <SerializationSetting>{ value });
+        result.messageValueFormatSettings = settings;
+    }
     return result;
 }
 

src/client/producer.ts

@@ -184,7 +184,7 @@ export interface ProducerInfoUri {
     clusterId: string;
     topicId?: string;
     key?: string;
-    value: string;
+    value?: string;
 }
 
 const TOPIC_QUERY_PARAMETER = 'topic';

src/client/serialization.ts

@@ -1,26 +1,31 @@
-export type MessageFormat = "none" | "string" | "double" | "float" | "integer" | "long" | "short" ;
+export type MessageFormat = "none" | "string" | "double" | "float" | "integer" | "long" | "short";
 
 export type SerializationdResult = any | Error;
 
 export class SerializationException extends Error { }
 
+export interface SerializationSetting {
+    name?: string;
+    value?: string;
+}
+
 // ---------------- Serializers ----------------
 
 interface Serializer {
-    serialize(data: string): Buffer | string | null;
+    serialize(data: string, settings?: SerializationSetting[]): Buffer | string | null;
 }
 
 const serializerRegistry: Map<MessageFormat, Serializer> = new Map();
 
-export function serialize(data?: string, format?: MessageFormat): Buffer | string | null   {
+export function serialize(data?: string, format?: MessageFormat, settings?: SerializationSetting[]): Buffer | string | null {
     if (!data || !format) {
         return data || null;
     }
     const serializer = getSerializer(format);
     if (!serializer) {
         throw new SerializationException(`Cannot find a serializer for ${format} format.`);
     }
-    return serializer.serialize(data);
+    return serializer.serialize(data, settings);
 }
 
 function getSerializer(format: MessageFormat): Serializer | undefined {
@@ -79,7 +84,11 @@ class ShortSerializer implements Serializer {
 
 class StringSerializer implements Serializer {
 
-    serialize(value: string): Buffer | string | null {
+    serialize(value: string, settings?: SerializationSetting[]): Buffer | string | null {
+        const encoding = settings?.[0].value;
+        if (encoding) {
+            return Buffer.from(value, <BufferEncoding>encoding);
+        }
         return value;
     };
 }
@@ -94,12 +103,12 @@ serializerRegistry.set("string", new StringSerializer());
 // ---------------- Deserializers ----------------
 
 interface Deserializer {
-    deserialize(data: Buffer): any;
+    deserialize(data: Buffer, settings?: SerializationSetting[]): any;
 }
 
 const deserializerRegistry: Map<MessageFormat, Deserializer> = new Map();
 
-export function deserialize(data: Buffer | null, format?: MessageFormat): SerializationdResult | null {
+export function deserialize(data: Buffer | null, format?: MessageFormat, settings?: SerializationSetting[]): SerializationdResult | null {
     if (data === null || !format) {
         return data;
     }
@@ -111,7 +120,7 @@ export function deserialize(data: Buffer | null, format?: MessageFormat): Serial
         if (!deserializer) {
             throw new SerializationException(`Cannot find a deserializer for ${format} format.`);
         }
-        return deserializer.deserialize(data);
+        return deserializer.deserialize(data, settings);
     }
     catch (e) {
         return e;
@@ -189,11 +198,12 @@ class ShortDeserializer implements Deserializer {
 
 class StringDeserializer implements Deserializer {
 
-    deserialize(data: Buffer | null): any {
+    deserialize(data: Buffer | null, settings?: SerializationSetting[]): any {
         if (data === null) {
             return null;
         }
-        return data.toString();
+        const encoding = settings?.[0].value;
+        return data.toString(encoding);
     }
 }
 

src/commands/producers.ts

@@ -7,15 +7,17 @@ import { OutputChannelProvider } from "../providers/outputChannelProvider";
 import { KafkaExplorer } from "../explorer";
 import { WorkspaceSettings } from "../settings";
 import { pickClient } from "./common";
-import { MessageFormat, serialize } from "../client/serialization";
+import { MessageFormat, SerializationSetting, serialize } from "../client/serialization";
 import { createProducerUri, ProducerCollection, ProducerInfoUri, ProducerLaunchState } from "../client/producer";
 import { ProducerRecord } from "kafkajs";
 import { ProducerValidator } from "../validators/producer";
 import { getErrorMessage } from "../errors";
 
 export interface ProduceRecordCommand extends ProducerInfoUri {
     messageKeyFormat?: MessageFormat;
+    messageKeyFormatSettings?: SerializationSetting[];
     messageValueFormat?: MessageFormat;
+    messageValueFormatSettings?: SerializationSetting[];
 }
 
 export class ProduceRecordCommandHandler {
@@ -46,6 +48,10 @@ export class ProduceRecordCommandHandler {
                 channel.appendLine("No topic");
                 return;
             }
+            if (value === undefined) {
+                channel.appendLine("No value");
+                return;
+            }
             if (this.settings.producerFakerJSEnabled) {
                 faker.setLocale(this.settings.producerFakerJSLocale);
             }
@@ -61,15 +67,15 @@ export class ProduceRecordCommandHandler {
                     faker.seed(seed);
                     const randomizedValue = faker.fake(value);
                     return {
-                        key: serialize(randomizedKey, command.messageKeyFormat),
-                        value: serialize(randomizedValue, command.messageValueFormat)
+                        key: serialize(randomizedKey, command.messageKeyFormat, command.messageKeyFormatSettings),
+                        value: serialize(randomizedValue, command.messageValueFormat, command.messageValueFormatSettings)
                     };
                 }
 
                 // Return key/value message as-is
                 return {
-                    key: serialize(key, command.messageKeyFormat),
-                    value: serialize(value, command.messageValueFormat)
+                    key: serialize(key, command.messageKeyFormat, command.messageKeyFormatSettings),
+                    value: serialize(value, command.messageValueFormat, command.messageValueFormatSettings)
                 };
             });
 

src/kafka-file/kafkaFileClient.ts

@@ -157,7 +157,7 @@ export function startLanguageClient(
     // Completion
     const completion = new KafkaFileCompletionItemProvider(kafkaFileDocuments, languageService, workspaceSettings);
     context.subscriptions.push(
-        vscode.languages.registerCompletionItemProvider(documentSelector, completion, ':', '{', '.')
+        vscode.languages.registerCompletionItemProvider(documentSelector, completion, ':', '{', '.', '(')
     );
 
     // Validation