Produce message

assets/css/main.scss

@@ -7,7 +7,9 @@
 @import "../modules/datas/filter";
 @import "../modules/datas/highlight";
 
+@import "../modules/forms/ace";
 @import "../modules/forms/datetime";
+@import "../modules/forms/multiple";
 
 @import "../modules/tables/table";
 @import "../modules/tables/row-action";

assets/css/vendor.scss

@@ -4,6 +4,7 @@
 @import "../../node_modules/font-awesome/scss/font-awesome";
 @import "../../node_modules/sweetalert2/src/sweetalert2";
 $bs-datetimepicker-btn-hover-bg: $gray-800;
+$bs-datetimepicker-primary-border-color: $dropdown-border-color;
 @import "../../node_modules/tempusdominus-bootstrap-4/src/sass/tempusdominus-bootstrap-4";
 @import "../../node_modules/highlight.js/styles/darkula.css";
 

assets/js/main.js

@@ -7,8 +7,10 @@ import '../modules/datas/data-confirm';
 import '../modules/datas/highlight';
 import '../modules/datas/search-navbar';
 import '../modules/datas/search-sse';
+import '../modules/forms/ace';
 import '../modules/forms/config';
 import '../modules/forms/get';
+import '../modules/forms/multiple';
 import '../modules/tables/row-action';
 import '../modules/templates/sidebar';
 import '../modules/templates/toast'

assets/modules/forms/ace.js

@@ -0,0 +1,26 @@
+import $ from "jquery";
+import "../widget";
+import ace from 'ace-builds';
+import "ace-builds/src-noconflict/theme-tomorrow";
+import "ace-builds/webpack-resolver";
+
+$.widget("khq.ace-editor", $.khq.widget, {
+
+    _create: function () {
+        let textarea = this.element.find('> textarea');
+
+        let editor = ace.edit(this.element.find('> div')[0], {
+            minLines: 5,
+            maxLines: 48,
+            autoScrollEditorIntoView: true,
+            theme: "ace/theme/tomorrow"
+        });
+
+        editor.renderer.setScrollMargin(10, 10, 10, 10);
+
+        editor.getSession().setValue(textarea.val());
+        editor.getSession().on('change', function () {
+            textarea.val(editor.getSession().getValue());
+        });
+    }
+});

assets/modules/forms/ace.scss

@@ -0,0 +1,9 @@
+.khq-ace-editor {
+    > div {
+        height: $input-height-inner * 3 ;
+    }
+
+    > textarea {
+        display: none;
+    }
+}

assets/modules/forms/datetime.js

@@ -8,6 +8,25 @@ $.widget("khq.datetime", $.khq.widget, {
         const self = this;
         const element = this.element;
 
+        if (element.find('.datetime-container').length > 0) {
+            this._inline(element);
+        } else {
+            this._hover(element);
+        }
+
+    },
+
+    _onChange: function(event) {
+        const element = this.element;
+
+        if (event.date) {
+            element
+                .find('input')
+                .val(event.date.toISOString());
+        }
+    },
+
+    _inline: function (element) {
         let datetimeTempusOptions = {
             inline: true,
             sideBySide: true,
@@ -20,14 +39,17 @@ $.widget("khq.datetime", $.khq.widget, {
 
         element
             .find('.datetime-container')
-            .on('change.datetimepicker', function (event) {
-                if (event.date) {
-                    element
-                        .find('input')
-                        .val(event.date.toISOString());
-                }
-            })
+            .on('change.datetimepicker', $.proxy(this._onChange, this))
             .datetimepicker(datetimeTempusOptions);
+    },
 
+    _hover: function (element) {
+        element
+            .find('input')
+            .on('change.datetimepicker', $.proxy(this._onChange, this))
+            .datetimepicker({
+                sideBySide: true,
+                useCurrent: false,
+            });
     },
 });

assets/modules/forms/datetime.scss

@@ -11,25 +11,25 @@
                 border-top: 2px solid $gray-700;
             }
         }
+    }
+
+    .bootstrap-datetimepicker-widget {
+        .timepicker-hour, .timepicker-minute, .timepicker-second {
+            width: 16px;
+            font-weight: normal;
+        }
 
-        .bootstrap-datetimepicker-widget {
-            .timepicker-hour, .timepicker-minute, .timepicker-second {
+        table {
+            & td {
+                height: 16px;
+                line-height: 16px;
                 width: 16px;
-                font-weight: normal;
-            }
 
-            table {
-                & td {
+                span {
+                    display: inline-block;
+                    width: 16px;
                     height: 16px;
                     line-height: 16px;
-                    width: 16px;
-
-                    span {
-                        display: inline-block;
-                        width: 16px;
-                        height: 16px;
-                        line-height: 16px;
-                    }
                 }
             }
         }

assets/modules/forms/multiple.js

@@ -0,0 +1,20 @@
+import $ from "jquery";
+import "../widget";
+
+$.widget("khq.multiple", $.khq.widget, {
+
+    _create: function () {
+        let self = this;
+
+        self.element.find('button').on("click", function (e) {
+            e.preventDefault();
+
+            let clone = $(this).closest('div').clone();
+            clone.find('input').each(function(key, value) {
+                $(value).val('');
+            });
+
+            self.element.append(clone);
+        });
+    }
+});

assets/modules/forms/multiple.scss

@@ -0,0 +1,19 @@
+.khq-multiple {
+    > div {
+        display: flex;
+        flex-direction: row;
+        margin-bottom:  $form-group-margin-bottom;
+
+        &:last-child {
+            margin-bottom: 0;
+        }
+
+        > * {
+            margin-right:  $form-group-margin-bottom;
+
+            &:last-child {
+                margin-right: 0;
+            }
+        }
+    }
+}

package.json

@@ -4,6 +4,7 @@
   "description": "Kafka GUI for topics, topics data, consumers group and more",
   "main": "index.js",
   "dependencies": {
+    "ace-builds": "^1.4.2",
     "bootstrap": "^4.2.1",
     "bytes": "^3.0.0",
     "font-awesome": "^4.7.0",

public/topic.ftl

@@ -72,6 +72,9 @@
     </div>
 </div>
 
+<@template.bottom>
+    <a href="${basePath}/${clusterId}/topic/${topic.getName()}/produce" type="submit" class="btn btn-primary">Produce to topic</a>
+</@template.bottom>
 
 <@template.footer/>
 

public/topicProduce.ftl

@@ -0,0 +1,57 @@
+<#-- @ftlvariable name="topic" type="org.kafkahq.models.Topic" -->
+
+<#import "/includes/template.ftl" as template>
+<#import "/includes/functions.ftl" as functions>
+
+<@template.header "Produce to  " + topic.getName(), "topic" />
+
+<form enctype="multipart/form-data" method="post" class="khq-form khq-form-config">
+    <div class="form-group row">
+        <label for="partition" class="col-sm-2 col-form-label">Partition</label>
+        <div class="col-sm-10">
+            <select class="form-control" name="partition" id="partition">
+                <option></option>
+                <#list topic.getPartitions() as partition>
+                    <option>${partition.getId()}</option>
+                </#list>
+            </select>
+        </div>
+    </div>
+    <div class="form-group row">
+        <label for="key" class="col-sm-2 col-form-label">Key</label>
+        <div class="col-sm-10">
+            <input type="text" class="form-control" name="key" id="key" autocomplete="off" placeholder="Key">
+        </div>
+    </div>
+    <div class="form-group row">
+        <label class="col-sm-2 col-form-label">Headers</label>
+        <div class="col-sm-10 khq-multiple">
+            <div>
+                <input type="text" class="form-control" name="headers[key][]" autocomplete="off" placeholder="Key">
+                <input type="text" class="form-control" name="headers[value][]" autocomplete="off" placeholder="Value">
+                <button class="btn btn-secondary"><i class="fa fa-plus"></i></button>
+            </div>
+        </div>
+    </div>
+    <div class="form-group row">
+        <label for="timestamp" class="col-sm-2 col-form-label">Timestamp</label>
+        <div class="col-sm-10 khq-datetime">
+            <input type="text" class="form-control datetimepicker-input" name="timestamp" id="timestamp" autocomplete="off" data-toggle="datetimepicker" data-target="#timestamp" placeholder="Timestamp"/>
+        </div>
+    </div>
+    <div class="form-group row">
+        <label for="value" class="col-sm-2 col-form-label">Value</label>
+        <div class="col-sm-10">
+            <div class="khq-ace-editor">
+                <div>></div>
+                <textarea class="form-control" name="value" id="value" placeholder="Value"></textarea>
+            </div>
+        </div>
+    </div>
+
+    <div class="khq-submit">
+        <button type="submit" class="btn btn-primary">Produce</button>
+    </div>
+</form>
+
+<@template.footer/>

src/main/java/org/kafkahq/controllers/GroupController.java

@@ -128,11 +128,11 @@ public Result offsetsStart(Request request, String cluster, String groupName, St
 
     @GET
     @Path("{groupName}/delete")
-    public Result delete(Request request, String cluster, String id) {
+    public Result delete(Request request, String cluster, String groupName) {
         this.toast(request, RequestHelper.runnableToToast(() ->
-                this.consumerGroupRepository.delete(cluster, id),
-            "Consumer group '" + id + "' is deleted",
-            "Failed to consumer group " + id
+                this.consumerGroupRepository.delete(cluster, groupName),
+            "Consumer group '" + groupName + "' is deleted",
+            "Failed to consumer group " + groupName
         ));
 
         return Results.ok();

src/main/java/org/kafkahq/controllers/TopicController.java

@@ -17,10 +17,8 @@
 import org.kafkahq.repositories.RecordRepository;
 import org.kafkahq.repositories.TopicRepository;
 
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Optional;
+import java.time.Instant;
+import java.util.*;
 import java.util.concurrent.ExecutionException;
 
 @Path("/{cluster}/topic")
@@ -85,6 +83,57 @@ public void createSubmit(Request request, Response response, String cluster) thr
         response.redirect("/" + cluster + "/topic");
     }
 
+    @GET
+    @Path("{topicName}/produce")
+    public View produce(Request request, String cluster, String topicName) throws ExecutionException, InterruptedException {
+        Topic topic = this.topicRepository.findByName(topicName);
+
+        return this.template(
+            request,
+            cluster,
+            Results
+                .html("topicProduce")
+                .put("topic", topic)
+        );
+    }
+
+    @POST
+    @Path("{topicName}/produce")
+    public void produceSubmit(Request request, Response response, String cluster, String topicName) throws Throwable {
+        List<String> headersKey = request.param("headers[key][]").toList();
+        List<String> headersValue = request.param("headers[value][]").toList();
+
+        Map<String, String> headers = new HashMap<>();
+
+        int i = 0;
+        for (String key : headersKey) {
+            if (key != null && !key.equals("") && headersValue.get(i) != null && !headersValue.get(i).equals("")) {
+                headers.put(key, headersValue.get(i));
+            }
+            i++;
+        }
+
+        this.toast(request, RequestHelper.runnableToToast(() -> {
+                this.recordRepository.produce(
+                    cluster,
+                    topicName,
+                    request.param("value").value(),
+                    headers,
+                    request.param("key").toOptional(),
+                    request.param("partition").toOptional().filter(r -> !r.equals("")).map(Integer::valueOf),
+                    request.param("timestamp")
+                        .toOptional(String.class)
+                        .filter(r -> !r.equals(""))
+                        .map(s -> Instant.parse(s).toEpochMilli())
+                );
+            },
+            "Record created",
+            "Failed to produce record"
+        ));
+
+        response.redirect(request.path());
+    }
+
     @GET
     @Path("{topicName}")
     public View home(Request request, String cluster, String topicName) throws ExecutionException, InterruptedException {
@@ -201,7 +250,6 @@ public void updateConfig(Request request, Response response, String cluster, Str
     @GET
     @Path("{topicName}/deleteRecord")
     public Result deleteRecord(Request request, Response response, String cluster, String topicName, Integer partition, String key) throws Throwable {
-
         this.toast(request, RequestHelper.runnableToToast(() -> this.recordRepository.delete(
                 cluster,
                 topicName,