Support produce message to pulsar

lib/api/pulsar/pulsar_partitioned_topic_api.dart

@@ -4,10 +4,12 @@ import 'dart:developer';
 import 'package:http/http.dart' as http;
 import 'package:paas_dashboard_flutter/api/http_util.dart';
 import 'package:paas_dashboard_flutter/api/pulsar/pulsar_stat_api.dart';
+import 'package:paas_dashboard_flutter/module/pulsar/const.dart';
 import 'package:paas_dashboard_flutter/module/pulsar/pulsar_consume.dart';
 import 'package:paas_dashboard_flutter/module/pulsar/pulsar_partitioned_topic_base.dart';
 import 'package:paas_dashboard_flutter/module/pulsar/pulsar_partitioned_topic_detail.dart';
 import 'package:paas_dashboard_flutter/module/pulsar/pulsar_produce.dart';
+import 'package:paas_dashboard_flutter/module/pulsar/pulsar_producer.dart';
 import 'package:paas_dashboard_flutter/module/pulsar/pulsar_subscription.dart';
 import 'package:paas_dashboard_flutter/module/pulsar/pulsar_topic.dart';
 import 'package:paas_dashboard_flutter/ui/util/string_util.dart';
@@ -245,4 +247,23 @@ class PulsarPartitionedTopicApi {
     return new PulsarPartitionedTopicBaseResp(
         topicName, partitionNum, msgRateIn, msgRateOut, msgInCounter, msgOutCounter, storageSize);
   }
+
+  static Future<String> sendMsgToPartitionTopic(
+      String host, int port, String tenant, String namespace, String topic, String key, String value) async {
+    ProducerMessage producerMessage = new ProducerMessage(key, value);
+    List<ProducerMessage> messageList = new List.empty(growable: true);
+    messageList.add(producerMessage);
+    PublishMessagesReq messagesReq = new PublishMessagesReq(PulsarConst.defaultProducerName, messageList);
+    var url = 'http://$host:${port.toString()}/topics/persistent/$tenant/$namespace/$topic/';
+    var response = await http.post(Uri.parse(url),
+        headers: <String, String>{
+          'Content-Type': 'application/json; charset=UTF-8',
+        },
+        body: json.encode(messagesReq));
+    if (HttpUtil.abnormal(response.statusCode)) {
+      log('ErrorCode is ${response.statusCode}, body is ${response.body}');
+      return "send msg failed, " + response.body;
+    }
+    return "send msg success";
+  }
 }

lib/api/pulsar/pulsar_topic_api.dart

@@ -5,7 +5,7 @@ import 'package:http/http.dart' as http;
 import 'package:paas_dashboard_flutter/api/http_util.dart';
 import 'package:paas_dashboard_flutter/api/pulsar/pulsar_stat_api.dart';
 import 'package:paas_dashboard_flutter/module/pulsar/pulsar_consume.dart';
-import 'package:paas_dashboard_flutter/module/pulsar/pulsar_produce.dart';
+import 'package:paas_dashboard_flutter/module/pulsar/pulsar_producer.dart';
 import 'package:paas_dashboard_flutter/module/pulsar/pulsar_subscription.dart';
 import 'package:paas_dashboard_flutter/module/pulsar/pulsar_topic.dart';
 import 'package:paas_dashboard_flutter/module/pulsar/pulsar_topic_base.dart';

lib/module/pulsar/const.dart

@@ -2,4 +2,5 @@ class PulsarConst {
   static const String defaultHost = "localhost";
   static const int defaultBrokerPort = 8080;
   static const int defaultFunctionPort = 6650;
+  static const String defaultProducerName = "flutter-dashboard-producer";
 }

lib/module/pulsar/pulsar_produce.dart

@@ -1,15 +1,26 @@
-class ProducerResp {
+class PublishMessagesReq {
   final String producerName;
-  final double rateIn;
-  final double throughputIn;
-  final String clientVersion;
-  final double averageMsgSize;
-  final String address;
+  final List<ProducerMessage> messages;
 
-  ProducerResp(
-      this.producerName, this.rateIn, this.throughputIn, this.clientVersion, this.averageMsgSize, this.address);
+  PublishMessagesReq(this.producerName, this.messages);
 
-  ProducerResp deepCopy() {
-    return new ProducerResp(producerName, rateIn, throughputIn, clientVersion, averageMsgSize, address);
+  Map toJson() {
+    Map map = new Map();
+    map["producerName"] = this.producerName;
+    map["messages"] = this.messages;
+    return map;
+  }
+}
+
+class ProducerMessage {
+  final String payload;
+  final String key;
+
+  ProducerMessage(this.key, this.payload);
+  Map toJson() {
+    Map map = new Map();
+    map["payload"] = this.payload;
+    map["key"] = this.key;
+    return map;
   }
 }

lib/module/pulsar/pulsar_producer.dart

@@ -0,0 +1,15 @@
+class ProducerResp {
+  final String producerName;
+  final double rateIn;
+  final double throughputIn;
+  final String clientVersion;
+  final double averageMsgSize;
+  final String address;
+
+  ProducerResp(
+      this.producerName, this.rateIn, this.throughputIn, this.clientVersion, this.averageMsgSize, this.address);
+
+  ProducerResp deepCopy() {
+    return new ProducerResp(producerName, rateIn, throughputIn, clientVersion, averageMsgSize, address);
+  }
+}

lib/ui/pulsar/screen/pulsar_partitioned_topic.dart

@@ -13,6 +13,9 @@ import 'package:paas_dashboard_flutter/vm/pulsar/pulsar_partitioned_topic_subscr
 import 'package:paas_dashboard_flutter/vm/pulsar/pulsar_partitioned_topic_view_model.dart';
 import 'package:provider/provider.dart';
 
+import 'package:paas_dashboard_flutter/vm/pulsar/pulsar_partitioned_topic_produce_view_model.dart';
+import 'package:paas_dashboard_flutter/ui/pulsar/widget/pulsar_partitioned_topic_produce.dart';
+
 class PulsarPartitionedTopic extends StatefulWidget {
   PulsarPartitionedTopic();
 
@@ -29,7 +32,7 @@ class _PulsarPartitionedTopicState extends State<PulsarPartitionedTopic> {
   Widget build(BuildContext context) {
     final vm = Provider.of<PulsarPartitionedTopicViewModel>(context);
     return DefaultTabController(
-      length: 5,
+      length: 6,
       child: Scaffold(
         appBar: AppBar(
           title: Text(
@@ -41,6 +44,7 @@ class _PulsarPartitionedTopicState extends State<PulsarPartitionedTopic> {
               Tab(text: S.of(context).subscription),
               Tab(text: S.of(context).consumer),
               Tab(text: S.of(context).producer),
+              Tab(text: S.of(context).produce),
             ],
           ),
         ),
@@ -71,6 +75,11 @@ class _PulsarPartitionedTopicState extends State<PulsarPartitionedTopic> {
                   vm.pulsarInstancePo, vm.tenantResp, vm.namespaceResp, vm.topicResp),
               child: PulsarPartitionedTopicProducerWidget(),
             ).build(context),
+            ChangeNotifierProvider(
+              create: (context) => PulsarPartitionedTopicProduceViewModel(
+                  vm.pulsarInstancePo, vm.tenantResp, vm.namespaceResp, vm.topicResp),
+              child: PulsarPartitionedTopicProduceWidget(),
+            ).build(context),
           ],
         ),
       ),

lib/ui/pulsar/widget/pulsar_partitioned_topic_produce.dart

@@ -0,0 +1,60 @@
+import 'package:flutter/material.dart';
+import 'package:paas_dashboard_flutter/ui/util/exception_util.dart';
+import 'package:paas_dashboard_flutter/ui/util/form_util.dart';
+import 'package:paas_dashboard_flutter/ui/util/spinner_util.dart';
+import 'package:paas_dashboard_flutter/vm/pulsar/pulsar_partitioned_topic_produce_view_model.dart';
+import 'package:provider/provider.dart';
+
+class PulsarPartitionedTopicProduceWidget extends StatefulWidget {
+  PulsarPartitionedTopicProduceWidget();
+
+  @override
+  State<StatefulWidget> createState() {
+    return new PulsarPartitionedTopicProduceWidgetState();
+  }
+}
+
+class PulsarPartitionedTopicProduceWidgetState extends State<PulsarPartitionedTopicProduceWidget> {
+  @override
+  void initState() {
+    super.initState();
+  }
+
+  @override
+  Widget build(BuildContext context) {
+    final vm = Provider.of<PulsarPartitionedTopicProduceViewModel>(context);
+    if (vm.loading) {
+      WidgetsBinding.instance!.addPostFrameCallback((timeStamp) {
+        SpinnerUtil.create();
+      });
+    }
+    ExceptionUtil.processLoadException(vm, context);
+    ExceptionUtil.processOpException(vm, context);
+
+    var produceMsgButton = createInstanceButton(context);
+    var body = ListView(
+      children: [
+        Container(
+          height: 50,
+          child: ListView(
+            scrollDirection: Axis.horizontal,
+            shrinkWrap: true,
+            children: [produceMsgButton],
+          ),
+        ),
+      ],
+    );
+    return Scaffold(body: body);
+  }
+
+  ButtonStyleButton createInstanceButton(BuildContext context) {
+    final vm = Provider.of<PulsarPartitionedTopicProduceViewModel>(context, listen: false);
+    var list = [
+      FormFieldDef('message key'),
+      FormFieldDef('message value'),
+    ];
+    return FormUtil.createButton2NoText("Send Message To Pulsar", list, context, (key, value) {
+      vm.sendMsg(key, value);
+    });
+  }
+}

lib/ui/util/form_util.dart

@@ -95,6 +95,57 @@ class FormUtil {
         child: Text('Create ' + resourceName));
   }
 
+  static ButtonStyleButton createButton2NoText(String resourceName, List<FormFieldDef> formFieldDefList,
+      BuildContext context, Function(String, String) callback) {
+    if (formFieldDefList.length != 2) {
+      throw AssertionError('args not match');
+    }
+    return createButtonNoText(resourceName, formFieldDefList, context, (list) => callback(list[0], list[1]));
+  }
+
+  static ButtonStyleButton createButtonNoText(
+      String resourceName, List<FormFieldDef> formFieldDefList, BuildContext context, Function(List<String>) callback) {
+    return TextButton(
+        onPressed: () {
+          var editControllerList = formFieldDefList.map((e) => TextEditingController()).toList();
+          List<TextFormField> formFieldsList = List.generate(
+              formFieldDefList.length,
+              (index) => TextFormField(
+                    decoration: InputDecoration(labelText: formFieldDefList[index].fieldName),
+                    controller: editControllerList[index],
+                  ));
+          showDialog(
+              context: context,
+              builder: (BuildContext context) {
+                return AlertDialog(
+                  scrollable: true,
+                  title: Text(resourceName),
+                  content: Form(
+                      child: Column(
+                    children: formFieldsList,
+                  )),
+                  actions: [
+                    ElevatedButton(
+                      child: Text(CREATE),
+                      onPressed: () {
+                        var list = editControllerList.map((e) => e.value.text).toList();
+                        callback(list);
+                        Navigator.of(context).pop();
+                      },
+                    ),
+                    ElevatedButton(
+                      child: Text(CANCEL),
+                      onPressed: () {
+                        Navigator.of(context).pop();
+                      },
+                    )
+                  ],
+                );
+              });
+        },
+        child: Text(resourceName));
+  }
+
   static ButtonStyleButton updateButton3(String resourceName, List<FormFieldDef> formFieldDefList, BuildContext context,
       Function(String, String, String) callback) {
     if (formFieldDefList.length != 3) {

lib/vm/pulsar/pulsar_partitioned_topic_produce_view_model.dart

@@ -0,0 +1,51 @@
+import 'package:paas_dashboard_flutter/api/pulsar/pulsar_partitioned_topic_api.dart';
+import 'package:paas_dashboard_flutter/module/pulsar/pulsar_namespace.dart';
+import 'package:paas_dashboard_flutter/module/pulsar/pulsar_tenant.dart';
+import 'package:paas_dashboard_flutter/module/pulsar/pulsar_topic.dart';
+import 'package:paas_dashboard_flutter/persistent/po/pulsar_instance_po.dart';
+import 'package:paas_dashboard_flutter/vm/base_load_list_view_model.dart';
+
+class PulsarPartitionedTopicProduceViewModel extends BaseLoadListViewModel {
+  final PulsarInstancePo pulsarInstancePo;
+  final TenantResp tenantResp;
+  final NamespaceResp namespaceResp;
+  final TopicResp topicResp;
+
+  PulsarPartitionedTopicProduceViewModel(this.pulsarInstancePo, this.tenantResp, this.namespaceResp, this.topicResp);
+
+  int get id {
+    return this.pulsarInstancePo.id;
+  }
+
+  String get name {
+    return this.pulsarInstancePo.name;
+  }
+
+  String get host {
+    return this.pulsarInstancePo.host;
+  }
+
+  int get port {
+    return this.pulsarInstancePo.port;
+  }
+
+  String get tenant {
+    return this.tenantResp.tenant;
+  }
+
+  String get namespace {
+    return this.namespaceResp.namespace;
+  }
+
+  String get topic {
+    return this.topicResp.topicName;
+  }
+
+  String get message {
+    return this.topicResp.topicName;
+  }
+
+  Future<String> sendMsg(key, value) {
+    return PulsarPartitionedTopicApi.sendMsgToPartitionTopic(host, port, tenant, namespace, topic, key, value);
+  }
+}

lib/vm/pulsar/pulsar_partitioned_topic_producer_view_model.dart

@@ -1,6 +1,6 @@
 import 'package:paas_dashboard_flutter/api/pulsar/pulsar_partitioned_topic_api.dart';
 import 'package:paas_dashboard_flutter/module/pulsar/pulsar_namespace.dart';
-import 'package:paas_dashboard_flutter/module/pulsar/pulsar_produce.dart';
+import 'package:paas_dashboard_flutter/module/pulsar/pulsar_producer.dart';
 import 'package:paas_dashboard_flutter/module/pulsar/pulsar_tenant.dart';
 import 'package:paas_dashboard_flutter/module/pulsar/pulsar_topic.dart';
 import 'package:paas_dashboard_flutter/persistent/po/pulsar_instance_po.dart';
@@ -13,7 +13,6 @@ class PulsarPartitionedTopicProducerViewModel extends BaseLoadListViewModel<Prod
   final TopicResp topicResp;
 
   PulsarPartitionedTopicProducerViewModel(this.pulsarInstancePo, this.tenantResp, this.namespaceResp, this.topicResp);
-
   PulsarPartitionedTopicProducerViewModel deepCopy() {
     return new PulsarPartitionedTopicProducerViewModel(
         pulsarInstancePo.deepCopy(), tenantResp.deepCopy(), namespaceResp.deepCopy(), topicResp.deepCopy());

lib/vm/pulsar/pulsar_topic_producer_view_model.dart

@@ -1,6 +1,6 @@
 import 'package:paas_dashboard_flutter/api/pulsar/pulsar_topic_api.dart';
 import 'package:paas_dashboard_flutter/module/pulsar/pulsar_namespace.dart';
-import 'package:paas_dashboard_flutter/module/pulsar/pulsar_produce.dart';
+import 'package:paas_dashboard_flutter/module/pulsar/pulsar_producer.dart';
 import 'package:paas_dashboard_flutter/module/pulsar/pulsar_tenant.dart';
 import 'package:paas_dashboard_flutter/module/pulsar/pulsar_topic.dart';
 import 'package:paas_dashboard_flutter/persistent/po/pulsar_instance_po.dart';