From bd814d550d69e5961326b15d969e70b25cf621ef Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E8=AE=B8=E6=99=93=E4=B8=9C?= <763795151@qq.com>
Date: Fri, 17 Dec 2021 20:06:23 +0800
Subject: [PATCH] =?UTF-8?q?=E6=8C=89=E6=97=B6=E9=97=B4=E6=9F=A5=E8=AF=A2?=
 =?UTF-8?q?=E6=B6=88=E6=81=AF=E5=8F=8A=E6=97=B6=E9=87=8A=E6=94=BE=E5=86=85?=
 =?UTF-8?q?=E5=AD=98?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../scala/kafka/console/MessageConsole.scala  |  36 ++++-
 ui/src/views/message/Message.vue              |   9 +-
 ui/src/views/message/SearchByOffset.vue       |   9 +-
 ui/src/views/message/SearchByTime.vue         |   7 +-
 ui/src/views/message/SendMessage.vue          | 144 ++++++++++++++++++
 5 files changed, 191 insertions(+), 14 deletions(-)
 create mode 100644 ui/src/views/message/SendMessage.vue

diff --git a/src/main/scala/kafka/console/MessageConsole.scala b/src/main/scala/kafka/console/MessageConsole.scala
index 324f87b..0679a8a 100644
--- a/src/main/scala/kafka/console/MessageConsole.scala
+++ b/src/main/scala/kafka/console/MessageConsole.scala
@@ -8,7 +8,7 @@ import java.time.Duration
 import java.util
 import java.util.Properties
 import scala.collection.immutable
-import scala.jdk.CollectionConverters.{CollectionHasAsScala, MapHasAsScala}
+import scala.jdk.CollectionConverters.{CollectionHasAsScala, MapHasAsScala, SeqHasAsJava}
 
 /**
  * kafka-console-ui.
@@ -60,20 +60,48 @@ class MessageConsole(config: KafkaConfig) extends KafkaConsole(config: KafkaConf
                         } else {
                             for ((tp, endOff) <- endOffTable) {
                                 if (!terminate) {
-                                    val recordList = records.records(tp)
+                                    var recordList = records.records(tp)
                                     if (!recordList.isEmpty) {
                                         val first = recordList.get(0)
                                         if (first.offset() >= endOff) {
                                             arrive.remove(tp)
                                         } else {
-                                            res.addAll(recordList)
+                                            //
+                                            //                                            (String topic,
+                                            //                                                int partition,
+                                            //                                                long offset,
+                                            //                                                long timestamp,
+                                            //                                                TimestampType timestampType,
+                                            //                                                Long checksum,
+                                            //                                                int serializedKeySize,
+                                            //                                                int serializedValueSize,
+                                            //                                                K key,
+                                            //                                                V value,
+                                            //                                                Headers headers,
+                                            //                                                Optional<Integer> leaderEpoch)
+                                            val nullVList = recordList.asScala.map(record => new ConsumerRecord[Array[Byte], Array[Byte]](record.topic(),
+                                                record.partition(),
+                                                record.offset(),
+                                                record.timestamp(),
+                                                record.timestampType(),
+                                                record.checksum(),
+                                                record.serializedKeySize(),
+                                                record.serializedValueSize(),
+                                                record.key(),
+                                                null,
+                                                record.headers(),
+                                                record.leaderEpoch())).toSeq.asJava
+                                            res.addAll(nullVList)
                                             if (recordList.get(recordList.size() - 1).offset() >= endOff) {
                                                 arrive.remove(tp)
                                             }
+                                            if (recordList != null) {
+                                                recordList = null
+                                            }
                                         }
                                     }
                                     if (arrive.isEmpty) {
-                                        terminate = true;
+                                        terminate = true
                                     }
                                 }
                             }
diff --git a/ui/src/views/message/Message.vue b/ui/src/views/message/Message.vue
index 3f6875b..2ee892f 100644
--- a/ui/src/views/message/Message.vue
+++ b/ui/src/views/message/Message.vue
@@ -5,10 +5,12 @@
         <a-tab-pane key="1" tab="根据时间查询消息">
           <SearchByTime :topic-list="topicList"></SearchByTime>
         </a-tab-pane>
-        <a-tab-pane key="2" tab="根据偏移查询消息" force-render>
+        <a-tab-pane key="2" tab="根据偏移查询消息">
           <SearchByOffset :topic-list="topicList"></SearchByOffset>
         </a-tab-pane>
-        <!--        <a-tab-pane key="3" tab="消息发送"> 消息发送1 </a-tab-pane>-->
+        <a-tab-pane key="3" tab="在线发送">
+          <SendMessage :topic-list="topicList"></SendMessage>
+        </a-tab-pane>
       </a-tabs>
     </a-spin>
   </div>
@@ -20,9 +22,10 @@ import SearchByOffset from "@/views/message/SearchByOffset";
 import request from "@/utils/request";
 import { KafkaTopicApi } from "@/utils/api";
 import notification from "ant-design-vue/lib/notification";
+import SendMessage from "@/views/message/SendMessage";
 export default {
   name: "Message",
-  components: { SearchByTime, SearchByOffset },
+  components: { SearchByTime, SearchByOffset, SendMessage },
   data() {
     return {
       loading: false,
diff --git a/ui/src/views/message/SearchByOffset.vue b/ui/src/views/message/SearchByOffset.vue
index 9b36cbf..e131e61 100644
--- a/ui/src/views/message/SearchByOffset.vue
+++ b/ui/src/views/message/SearchByOffset.vue
@@ -1,7 +1,7 @@
 <template>
   <div class="tab-content">
     <a-spin :spinning="loading">
-      <div id="components-form-advanced-search">
+      <div id="search-offset-form-advanced-search">
         <a-form
           class="ant-advanced-search-form"
           :form="form"
@@ -87,7 +87,7 @@ export default {
   data() {
     return {
       loading: false,
-      form: this.$form.createForm(this, { name: "message_search_time" }),
+      form: this.$form.createForm(this, { name: "message_search_offset" }),
       partitions: [],
       selectPartition: undefined,
       rangeConfig: {
@@ -97,7 +97,8 @@ export default {
     };
   },
   methods: {
-    handleSearch() {
+    handleSearch(e) {
+      e.preventDefault();
       this.form.validateFields((err, values) => {
         if (!err) {
           const data = Object.assign({}, values, {
@@ -176,7 +177,7 @@ const defaultData = [];
   margin-bottom: 1%;
 }
 
-#components-form-advanced-search .search-result-list {
+#search-offset-form-advanced-search .search-result-list {
   margin-top: 16px;
   border: 1px dashed #e9e9e9;
   border-radius: 6px;
diff --git a/ui/src/views/message/SearchByTime.vue b/ui/src/views/message/SearchByTime.vue
index c825968..401ec6a 100644
--- a/ui/src/views/message/SearchByTime.vue
+++ b/ui/src/views/message/SearchByTime.vue
@@ -1,7 +1,7 @@
 <template>
   <div class="tab-content">
     <a-spin :spinning="loading">
-      <div id="components-form-advanced-search">
+      <div id="search-time-form-advanced-search">
         <a-form
           class="ant-advanced-search-form"
           :form="form"
@@ -100,7 +100,8 @@ export default {
     };
   },
   methods: {
-    handleSearch() {
+    handleSearch(e) {
+      e.preventDefault();
       this.form.validateFields((err, values) => {
         if (!err) {
           const data = Object.assign({}, values, {
@@ -181,7 +182,7 @@ const defaultData = { realNum: 0, maxNum: 0 };
   margin-bottom: 1%;
 }
 
-#components-form-advanced-search .search-result-list {
+#search-time-form-advanced-search .search-result-list {
   margin-top: 16px;
   border: 1px dashed #e9e9e9;
   border-radius: 6px;
diff --git a/ui/src/views/message/SendMessage.vue b/ui/src/views/message/SendMessage.vue
new file mode 100644
index 0000000..e4314b4
--- /dev/null
+++ b/ui/src/views/message/SendMessage.vue
@@ -0,0 +1,144 @@
+<template>
+  <div class="content">
+    <a-spin :spinning="loading">
+      <a-form :form="form">
+        <a-form-item label="Topic">
+          <a-select
+            class="topic-select"
+            @change="handleTopicChange"
+            show-search
+            option-filter-prop="children"
+            v-decorator="[
+              'topic',
+              {
+                rules: [{ required: true, message: '请选择一个topic!' }],
+              },
+            ]"
+            placeholder="请选择一个topic"
+          >
+            <a-select-option v-for="v in topicList" :key="v" :value="v">
+              {{ v }}
+            </a-select-option>
+          </a-select>
+        </a-form-item>
+        <a-form-item label="分区">
+          <a-select
+            class="type-select"
+            show-search
+            option-filter-prop="children"
+            v-model="selectPartition"
+            placeholder="请选择一个分区"
+          >
+            <a-select-option v-for="v in partitions" :key="v" :value="v">
+              <span v-if="v == -1">默认</span> <span v-else>{{ v }}</span>
+            </a-select-option>
+          </a-select>
+        </a-form-item>
+        <a-form-item label="消息Key">
+          <a-input v-decorator="['key', { initialValue: 'key' }]" />
+        </a-form-item>
+        <a-form-item label="消息体" has-feedback>
+          <a-input
+            v-decorator="[
+              'body',
+              {
+                rules: [
+                  {
+                    required: true,
+                    message: '输入消息体!',
+                  },
+                ],
+              },
+            ]"
+            placeholder="输入消息体!"
+          />
+        </a-form-item>
+        <a-form-item label="发送的消息数">
+          <a-input-number
+            v-decorator="[
+              'nums',
+              {
+                initialValue: 1,
+                rules: [
+                  {
+                    required: true,
+                    message: '输入消息数!',
+                  },
+                ],
+              },
+            ]"
+            :min="1"
+            :max="32"
+          />
+        </a-form-item>
+      </a-form>
+    </a-spin>
+  </div>
+</template>
+
+<script>
+import request from "@/utils/request";
+import { KafkaTopicApi } from "@/utils/api";
+import notification from "ant-design-vue/lib/notification";
+export default {
+  name: "SendMessage",
+  components: {},
+  props: {
+    topicList: {
+      type: Array,
+    },
+  },
+  data() {
+    return {
+      form: this.$form.createForm(this, { name: "message_send" }),
+      loading: false,
+      partitions: [],
+      selectPartition: undefined,
+    };
+  },
+  methods: {
+    getTopicNameList() {
+      request({
+        url: KafkaTopicApi.getTopicNameList.url,
+        method: KafkaTopicApi.getTopicNameList.method,
+      }).then((res) => {
+        if (res.code == 0) {
+          this.topicList = res.data;
+        } else {
+          notification.error({
+            message: "error",
+            description: res.msg,
+          });
+        }
+      });
+    },
+    getPartitionInfo(topic) {
+      this.loading = true;
+      request({
+        url: KafkaTopicApi.getPartitionInfo.url + "?topic=" + topic,
+        method: KafkaTopicApi.getPartitionInfo.method,
+      }).then((res) => {
+        this.loading = false;
+        if (res.code != 0) {
+          notification.error({
+            message: "error",
+            description: res.msg,
+          });
+        } else {
+          this.partitions = res.data.map((v) => v.partition);
+          this.partitions.splice(0, 0, -1);
+        }
+      });
+    },
+    handleTopicChange(topic) {
+      this.selectPartition = -1;
+      this.getPartitionInfo(topic);
+    },
+  },
+  created() {
+    this.getTopicNameList();
+  },
+};
+</script>
+
+<style scoped></style>