diff --git a/src/main/java/net/coreprotect/consumer/Queue.java b/src/main/java/net/coreprotect/consumer/Queue.java index 00c95740..a372a097 100755 --- a/src/main/java/net/coreprotect/consumer/Queue.java +++ b/src/main/java/net/coreprotect/consumer/Queue.java @@ -1,9 +1,11 @@ package net.coreprotect.consumer; import java.util.ArrayList; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.concurrent.ConcurrentHashMap; import org.bukkit.Bukkit; import org.bukkit.Location; @@ -103,10 +105,17 @@ public class Queue { return chestId; } - protected static synchronized int getItemId(String id) { - int chestId = ConfigHandler.loggingItem.getOrDefault(id, -1) + 1; - ConfigHandler.loggingItem.put(id, chestId); - return chestId; + // The Queue lock keeps ItemTransactionProcess.discard from clearing an item before its generation is registered, + // and compute keeps the append atomic with the consumer's take in ItemLogger + protected static synchronized int addPendingItems(ConcurrentHashMap> pendingItems, String id, ItemStack... items) { + pendingItems.compute(id, (key, list) -> { + List result = list == null ? new ArrayList<>(items.length) : list; + Collections.addAll(result, items); + return result; + }); + int itemId = ConfigHandler.loggingItem.getOrDefault(id, -1) + 1; + ConfigHandler.loggingItem.put(id, itemId); + return itemId; } private static boolean queueStandardData(Object[] data, String[] user, Object object, boolean first, long reservation) { diff --git a/src/main/java/net/coreprotect/consumer/process/ItemTransactionProcess.java b/src/main/java/net/coreprotect/consumer/process/ItemTransactionProcess.java index b11a3938..b8f5a882 100644 --- a/src/main/java/net/coreprotect/consumer/process/ItemTransactionProcess.java +++ b/src/main/java/net/coreprotect/consumer/process/ItemTransactionProcess.java @@ -26,23 +26,17 @@ class ItemTransactionProcess extends Queue { if (ConfigHandler.loggingItem.get(loggingItemId) != null) { int current_chest = ConfigHandler.loggingItem.get(loggingItemId); if (ConfigHandler.itemsPickup.get(loggingItemId) == null && ConfigHandler.itemsDrop.get(loggingItemId) == null && ConfigHandler.itemsThrown.get(loggingItemId) == null && ConfigHandler.itemsShot.get(loggingItemId) == null && ConfigHandler.itemsBreak.get(loggingItemId) == null && ConfigHandler.itemsDestroy.get(loggingItemId) == null && ConfigHandler.itemsCreate.get(loggingItemId) == null && ConfigHandler.itemsSell.get(loggingItemId) == null && ConfigHandler.itemsBuy.get(loggingItemId) == null) { + ConfigHandler.loggingItem.remove(loggingItemId, forceData); return; } if (current_chest == forceData) { int currentTime = (int) (System.currentTimeMillis() / 1000L); if (currentTime > time) { - ItemLogger.PreparedTransaction prepared = null; - if (ConfigHandler.databaseType.isColumnar()) { - prepared = ItemLogger.prepare(location, offset, user); - Consumer.consumerObjects.get(processId).put(id, prepared); - } - else { - ItemLogger.log(preparedStmt, batchCount, location, offset, user); - } - clearItemTransaction(loggingItemId); - if (prepared != null) { - prepared.log(preparedStmt, batchCount, user); - } + // prepare takes the pending lists, so a retried batch has to log from the stored copy + ItemLogger.PreparedTransaction prepared = ItemLogger.prepare(location, offset, user); + Consumer.consumerObjects.get(processId).put(id, prepared); + ConfigHandler.loggingItem.remove(loggingItemId, forceData); + prepared.log(preparedStmt, batchCount, user); } else { Queue.queueItemTransaction(user, location, time, offset, forceData); diff --git a/src/main/java/net/coreprotect/database/logger/ItemLogger.java b/src/main/java/net/coreprotect/database/logger/ItemLogger.java index e5ce4c66..c4fda982 100644 --- a/src/main/java/net/coreprotect/database/logger/ItemLogger.java +++ b/src/main/java/net/coreprotect/database/logger/ItemLogger.java @@ -59,20 +59,19 @@ public class ItemLogger { public static PreparedTransaction prepare(Location location, int offset, String user) { String key = user.toLowerCase(Locale.ROOT) + "." + location.getBlockX() + "." + location.getBlockY() + "." + location.getBlockZ(); ItemStack[][] items = { - snapshot(ConfigHandler.itemsPickup, key), snapshot(ConfigHandler.itemsDrop, key), - snapshot(ConfigHandler.itemsThrown, key), snapshot(ConfigHandler.itemsShot, key), - snapshot(ConfigHandler.itemsBreak, key), snapshot(ConfigHandler.itemsDestroy, key), - snapshot(ConfigHandler.itemsCreate, key), snapshot(ConfigHandler.itemsSell, key), snapshot(ConfigHandler.itemsBuy, key) + take(ConfigHandler.itemsPickup, key), take(ConfigHandler.itemsDrop, key), + take(ConfigHandler.itemsThrown, key), take(ConfigHandler.itemsShot, key), + take(ConfigHandler.itemsBreak, key), take(ConfigHandler.itemsDestroy, key), + take(ConfigHandler.itemsCreate, key), take(ConfigHandler.itemsSell, key), take(ConfigHandler.itemsBuy, key) }; - for (ItemStack[] group : items) { - ItemUtils.mergeItems(null, group); - } return new PreparedTransaction(location, (int) (System.currentTimeMillis() / 1000L) - offset, items); } - private static ItemStack[] snapshot(Map> source, String key) { - List values = source.get(key); - return values == null ? new ItemStack[0] : ItemUtils.getContainerState(values.toArray(new ItemStack[0])); + // Removing the whole list means an item appended after this point starts a new list for the next transaction. + // Cloning waits for PreparedTransaction.log, so the detached items are stored for a retry before anything can throw + private static ItemStack[] take(Map> source, String key) { + List values = source.remove(key); + return values == null ? new ItemStack[0] : values.toArray(new ItemStack[0]); } public static final class PreparedTransaction { @@ -92,7 +91,10 @@ public class ItemLogger { return; } for (int index = 0; index < items.length; index++) { - logTransaction(batch, batchCount, 0, user, location.clone(), ItemUtils.getContainerState(items[index]), ACTIONS[index], time); + // Clone before merging so a retry logs from the untouched detached items + ItemStack[] group = ItemUtils.getContainerState(items[index]); + ItemUtils.mergeItems(null, group); + logTransaction(batch, batchCount, 0, user, location.clone(), group, ACTIONS[index], time); } } } diff --git a/src/main/java/net/coreprotect/listener/entity/EntityPickupItemListener.java b/src/main/java/net/coreprotect/listener/entity/EntityPickupItemListener.java index 9f679690..007298fc 100644 --- a/src/main/java/net/coreprotect/listener/entity/EntityPickupItemListener.java +++ b/src/main/java/net/coreprotect/listener/entity/EntityPickupItemListener.java @@ -1,7 +1,5 @@ package net.coreprotect.listener.entity; -import java.util.ArrayList; -import java.util.List; import java.util.Locale; import org.bukkit.Location; @@ -26,11 +24,7 @@ public final class EntityPickupItemListener extends Queue implements Listener { } String loggingItemId = player.getName().toLowerCase(Locale.ROOT) + "." + location.getBlockX() + "." + location.getBlockY() + "." + location.getBlockZ(); - int itemId = getItemId(loggingItemId); - - List list = ConfigHandler.itemsPickup.getOrDefault(loggingItemId, new ArrayList<>()); - list.add(itemStack.clone()); - ConfigHandler.itemsPickup.put(loggingItemId, list); + int itemId = addPendingItems(ConfigHandler.itemsPickup, loggingItemId, itemStack.clone()); int time = (int) (System.currentTimeMillis() / 1000L) + 1; Queue.queueItemTransaction(player.getName(), location.clone(), time, 0, itemId); diff --git a/src/main/java/net/coreprotect/listener/player/CraftItemListener.java b/src/main/java/net/coreprotect/listener/player/CraftItemListener.java index b93ccf14..e65c80c2 100644 --- a/src/main/java/net/coreprotect/listener/player/CraftItemListener.java +++ b/src/main/java/net/coreprotect/listener/player/CraftItemListener.java @@ -39,27 +39,18 @@ public final class CraftItemListener extends Queue implements Listener { } String loggingItemId = user.toLowerCase(Locale.ROOT) + "." + location.getBlockX() + "." + location.getBlockY() + "." + location.getBlockZ(); - int itemId = getItemId(loggingItemId); - + int itemId; if (action == ItemLogger.ITEM_BUY) { - List list = ConfigHandler.itemsBuy.getOrDefault(loggingItemId, new ArrayList<>()); - list.add(itemStack); - ConfigHandler.itemsBuy.put(loggingItemId, list); + itemId = addPendingItems(ConfigHandler.itemsBuy, loggingItemId, itemStack); } else if (action == ItemLogger.ITEM_SELL) { - List list = ConfigHandler.itemsSell.getOrDefault(loggingItemId, new ArrayList<>()); - list.add(itemStack); - ConfigHandler.itemsSell.put(loggingItemId, list); + itemId = addPendingItems(ConfigHandler.itemsSell, loggingItemId, itemStack); } else if (action == ItemLogger.ITEM_CREATE) { - List list = ConfigHandler.itemsCreate.getOrDefault(loggingItemId, new ArrayList<>()); - list.add(itemStack); - ConfigHandler.itemsCreate.put(loggingItemId, list); + itemId = addPendingItems(ConfigHandler.itemsCreate, loggingItemId, itemStack); } else { - List list = ConfigHandler.itemsDestroy.getOrDefault(loggingItemId, new ArrayList<>()); - list.add(itemStack); - ConfigHandler.itemsDestroy.put(loggingItemId, list); + itemId = addPendingItems(ConfigHandler.itemsDestroy, loggingItemId, itemStack); } int time = (int) (System.currentTimeMillis() / 1000L) + 1; diff --git a/src/main/java/net/coreprotect/listener/player/InventoryChangeListener.java b/src/main/java/net/coreprotect/listener/player/InventoryChangeListener.java index 2649f2c2..04185816 100644 --- a/src/main/java/net/coreprotect/listener/player/InventoryChangeListener.java +++ b/src/main/java/net/coreprotect/listener/player/InventoryChangeListener.java @@ -547,18 +547,12 @@ public final class InventoryChangeListener extends Queue implements Listener { // Process the enchantment operation Location location = player.getLocation(); String loggingItemId = player.getName().toLowerCase(Locale.ROOT) + "." + location.getBlockX() + "." + location.getBlockY() + "." + location.getBlockZ(); - int itemId = getItemId(loggingItemId); // Log the input items as removed - List removedItems = new ArrayList<>(); - removedItems.add(firstItem.clone()); - removedItems.add(secondItem.clone()); - ConfigHandler.itemsDestroy.put(loggingItemId, removedItems); + addPendingItems(ConfigHandler.itemsDestroy, loggingItemId, firstItem.clone(), secondItem.clone()); // Log the output item as created - List createdItems = new ArrayList<>(); - createdItems.add(resultItem.clone()); - ConfigHandler.itemsCreate.put(loggingItemId, createdItems); + int itemId = addPendingItems(ConfigHandler.itemsCreate, loggingItemId, resultItem.clone()); int time = (int) (System.currentTimeMillis() / 1000L) + 1; Queue.queueItemTransaction(player.getName(), location.clone(), time, 0, itemId); diff --git a/src/main/java/net/coreprotect/listener/player/PlayerDropItemListener.java b/src/main/java/net/coreprotect/listener/player/PlayerDropItemListener.java index 8247934d..7b3c4955 100644 --- a/src/main/java/net/coreprotect/listener/player/PlayerDropItemListener.java +++ b/src/main/java/net/coreprotect/listener/player/PlayerDropItemListener.java @@ -1,7 +1,5 @@ package net.coreprotect.listener.player; -import java.util.ArrayList; -import java.util.List; import java.util.Locale; import java.util.Map; import java.util.UUID; @@ -41,11 +39,7 @@ public final class PlayerDropItemListener extends Queue implements Listener { } String loggingItemId = user.toLowerCase(Locale.ROOT) + "." + location.getBlockX() + "." + location.getBlockY() + "." + location.getBlockZ(); - int itemId = getItemId(loggingItemId); - - List list = ConfigHandler.itemsDrop.getOrDefault(loggingItemId, new ArrayList<>()); - list.add(itemStack.clone()); - ConfigHandler.itemsDrop.put(loggingItemId, list); + int itemId = addPendingItems(ConfigHandler.itemsDrop, loggingItemId, itemStack.clone()); int time = (int) (System.currentTimeMillis() / 1000L) + 1; Queue.queueItemTransaction(user, location.clone(), time, 0, itemId); diff --git a/src/main/java/net/coreprotect/listener/player/PlayerItemBreakListener.java b/src/main/java/net/coreprotect/listener/player/PlayerItemBreakListener.java index 45648561..04ed3144 100644 --- a/src/main/java/net/coreprotect/listener/player/PlayerItemBreakListener.java +++ b/src/main/java/net/coreprotect/listener/player/PlayerItemBreakListener.java @@ -1,7 +1,5 @@ package net.coreprotect.listener.player; -import java.util.ArrayList; -import java.util.List; import java.util.Locale; import org.bukkit.Location; @@ -23,11 +21,7 @@ public final class PlayerItemBreakListener extends Queue implements Listener { } String loggingItemId = user.toLowerCase(Locale.ROOT) + "." + location.getBlockX() + "." + location.getBlockY() + "." + location.getBlockZ(); - int itemId = getItemId(loggingItemId); - - List list = ConfigHandler.itemsBreak.getOrDefault(loggingItemId, new ArrayList<>()); - list.add(itemStack.clone()); - ConfigHandler.itemsBreak.put(loggingItemId, list); + int itemId = addPendingItems(ConfigHandler.itemsBreak, loggingItemId, itemStack.clone()); int time = (int) (System.currentTimeMillis() / 1000L) + 1; Queue.queueItemTransaction(user, location.clone(), time, 0, itemId); diff --git a/src/main/java/net/coreprotect/listener/player/ProjectileLaunchListener.java b/src/main/java/net/coreprotect/listener/player/ProjectileLaunchListener.java index 101e6977..5990a9e0 100644 --- a/src/main/java/net/coreprotect/listener/player/ProjectileLaunchListener.java +++ b/src/main/java/net/coreprotect/listener/player/ProjectileLaunchListener.java @@ -1,10 +1,8 @@ package net.coreprotect.listener.player; -import java.util.ArrayList; import java.util.Arrays; import java.util.HashSet; import java.util.Iterator; -import java.util.List; import java.util.Locale; import java.util.Map; import java.util.Map.Entry; @@ -36,22 +34,18 @@ public final class ProjectileLaunchListener extends Queue implements Listener { } String loggingItemId = user.toLowerCase(Locale.ROOT) + "." + location.getBlockX() + "." + location.getBlockY() + "." + location.getBlockZ(); - int itemId = getItemId(loggingItemId); itemStack = itemStack.clone(); if (amount > 0) { itemStack.setAmount(amount); } + int itemId; if (action == ItemLogger.ITEM_SHOOT) { - List list = ConfigHandler.itemsShot.getOrDefault(loggingItemId, new ArrayList<>()); - list.add(itemStack); - ConfigHandler.itemsShot.put(loggingItemId, list); + itemId = addPendingItems(ConfigHandler.itemsShot, loggingItemId, itemStack); } else { - List list = ConfigHandler.itemsThrown.getOrDefault(loggingItemId, new ArrayList<>()); - list.add(itemStack); - ConfigHandler.itemsThrown.put(loggingItemId, list); + itemId = addPendingItems(ConfigHandler.itemsThrown, loggingItemId, itemStack); } int time = (int) (System.currentTimeMillis() / 1000L) + delay;