Merge pull request #32 from zph95/feature/send_with_header

send message with header
2023-09-20 15:32:27 +08:00
parent 93c1d3cd9d b27db32ee2
commit 0041bafb83
7 changed files with 114 additions and 42 deletions
--- a/src/main/java/com/xuxd/kafka/console/beans/SendMessage.java
+++ b/src/main/java/com/xuxd/kafka/console/beans/SendMessage.java
@@ -1,5 +1,6 @@
 package com.xuxd.kafka.console.beans;

+import java.util.List;
 import lombok.Data;

 /**
@@ -22,4 +23,13 @@ public class SendMessage {
    private int num;

    private long offset;
+
+    private List<Header> headers;
+
+    @Data
+    public static class Header{
+        private String headerKey;
+
+        private String headerValue;
+    }
 }
--- a/src/main/java/com/xuxd/kafka/console/controller/MessageController.java
+++ b/src/main/java/com/xuxd/kafka/console/controller/MessageController.java
@@ -56,6 +56,13 @@ public class MessageController {
        return messageService.send(message);
    }

+    @PostMapping("/sendWithHeader")
+    @ControllerLog("在线发送消息")
+    @Permission("message:send")
+    public Object sendWithHeader(@RequestBody SendMessage message) {
+        return messageService.sendWithHeader(message);
+    }
+
    @ControllerLog("重新发送消息")
    @Permission("message:resend")
    @PostMapping("/resend")
--- a/src/main/java/com/xuxd/kafka/console/service/MessageService.java
+++ b/src/main/java/com/xuxd/kafka/console/service/MessageService.java
@@ -24,6 +24,8 @@ public interface MessageService {

    ResponseData send(SendMessage message);

+    ResponseData sendWithHeader(SendMessage message);
+
    ResponseData resend(SendMessage message);

    ResponseData delete(List<QueryMessage> messages);
--- a/src/main/java/com/xuxd/kafka/console/service/impl/MessageServiceImpl.java
+++ b/src/main/java/com/xuxd/kafka/console/service/impl/MessageServiceImpl.java
@@ -228,6 +228,14 @@ public class MessageServiceImpl implements MessageService, ApplicationContextAwa
        return ResponseData.create().success();
    }

+    @Override public ResponseData sendWithHeader(SendMessage message) {
+        String[] headerKeys= message.getHeaders().stream().map(SendMessage.Header::getHeaderKey).toArray(String[]::new);
+        String[] headerValues= message.getHeaders().stream().map(SendMessage.Header::getHeaderValue).toArray(String[]::new);
+        log.info("send with header:keys{},values{}",headerKeys, headerValues);
+        messageConsole.send(message.getTopic(), message.getPartition(), message.getKey(), message.getBody(), message.getNum(), headerKeys, headerValues);
+        return ResponseData.create().success();
+    }
+
    @Override public ResponseData resend(SendMessage message) {
        TopicPartition partition = new TopicPartition(message.getTopic(), message.getPartition());
        Map<TopicPartition, Object> offsetTable = new HashMap<>(1, 1.0f);
--- a/src/main/scala/kafka/console/MessageConsole.scala
+++ b/src/main/scala/kafka/console/MessageConsole.scala
@@ -8,10 +8,12 @@ import org.apache.kafka.clients.admin.{DeleteRecordsOptions, RecordsToDelete}
 import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
 import org.apache.kafka.clients.producer.ProducerRecord
 import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.header.Header
+import org.apache.kafka.common.header.internals.RecordHeader

 import java.time.Duration
 import java.util
-import java.util.{Properties}
+import java.util.Properties
 import scala.collection.immutable
 import scala.jdk.CollectionConverters.{CollectionHasAsScala, MapHasAsScala, SeqHasAsJava}

@@ -218,7 +220,7 @@ class MessageConsole(config: KafkaConfig) extends KafkaConsole(config: KafkaConf

    def send(topic: String, partition: Int, key: String, value: String, num: Int): Unit = {
        withProducerAndCatchError(producer => {
-            val nullKey = if (key != null && key.trim().length() == 0) null else key
+            val nullKey = if (key != null && key.trim().isEmpty) null else key
            for (a <- 1 to num) {
                val record = if (partition != -1) new ProducerRecord[String, String](topic, partition, nullKey, value)
                else new ProducerRecord[String, String](topic, nullKey, value)
@@ -228,6 +230,24 @@ class MessageConsole(config: KafkaConfig) extends KafkaConsole(config: KafkaConf

    }

+    def send(topic: String, partition: Int, key: String, value: String, num: Int, headerKeys: Array[String], headerValues: Array[String]): Unit = {
+        withProducerAndCatchError(producer => {
+            val nullKey = if (key != null && key.trim().isEmpty) null else key
+            for (a <- 1 to num) {
+                val record = if (partition != -1) new ProducerRecord[String, String](topic, partition, nullKey, value)
+                else new ProducerRecord[String, String](topic, nullKey, value)
+                if (!headerKeys.isEmpty && headerKeys.length == headerValues.length) {
+                    val headers: Array[Header] = headerKeys.zip(headerValues).map { case (key, value) =>
+                        new RecordHeader(key, value.getBytes())
+                    }
+                    headers.foreach(record.headers().add)
+                }
+                producer.send(record)
+            }
+        }, e => log.error("send error.", e))
+
+    }
+
    def sendSync(record: ProducerRecord[Array[Byte], Array[Byte]]): (Boolean, String) = {
        withByteProducerAndCatchError(producer => {
            val metadata = producer.send(record).get()
--- a/ui/src/utils/api.js
+++ b/ui/src/utils/api.js
@@ -288,6 +288,10 @@ export const KafkaMessageApi = {
    url: "/message/send",
    method: "post",
  },
+  sendWithHeader: {
+    url: "/message/sendWithHeader",
+    method: "post",
+  },
  resend: {
    url: "/message/resend",
    method: "post",
--- a/ui/src/views/message/SendMessage.vue
+++ b/ui/src/views/message/SendMessage.vue
@@ -15,61 +15,72 @@
              },
            ]"
            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 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="消息头">
+          <table>
+            <thead>
+              <tr>
+                <th>header key</th>
+                <th>header value</th>
+              </tr>
+            </thead>
+            <tbody>
+              <tr v-for="(row, index) in rows" :key="index">
+                <td> <a-input v-model="row.headerKey" /></td>
+                <td> <a-input v-model="row.headerValue" /></td>
+                <td><a-button type="primary" @click="deleteRow(index)">删除</a-button></td>
+              </tr>
+            </tbody>
+            <a-button type="primary" @click="addRow">添加</a-button>
+          </table>
+        </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-textarea
-            v-decorator="[
-              'body',
-              {
-                rules: [
-                  {
-                    required: true,
-                    message: '输入消息体!',
-                  },
-                ],
-              },
-            ]"
-            placeholder="输入消息体!"
-          />
+           v-decorator="[
+            'body',
+            {
+              rules: [
+                {
+                  required: true,
+                  message: '输入消息体!',
+                },
+              ],
+            },
+          ]"
+           placeholder="输入消息体!" />
        </a-form-item>
        <a-form-item label="发送的消息数">
-          <a-input-number
-            v-decorator="[
-              'num',
-              {
-                initialValue: 1,
-                rules: [
-                  {
-                    required: true,
-                    message: '输入消息数!',
-                  },
-                ],
-              },
-            ]"
-            :min="1"
-            :max="32"
-          />
+          <a-input-number v-decorator="[
+            'num',
+            {
+              initialValue: 1,
+              rules: [
+                {
+                  required: true,
+                  message: '输入消息数!',
+                },
+              ],
+            },
+          ]"
+           :min="1"
+           :max="32"
+         />
        </a-form-item>
        <a-form-item :wrapper-col="{ span: 12, offset: 5 }">
          <a-button type="primary" html-type="submit"> 提交 </a-button>
@@ -97,6 +108,7 @@ export default {
      loading: false,
      partitions: [],
      selectPartition: undefined,
+      rows: [{ headerKey: '', headerValue: '' }],
    };
  },
  methods: {
@@ -137,17 +149,26 @@ export default {
      this.selectPartition = -1;
      this.getPartitionInfo(topic);
    },
+    addRow() {
+      if(this.rows.length<32){
+        this.rows.push({ HeaderKey: '', HeaderValue: '' });
+      }
+    },
+    deleteRow(index) {
+      this.rows.splice(index, 1);
+    },
    handleSubmit(e) {
-      e.preventDefault();
+    e.preventDefault();
      this.form.validateFields((err, values) => {
        if (!err) {
          const param = Object.assign({}, values, {
            partition: this.selectPartition,
+            headers: this.rows,
          });
          this.loading = true;
          request({
-            url: KafkaMessageApi.send.url,
-            method: KafkaMessageApi.send.method,
+            url: KafkaMessageApi.sendWithHeader.url,
+            method: KafkaMessageApi.sendWithHeader.method,
            data: param,
          }).then((res) => {
            this.loading = false;