From 25afa5a7e6a4f42d18d65ee982faa146f9d03375 Mon Sep 17 00:00:00 2001
From: Kai Moritz <kai@juplo.de>
Date: Sat, 25 May 2024 17:10:39 +0200
Subject: [PATCH] top10: 1.1.2 - (GREEN) Fixed implementation of `Ranking`

---
 .../juplo/kafka/wordcount/top10/Ranking.java  | 65 +++++++++++++++++--
 1 file changed, 58 insertions(+), 7 deletions(-)

diff --git a/src/main/java/de/juplo/kafka/wordcount/top10/Ranking.java b/src/main/java/de/juplo/kafka/wordcount/top10/Ranking.java
index 0635384..110ee68 100644
--- a/src/main/java/de/juplo/kafka/wordcount/top10/Ranking.java
+++ b/src/main/java/de/juplo/kafka/wordcount/top10/Ranking.java
@@ -2,9 +2,7 @@ package de.juplo.kafka.wordcount.top10;
 
 import lombok.*;
 
-import java.util.Arrays;
-import java.util.LinkedList;
-import java.util.List;
+import java.util.*;
 
 
 @AllArgsConstructor(access = AccessLevel.PRIVATE)
@@ -27,13 +25,26 @@ public class Ranking
     }
 
     List<Entry> list = new LinkedList<>(Arrays.asList(entries));
+    int oldPosition = -1;
     for (int i = 0; i < list.size(); i++)
     {
-      Entry entry;
+      Entry entry = list.get(i);
 
-      entry = list.get(i);
-      if (entry.getCounter() <= newEntry.getCounter())
+      if (entry.getCounter() < newEntry.getCounter())
       {
+        if (oldPosition > -1)
+        {
+          if (list.get(oldPosition).getCounter() > newEntry.getCounter())
+          {
+            throw new IllegalArgumentException("The ranking already contains an entry with a higher counting for " + newEntry);
+          }
+          else
+          {
+            // Entry for word already exists with the same counting! Nothing changed...
+            return this;
+          }
+        }
+
         list.add(i, newEntry);
         for (int j = i+1; j < list.size(); j++)
         {
@@ -51,6 +62,44 @@ public class Ranking
         entries = list.toArray(num -> new Entry[num]);
         return this;
       }
+
+      if (entry.getWord().equals(newEntry.getWord()))
+        oldPosition = i;
+    }
+
+    if (oldPosition > -1 && list.get(oldPosition).getCounter() > newEntry.getCounter())
+    {
+      throw new IllegalArgumentException("The ranking already contains an entry with a higher counting for " + newEntry);
+    }
+
+    if (list.size() < MAX_ENTRIES)
+    {
+      list.add(newEntry);
+      entries = list.toArray(num -> new Entry[num]);
+    }
+
+    return this;
+  }
+
+  public Ranking validate() throws IllegalArgumentException
+  {
+    if (this.entries.length > MAX_ENTRIES)
+      throw new IllegalArgumentException("Invalid Ranking: a valid ranking cannot have more entries than " + MAX_ENTRIES );
+
+    Set<String> seenWords = new HashSet<>();
+    long lowesCounting = Long.MAX_VALUE;
+
+    for (int i=0; i<this.entries.length; i++)
+    {
+      Entry entry = this.entries[i];
+
+      if (seenWords.contains(entry.getWord()))
+        throw new IllegalArgumentException("Invalid Ranking: Multiple occurrences of word -> " + entry.getWord());
+      if (entry.getCounter() > lowesCounting)
+        throw new IllegalArgumentException("Invalid Ranking: Entries are not sorted correctly");
+
+      seenWords.add(entry.getWord());
+      lowesCounting = entry.getCounter();
     }
 
     return this;
@@ -58,6 +107,8 @@ public class Ranking
 
   public static Ranking of(Entry... entries)
   {
-    return new Ranking(entries);
+    Ranking ranking = new Ranking(entries);
+    ranking.validate();
+    return ranking;
   }
 }
-- 
2.20.1