From 6241bd087d1fb3a6d793b88d242cf7e85631e15c Mon Sep 17 00:00:00 2001
From: Paolo Haider <51787541+xpr3sso@users.noreply.github.com>
Date: Thu, 16 Sep 2021 22:13:36 +0200
Subject: [PATCH] feat(ui): add multi-message produce feature (#791)

Co-authored-by: xpr3sso <xpr3sso@noreply.org>
---
 client/src/components/Form/Form.jsx           | 21 +++-
 client/src/components/Form/styles.scss        | 16 ++-
 .../Topic/TopicProduce/TopicProduce.jsx       | 98 +++++++++++++++----
 .../org/akhq/controllers/TopicController.java | 28 +++---
 src/main/java/org/akhq/models/KeyValue.java   | 14 +++
 .../akhq/repositories/RecordRepository.java   | 48 +++++++++
 .../akhq/controllers/TopicControllerTest.java | 53 +++++++---
 7 files changed, 231 insertions(+), 47 deletions(-)
 create mode 100644 src/main/java/org/akhq/models/KeyValue.java

diff --git a/client/src/components/Form/Form.jsx b/client/src/components/Form/Form.jsx
index 7898a5301..e6909fdad 100644
--- a/client/src/components/Form/Form.jsx
+++ b/client/src/components/Form/Form.jsx
@@ -112,8 +112,9 @@ class Form extends Root {
     );
   };
 
-  renderJSONInput = (name, label, onChange) => {
+  renderJSONInput = (name, label, onChange, textMode, options) => {
     const { formData, errors } = this.state;
+    const inputMode = textMode ? "text" : (formData.schemaType === "PROTOBUF" ? "protobuf"  : "json")
     return (
       <div className="form-group row">
         {label !== '' ? (
@@ -125,7 +126,7 @@ class Form extends Root {
         )}
         <div className="col-sm-10" style={{ height: '100%' }}>
           <AceEditor
-            mode={ formData.schemaType === "PROTOBUF"? "protobuf"  : "json" }
+            mode={ inputMode }
             id={name}
             theme="merbivore_soft"
             value={formData[name]}
@@ -134,6 +135,7 @@ class Form extends Root {
             }}
             name="UNIQUE_ID_OF_DIV"
             editorProps={{ $blockScrolling: true }}
+            setOptions={options}
             style={{ width: '100%', minHeight: '25vh' }}
           />
           {errors[name] && <div className="alert alert-danger mt-1 p-1">{errors[name]}</div>}
@@ -238,6 +240,21 @@ class Form extends Root {
       </React.Fragment>
     );
   };
+
+  renderCheckbox = (name, label, isChecked, onChange, isDefaultChecked) => {
+    return (
+        <input
+        type="checkbox"
+        name={name}
+        id={name}
+        class="form-input-check"
+        checked={isChecked}
+        onChange={onChange}
+        defaultChecked={ isDefaultChecked ? isDefaultChecked : false}
+        />
+    );
+  };
 }
 
 export default Form;
+ 
diff --git a/client/src/components/Form/styles.scss b/client/src/components/Form/styles.scss
index 8e0808a46..c7c7b290e 100644
--- a/client/src/components/Form/styles.scss
+++ b/client/src/components/Form/styles.scss
@@ -24,4 +24,18 @@ input.placeholder{
     border-bottom-color: transparent !important;
     width: 100%;
     padding: 0px;
-}
\ No newline at end of file
+}
+
+.ace_active-line {
+    opacity: 0.2;
+}
+
+.ace_placeholder {
+    font-family: "Source Code Pro",SFMono-Regular,Menlo,Monaco,Consolas,"Liberation Mono","Courier New",monospace;
+    transform: scale(1);
+}
+
+.form-input-check{
+    margin-block: auto;
+}
+ 
diff --git a/client/src/containers/Topic/TopicProduce/TopicProduce.jsx b/client/src/containers/Topic/TopicProduce/TopicProduce.jsx
index 83a0aaf0e..bf79f2492 100644
--- a/client/src/containers/Topic/TopicProduce/TopicProduce.jsx
+++ b/client/src/containers/Topic/TopicProduce/TopicProduce.jsx
@@ -24,7 +24,8 @@ class TopicProduce extends Form {
       key: '',
       hKey0: '',
       hValue0: '',
-      value: ''
+      value: '',
+      keyValueSeparator: ':'
     },
     datetime: new Date(),
     openDateModal: false,
@@ -40,7 +41,9 @@ class TopicProduce extends Form {
     valueSchemaSearchValue: '',
     selectedValueSchema: '',
     clusterId: '',
-    topicId: ''
+    topicId: '',
+    multiMessage: false,
+    valuePlaceholder: '{"param": "value"}'
   };
 
   schema = {
@@ -58,7 +61,10 @@ class TopicProduce extends Form {
       .label('hValue0'),
     value: Joi.string()
       .allow('')
-      .label('Value')
+      .label('Value'),
+    keyValueSeparator: Joi.string()
+      .min(1)
+      .label('keyValueSeparator')
   };
 
   async componentDidMount() {
@@ -98,7 +104,8 @@ class TopicProduce extends Form {
       selectedKeySchema,
       selectedValueSchema,
       keySchema,
-      valueSchema
+      valueSchema,
+      multiMessage
     } = this.state;
     const { clusterId, topicId } = this.props.match.params;
 
@@ -110,9 +117,11 @@ class TopicProduce extends Form {
       partition: formData.partition,
       key: formData.key,
       timestamp: datetime.toISOString(),
-      value: JSON.parse(JSON.stringify(formData.value)),
+      value: multiMessage ? formData.value : JSON.parse(JSON.stringify(formData.value)),
       keySchema: schemaKeyToSend ? schemaKeyToSend.id : '',
-      valueSchema: schemaValueToSend ? schemaValueToSend.id : ''
+      valueSchema: schemaValueToSend ? schemaValueToSend.id : '',
+      multiMessage: multiMessage,
+      keyValueSeparator: formData.keyValueSeparator
     };
 
     let headers = {};
@@ -134,6 +143,55 @@ class TopicProduce extends Form {
       });
   }
 
+  renderMultiMessage() {
+    const { formData, multiMessage } = this.state;
+
+    return (
+      <div className="form-group row">
+        <label className="col-sm-2 col-form-label">Multi message</label>
+          <div className="row khq-multiple col-sm-7">
+            {this.renderCheckbox(
+              'isMultiMessage',
+              '',
+              multiMessage,
+              () => {
+                this.setState({multiMessage: !multiMessage,
+                  valuePlaceholder: this.getPlaceholderValue(!multiMessage, formData.keyValueSeparator)})
+              },
+              false
+            )}
+
+            <label className="col-auto col-form-label">Separator</label>
+            <input
+                type='text'
+                name='keyValueSeparator'
+                id='keyValueSeparator'
+                placeholder=':'
+                class='col-sm-2 form-control'
+                disabled={ !multiMessage }
+                onChange={
+                    event => {
+                        this.setState({
+                            formData: { ...formData,
+                                keyValueSeparator: event.target.value},
+                            valuePlaceholder: this.getPlaceholderValue(!multiMessage, event.target.value)})
+                    }
+                }
+            />
+          </div>
+      </div>
+    );
+  }
+
+  getPlaceholderValue(isMultiMessage, keyValueSeparator) {
+    if(isMultiMessage) {
+      return "key1" + keyValueSeparator + "{\"param\": \"value1\"}\n"
+        + "key2" + keyValueSeparator + "{\"param\": \"value2\"}";
+    } else {
+        return '{"param": "value"}';
+    }
+  }
+
   renderHeaders() {
     let headers = [];
 
@@ -284,13 +342,13 @@ class TopicProduce extends Form {
       selectedKeySchema,
       valueSchema,
       valueSchemaSearchValue,
-      selectedValueSchema
+      selectedValueSchema,
+      multiMessage
     } = this.state;
     let date = moment(datetime);
     return (
       <div>
         <form encType="multipart/form-data" className="khq-form khq-form-config">
-          <div>
             <Header title={`Produce to ${topicId} `} />
             {this.renderSelect('partition', 'Partition', partitions, value => {
               this.setState({ formData: { ...formData, partition: value.target.value } });
@@ -313,11 +371,11 @@ class TopicProduce extends Form {
                 'key'
               )
             )}
-            {this.renderInput('key', 'Key', 'Key', 'Key')}
-            <div></div>
-          </div>
+
+          {(this.renderInput('key', 'Key', 'Key', 'Key', undefined, undefined, undefined, undefined,{ disabled: multiMessage }))}
 
           {this.renderHeaders()}
+
           {this.renderDropdown(
             'Value schema',
             valueSchema.map(value => value.subject),
@@ -336,14 +394,19 @@ class TopicProduce extends Form {
               'value'
             )
           )}
+
+          {this.renderMultiMessage()}
+
           {this.renderJSONInput('value', 'Value', value => {
             this.setState({
-              formData: {
-                ...formData,
-                value: value
-              }
-            });
-          })}
+                formData: {
+                    ...formData,
+                    value: value
+                }
+            })},
+          multiMessage, // true -> 'text' mode; json, protobuff, ... mode otherwise
+          { placeholder: this.getPlaceholderValue(multiMessage, formData.keyValueSeparator) }
+          )}
           <div style={{ display: 'flex', flexDirection: 'row', width: '100%', padding: 0 }}>
             <label
               style={{ padding: 0, alignItems: 'center', display: 'flex' }}
@@ -429,3 +492,4 @@ class TopicProduce extends Form {
 }
 
 export default TopicProduce;
+ 
diff --git a/src/main/java/org/akhq/controllers/TopicController.java b/src/main/java/org/akhq/controllers/TopicController.java
index c1d5bd98a..91fb8c460 100644
--- a/src/main/java/org/akhq/controllers/TopicController.java
+++ b/src/main/java/org/akhq/controllers/TopicController.java
@@ -123,7 +123,7 @@ public Topic create(
     @Secured(Role.ROLE_TOPIC_DATA_INSERT)
     @Post(value = "api/{cluster}/topic/{topicName}/data")
     @Operation(tags = {"topic data"}, summary = "Produce data to a topic")
-    public Record produce(
+    public List<Record> produce(
         HttpRequest<?> request,
         String cluster,
         String topicName,
@@ -133,9 +133,12 @@ public Record produce(
         Optional<String> timestamp,
         Map<String, String> headers,
         Optional<Integer> keySchema,
-        Optional<Integer> valueSchema
+        Optional<Integer> valueSchema,
+        Boolean multiMessage,
+        Optional<String> keyValueSeparator
     ) throws ExecutionException, InterruptedException {
-        return new Record(
+        Topic targetTopic = topicRepository.findByName(cluster, topicName);
+        return
             this.recordRepository.produce(
                 cluster,
                 topicName,
@@ -145,14 +148,16 @@ public Record produce(
                 partition,
                 timestamp.map(r -> Instant.parse(r).toEpochMilli()),
                 keySchema,
-                valueSchema
-            ),
-            schemaRegistryRepository.getSchemaRegistryType(cluster),
-            key.map(String::getBytes).orElse(null),
-            value.getBytes(),
-            headers,
-            topicRepository.findByName(cluster, topicName)
-        );
+                valueSchema,
+                multiMessage,
+                keyValueSeparator).stream()
+                    .map(recordMetadata -> new Record(recordMetadata,
+                            schemaRegistryRepository.getSchemaRegistryType(cluster),
+                            key.map(String::getBytes).orElse(null),
+                            value.getBytes(),
+                            headers,
+                            targetTopic))
+                    .collect(Collectors.toList());
     }
 
     @Secured(Role.ROLE_TOPIC_DATA_READ)
@@ -504,3 +509,4 @@ public static class OffsetCopy {
         private long offset;
     }
 }
+ 
diff --git a/src/main/java/org/akhq/models/KeyValue.java b/src/main/java/org/akhq/models/KeyValue.java
new file mode 100644
index 000000000..d62ff260b
--- /dev/null
+++ b/src/main/java/org/akhq/models/KeyValue.java
@@ -0,0 +1,14 @@
+package org.akhq.models;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+
+/**
+ * Represents a simple key-value pair of any type
+ */
+@Getter
+@AllArgsConstructor
+public class KeyValue<K,V> {
+    K key;
+    V value;
+}
diff --git a/src/main/java/org/akhq/repositories/RecordRepository.java b/src/main/java/org/akhq/repositories/RecordRepository.java
index 7ccaaffe9..b9bc7afa9 100644
--- a/src/main/java/org/akhq/repositories/RecordRepository.java
+++ b/src/main/java/org/akhq/repositories/RecordRepository.java
@@ -13,6 +13,7 @@
 import lombok.extern.slf4j.Slf4j;
 import org.akhq.configs.SchemaRegistryType;
 import org.akhq.controllers.TopicController;
+import org.akhq.models.KeyValue;
 import org.akhq.models.Partition;
 import org.akhq.models.Record;
 import org.akhq.models.Topic;
@@ -463,6 +464,35 @@ private Record newRecord(ConsumerRecord<byte[], byte[]> record, BaseOptions opti
         );
     }
 
+    public List<RecordMetadata> produce(
+            String clusterId,
+            String topic,
+            String value,
+            Map<String, String> headers,
+            Optional<String> key,
+            Optional<Integer> partition,
+            Optional<Long> timestamp,
+            Optional<Integer> keySchemaId,
+            Optional<Integer> valueSchemaId,
+            Boolean multiMessage,
+            Optional<String> keyValueSeparator) throws ExecutionException, InterruptedException {
+
+        List<RecordMetadata> produceResults = new ArrayList<>();
+
+        // Distinguish between single record produce, and multiple messages
+        if (multiMessage.booleanValue()) {
+            // Split key-value pairs and produce them
+            for (KeyValue<String, String> kvPair : splitMultiMessage(value, keyValueSeparator.orElseThrow())) {
+                produceResults.add(produce(clusterId, topic, kvPair.getValue(), headers, Optional.of(kvPair.getKey()),
+                        partition, timestamp, keySchemaId, valueSchemaId));
+            }
+        } else {
+            produceResults.add(
+                    produce(clusterId, topic, value, headers, key, partition, timestamp, keySchemaId, valueSchemaId));
+        }
+        return produceResults;
+    }
+
     private RecordMetadata produce(
         String clusterId,
         String topic, byte[] value,
@@ -492,6 +522,23 @@ private RecordMetadata produce(
             .get();
     }
 
+    /**
+     * Splits a multi-message into a list of key-value pairs.
+     * @param value The multi-message string submitted by the {@link TopicController}
+     * @param keyValueSeparator The character(s) separating each key from their corresponding value
+     * @return A list of {@link KeyValue}, holding the split pairs
+     */
+    private List<KeyValue<String, String>> splitMultiMessage(String value, String keyValueSeparator) {
+        return List.of(value.split("\r\n|\r|\n")).stream().map(v -> splitKeyValue(v, keyValueSeparator))
+                .collect(Collectors.toList());
+    }
+
+    private KeyValue<String, String> splitKeyValue(String keyValueStr, String keyValueSeparator) {
+        String[] keyValue = null;
+        keyValue = keyValueStr.split(keyValueSeparator, 2);
+        return new KeyValue<>(keyValue[0].trim(),keyValue[1]);
+    }
+
     public void emptyTopic(String clusterId, String topicName) throws ExecutionException, InterruptedException {
         Map<TopicPartition, RecordsToDelete> recordsToDelete = new HashMap<>();
         var topic = topicRepository.findByName(clusterId, topicName);
@@ -1215,3 +1262,4 @@ private static class EndOffsetBound {
         private final KafkaConsumer<byte[], byte[]> consumer;
     }
 }
+ 
diff --git a/src/test/java/org/akhq/controllers/TopicControllerTest.java b/src/test/java/org/akhq/controllers/TopicControllerTest.java
index fad3712c3..ea6312edf 100644
--- a/src/test/java/org/akhq/controllers/TopicControllerTest.java
+++ b/src/test/java/org/akhq/controllers/TopicControllerTest.java
@@ -11,7 +11,9 @@
 import org.junit.jupiter.api.*;
 import org.akhq.models.Record;
 import java.util.Base64;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Random;
 import java.util.UUID;
 
@@ -153,24 +155,24 @@ void checkConfigs() {
     @Test
     @Order(3)
     void produce() {
-        Record record = this.retrieve(HttpRequest.POST(
-            CREATE_TOPIC_URL + "/data",
-            ImmutableMap.of(
-                "value", "my-value",
-                "key", "my-key",
-                "partition", 1,
-                "headers", ImmutableMap.of(
-                    "my-header-1", "1",
-                    "my-header-2", "2"
-                )
-            )
+        Map<String, Object> paramMap = new HashMap<>();
+        paramMap.put("value", "my-value");
+        paramMap.put("key", "my-key");
+        paramMap.put("partition", 1);
+        paramMap.put("headers", ImmutableMap.of(
+                "my-header-1", "1",
+                "my-header-2", "2"));
+        paramMap.put("multiMessage", false);
+        List<Record> response = this.retrieveList(HttpRequest.POST(
+            CREATE_TOPIC_URL + "/data", paramMap
         ), Record.class);
 
-        assertEquals("my-key", record.getKey());
-        assertEquals("my-value", record.getValue());
-        assertEquals(1, record.getPartition());
-        assertEquals(2, record.getHeaders().size());
-        assertEquals("1", record.getHeaders().get("my-header-1"));
+        assertEquals(1, response.size());
+        assertEquals("my-key", response.get(0).getKey());
+        assertEquals("my-value", response.get(0).getValue());
+        assertEquals(1, response.get(0).getPartition());
+        assertEquals(2, response.get(0).getHeaders().size());
+        assertEquals("1", response.get(0).getHeaders().get("my-header-1"));
     }
 
     @Test
@@ -206,6 +208,25 @@ void dataDelete() {
 
     @Test
     @Order(6)
+    void produceMultipleMessages() {
+        Map<String, Object> paramMap = new HashMap<String, Object>();
+        paramMap.put("value", "key1_{\"test_1\":1}\n"
+                            + "key2_{\"test_1\":2}\n"
+                            + "key3_{\"test_1\":3}");
+        paramMap.put("multiMessage", true);
+        paramMap.put("keyValueSeparator", "_");
+        List<Record> response = this.retrieveList(HttpRequest.POST(
+            CREATE_TOPIC_URL + "/data", paramMap
+        ), Record.class);
+
+        assertEquals(3, response.size());
+        assertTrue(response.get(0).getValue().contains("key1_{\"test_1\":1}"));
+        assertTrue(response.get(1).getValue().contains("key2_{\"test_1\":2}"));
+        assertTrue(response.get(2).getValue().contains("key3_{\"test_1\":3}"));
+    }
+
+    @Test
+    @Order(8)
     void delete() {
         this.exchange(
             HttpRequest.DELETE(CREATE_TOPIC_URL)