设备列表导出优化

common-lib/src/main/java/com/mortals/xhx/queue/DefaultTbQueueMsg.java

@@ -30,7 +30,7 @@ public class DefaultTbQueueMsg implements TbQueueMsg {
     /**
      * 数据载体
      */
-    private  byte[] data;
+    private  String data;
 
 
     /**
@@ -56,7 +56,7 @@ public class DefaultTbQueueMsg implements TbQueueMsg {
         // header.put(MessageHeader.CLIENTID, "abcd1234");
       //  header.put(MessageHeader.TIMESTAMP, DateUtils.getCurrStrDateTime());
         //header.put(MessageHeader.MESSAGESIGN,"ssdafasdfasdfasfd");
-        TbQueueMsg queueMsg = new DefaultTbQueueMsg(IdUtil.fastUUID(), "abcd1234".getBytes() , header);
+        TbQueueMsg queueMsg = new DefaultTbQueueMsg(IdUtil.fastUUID(), "eyJmbG93bnVtIjoiQzEwMTEifQ==" , header);
 
         String ret = JSON.toJSONString(queueMsg);
         System.out.println("pro:"+ret);

common-lib/src/main/java/com/mortals/xhx/queue/TbQueueMsg.java

@@ -14,5 +14,5 @@ public interface TbQueueMsg {
 
     TbQueueMsgHeaders getHeaders();
 
-    byte[] getData();
+    String getData();
 }

common-lib/src/main/java/com/mortals/xhx/queue/kafka/AbstractTbQueueConsumerTemplate.java

@@ -79,8 +79,10 @@ public abstract class AbstractTbQueueConsumerTemplate<R, T extends TbQueueMsg> i
                 subscribed = false;
                 partitions = subscribeQueue.poll();
             }
+            //新增新的订阅信息
             if (!subscribed) {
                 List<String> topicNames = partitions.stream().map(TopicPartitionInfo::getFullTopicName).collect(Collectors.toList());
+                //新加订阅项
                 doSubscribe(topicNames);
                 subscribed = true;
             }
@@ -97,17 +99,7 @@ public abstract class AbstractTbQueueConsumerTemplate<R, T extends TbQueueMsg> i
     }
 
     List<T> decodeRecords(List<R> records) {
-        List<T> result = new ArrayList<>(records.size());
-        records.forEach(record -> {
-            try {
-                if (record != null) {
-                    result.add(decode(record));
-                }
-            } catch (IOException e) {
-                log.error("Failed decode record: [{}]", record);
-                throw new RuntimeException("Failed to decode record: ", e);
-            }
-        });
+        List<T> result =records.stream().map(record->decode(record)).collect(Collectors.toList());
         return result;
     }
 
@@ -158,7 +150,7 @@ public abstract class AbstractTbQueueConsumerTemplate<R, T extends TbQueueMsg> i
 
     abstract protected List<R> doPoll(long durationInMillis);
 
-    abstract protected T decode(R record) throws IOException;
+    abstract protected T decode(R record);
 
     abstract protected void doSubscribe(List<String> topicNames);
 

common-lib/src/main/java/com/mortals/xhx/queue/kafka/KafkaTbQueueMsg.java

@@ -33,7 +33,7 @@ public class KafkaTbQueueMsg implements TbQueueMsg {
     }
 
     @Override
-    public byte[] getData() {
-        return data;
+    public String getData() {
+        return data.toString();
     }
 }

common-lib/src/main/java/com/mortals/xhx/queue/kafka/TbKafkaConsumerTemplate.java

@@ -77,7 +77,7 @@ public class TbKafkaConsumerTemplate<T extends TbQueueMsg> extends AbstractTbQue
     }
 
     @Override
-    public T decode(ConsumerRecord<String, byte[]> record) throws IOException {
+    public T decode(ConsumerRecord<String, byte[]> record)  {
         return decoder.decode(new KafkaTbQueueMsg(record));
     }
 

common-lib/src/main/java/com/mortals/xhx/queue/kafka/TbKafkaDecoder.java

@@ -13,6 +13,6 @@ import java.io.IOException;
  */
 public interface TbKafkaDecoder<T extends TbQueueMsg> {
 
-    T decode(TbQueueMsg msg) throws IOException;
+    T decode(TbQueueMsg msg);
 
 }

common-lib/src/main/java/com/mortals/xhx/queue/kafka/TbKafkaProducerTemplate.java

@@ -59,7 +59,7 @@ public class TbKafkaProducerTemplate<T extends TbQueueMsg> implements TbQueuePro
     @Override
     public void send(TopicPartitionInfo tpi, T msg, TbQueueCallback callback) {
         String key = msg.getKey().toString();
-        byte[] data = msg.getData();
+        byte[] data = msg.getData().getBytes();
         ProducerRecord<String, byte[]> record;
         if (tpi.getTopic() == null) {
             tpi.setTopic(this.defaultTopic);

common-lib/src/main/java/com/mortals/xhx/queue/provider/RabbitMqTbCoreQueueFactory.java

@@ -47,7 +47,7 @@ public class RabbitMqTbCoreQueueFactory implements TbCoreQueueFactory {
             }
 
             @Override
-            public byte[] getData() {
+            public String getData() {
                 return msg.getData();
             }
         });
@@ -69,7 +69,7 @@ public class RabbitMqTbCoreQueueFactory implements TbCoreQueueFactory {
                 }
 
                 @Override
-                public byte[] getData() {
+                public String getData() {
                     return msg.getData();
                 }
             });

common-lib/src/main/java/com/mortals/xhx/queue/rabbitmq/TbRabbitMqConsumerTemplate.java

@@ -84,9 +84,15 @@ public class TbRabbitMqConsumerTemplate<T extends TbQueueMsg> extends AbstractTb
 
     @Override
     protected void doSubscribe(List<String> topicNames) {
-        queues = partitions.stream()
-                .map(TopicPartitionInfo::getFullTopicName)
-                .collect(Collectors.toSet());
+
+        //新增的topkcnames
+        topicNames.stream().forEach(topic->{
+            queues.add(topic);
+        });
+
+//        queues = partitions.stream()
+//                .map(TopicPartitionInfo::getFullTopicName)
+//                .collect(Collectors.toSet());
         log.info("doSubscribe:{}", JSON.toJSONString(queues));
     }
 
@@ -125,20 +131,12 @@ public class TbRabbitMqConsumerTemplate<T extends TbQueueMsg> extends AbstractTb
     public T decode(GetResponse message) {
         try {
             DefaultTbQueueMsg msg = new DefaultTbQueueMsg();
-            // DefaultTbQueueMsg msg = gson.fromJson(new String(message.getBody()), DefaultTbQueueMsg.class);
-            if (ObjectUtils.isEmpty(message.getBody())) {
-                log.info("message is empty");
-                return null;
-            }
+            TbQueueMsgHeaders headers = new DefaultTbQueueMsgHeaders();
 
             Map<String, Object> map = JSON.parseObject(new String(message.getBody()), HashMap.class);
             msg.setKey((String) map.get("key"));
             String payloadStr = (String) map.get("data");
-            //log.info("payloadStr:{}", payloadStr);
-            //byte[] payloadDecodeByte = Base64.getDecoder().decode(payloadStr);
-
-            msg.setData(payloadStr.getBytes());
-            TbQueueMsgHeaders headers = new DefaultTbQueueMsgHeaders();
+            msg.setData(payloadStr);
             String headerStr = ((JSONObject) map.get("headers")).getString("data");
             HashMap<String, String> hashMap = JSON.parseObject(headerStr, HashMap.class);
             headers.setData(hashMap);

device-manager/src/main/java/com/mortals/xhx/module/device/web/DeviceController.java

@@ -7,6 +7,7 @@ import com.alibaba.fastjson.JSONObject;
 import com.mortals.framework.common.Rest;
 import com.mortals.framework.exception.AppException;
 import com.mortals.framework.model.Context;
+import com.mortals.framework.model.OrderCol;
 import com.mortals.framework.util.FileUtil;
 import com.mortals.framework.utils.ReflectUtils;
 import com.mortals.framework.utils.poi.ExcelUtil;
@@ -40,10 +41,7 @@ import org.springframework.web.bind.annotation.*;
 import org.springframework.web.multipart.MultipartFile;
 
 import java.io.IOException;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
@@ -91,6 +89,13 @@ public class DeviceController extends BaseCRUDJsonBodyMappingController<DeviceSe
         super.init(model, context);
     }
 
+    @Override
+    protected void doListBefore(DeviceEntity query, Map<String, Object> model, Context context) throws AppException {
+        List<OrderCol> orderColList = new ArrayList<>();
+        orderColList.add(new OrderCol("createTime",OrderCol.DESCENDING));
+        query.setOrderColList(orderColList);
+        super.doListBefore(query, model, context);
+    }
 
     /**
      * 下发信息

device-manager/src/test/java/com/mortals/httpclient/device/DeviceController.http

@@ -136,7 +136,7 @@ Content-Type: application/json
 Authorization: {{authToken}}
 
 {
-  "deviceCode": "a102",
+  "deviceCode": "B01",
   "action": "upload"
 }