#193: Order messages from latest to earliest

Signed-off-by: Jonas Arnhold <[email protected]>
amosproj · Jul 8, 2023 · 9549cf7 · 9549cf7
1 parent 76c9718
commit 9549cf7
Showing 1 changed file with 8 additions and 3 deletions.
diff --git a/backend/src/main/java/de/amos/apachepulsarui/service/MessageService.java b/backend/src/main/java/de/amos/apachepulsarui/service/MessageService.java
@@ -19,6 +19,8 @@
 
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Comparator;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -48,7 +50,7 @@ private Set<MessageDto> filterBySubscription(Set<MessageDto> messageDtos, Intege
 
         return messageDtos.stream()
                 .filter(m -> messageIds.contains(m.getMessageId()))
-                .collect(Collectors.toSet());
+                .collect(Collectors.toCollection(LinkedHashSet::new));
     }
 
     private List<String> peekMessageIds(String topic, String subscription, Integer numMessages) {
@@ -67,7 +69,7 @@ private List<String> peekMessageIds(String topic, String subscription, Integer n
     private Set<MessageDto> filterByProducers(Set<MessageDto> messageDtos, List<String> producers) {
         return messageDtos.stream()
                 .filter(m -> producers.contains(m.getProducer()))
-                .collect(Collectors.toSet());
+                .collect(Collectors.toCollection(LinkedHashSet::new));
     }
 
     private Set<MessageDto> getLatestMessagesOfTopic(String topic, Integer numMessages) {
@@ -86,7 +88,10 @@ private Set<MessageDto> getLatestMessagesOfTopic(String topic, Integer numMessag
             }
             return messages.stream()
                     .map(message -> MessageDto.fromExistingMessage(message, schema))
-                    .collect(Collectors.toSet());
+                    // latest message first in set
+                    .sorted(Comparator.comparing(MessageDto::getPublishTime, Comparator.reverseOrder()))
+                    // linked to keep the order!
+                    .collect(Collectors.toCollection(LinkedHashSet::new));
         } catch (PulsarAdminException e) {
             throw new PulsarApiException(
                     "Could not examine the amount of '%d' messages for topic '%s'".formatted(numMessages, topic),