Skip to content

Commit

Permalink
refactor: data save event order processing, use new method in DataSyn…
Browse files Browse the repository at this point in the history
…cer (#243)

* fix: fire DataSaveEvent before disconnect

* fix: revert rename `addSnapshot`

* docs: mention `addSnapshot` firing the API event

* refactor: use DataSyncer method for event saving, close #242

* fix: trailing semicolon
  • Loading branch information
WiIIiam278 authored Feb 11, 2024
1 parent f6773f4 commit 12e2236
Show file tree
Hide file tree
Showing 9 changed files with 136 additions and 67 deletions.
29 changes: 25 additions & 4 deletions common/src/main/java/net/william278/husksync/api/HuskSyncAPI.java
Original file line number Diff line number Diff line change
Expand Up @@ -31,11 +31,13 @@
import net.william278.husksync.user.User;
import org.jetbrains.annotations.ApiStatus;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;

import java.util.List;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.function.BiConsumer;

/**
* The common implementation of the HuskSync API, containing cross-platform API calls.
Expand Down Expand Up @@ -262,13 +264,32 @@ public void editLatestSnapshot(@NotNull User user, @NotNull ThrowingConsumer<Dat
*
* @param user The user to save the data for
* @param snapshot The snapshot to save
* @param callback A callback to run after the data has been saved (if the DataSaveEvent was not cancelled)
* @apiNote This will fire the {@link net.william278.husksync.event.DataSaveEvent} event, unless
* the save cause is {@link DataSnapshot.SaveCause#SERVER_SHUTDOWN}
* @since 3.3.2
*/
public void addSnapshot(@NotNull User user, @NotNull DataSnapshot snapshot,
@Nullable BiConsumer<User, DataSnapshot.Packed> callback) {
plugin.runAsync(() -> plugin.getDataSyncer().saveData(
user,
snapshot instanceof DataSnapshot.Unpacked unpacked
? unpacked.pack(plugin) : (DataSnapshot.Packed) snapshot,
callback
));
}

/**
* Adds a data snapshot to the database
*
* @param user The user to save the data for
* @param snapshot The snapshot to save
* @apiNote This will fire the {@link net.william278.husksync.event.DataSaveEvent} event, unless
* * the save cause is {@link DataSnapshot.SaveCause#SERVER_SHUTDOWN}
* @since 3.0
*/
public void addSnapshot(@NotNull User user, @NotNull DataSnapshot snapshot) {
plugin.runAsync(() -> plugin.getDatabase().addSnapshot(
user, snapshot instanceof DataSnapshot.Unpacked unpacked
? unpacked.pack(plugin) : (DataSnapshot.Packed) snapshot
));
this.addSnapshot(user, snapshot, null);
}

/**
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -72,8 +72,8 @@ protected void showItems(@NotNull OnlineUser viewer, @NotNull DataSnapshot.Unpac

// Creates a new snapshot with the updated enderChest
@SuppressWarnings("DuplicatedCode")
private void updateItems(@NotNull OnlineUser viewer, @NotNull Data.Items.Items items, @NotNull User user) {
final Optional<DataSnapshot.Packed> latestData = plugin.getDatabase().getLatestSnapshot(user);
private void updateItems(@NotNull OnlineUser viewer, @NotNull Data.Items.Items items, @NotNull User holder) {
final Optional<DataSnapshot.Packed> latestData = plugin.getDatabase().getLatestSnapshot(holder);
if (latestData.isEmpty()) {
plugin.getLocales().getLocale("error_no_data_to_display")
.ifPresent(viewer::sendMessage);
Expand All @@ -90,10 +90,12 @@ private void updateItems(@NotNull OnlineUser viewer, @NotNull Data.Items.Items i
);
});

// Save data
final RedisManager redis = plugin.getRedisManager();
plugin.getDatabase().addSnapshot(user, snapshot);
redis.sendUserDataUpdate(user, snapshot);
redis.getUserData(user).ifPresent(data -> redis.setUserData(user, snapshot, RedisKeyType.TTL_1_YEAR));
plugin.getDataSyncer().saveData(holder, snapshot, (user, data) -> {
redis.getUserData(user).ifPresent(d -> redis.setUserData(user, snapshot, RedisKeyType.TTL_1_YEAR));
redis.sendUserDataUpdate(user, data);
});
}

}
Original file line number Diff line number Diff line change
Expand Up @@ -72,8 +72,8 @@ protected void showItems(@NotNull OnlineUser viewer, @NotNull DataSnapshot.Unpac

// Creates a new snapshot with the updated inventory
@SuppressWarnings("DuplicatedCode")
private void updateItems(@NotNull OnlineUser viewer, @NotNull Data.Items.Items items, @NotNull User user) {
final Optional<DataSnapshot.Packed> latestData = plugin.getDatabase().getLatestSnapshot(user);
private void updateItems(@NotNull OnlineUser viewer, @NotNull Data.Items.Items items, @NotNull User holder) {
final Optional<DataSnapshot.Packed> latestData = plugin.getDatabase().getLatestSnapshot(holder);
if (latestData.isEmpty()) {
plugin.getLocales().getLocale("error_no_data_to_display")
.ifPresent(viewer::sendMessage);
Expand All @@ -90,10 +90,12 @@ private void updateItems(@NotNull OnlineUser viewer, @NotNull Data.Items.Items i
);
});

// Save data
final RedisManager redis = plugin.getRedisManager();
plugin.getDatabase().addSnapshot(user, snapshot);
redis.sendUserDataUpdate(user, snapshot);
redis.getUserData(user).ifPresent(data -> redis.setUserData(user, snapshot, RedisKeyType.TTL_1_YEAR));
plugin.getDataSyncer().saveData(holder, snapshot, (user, data) -> {
redis.getUserData(user).ifPresent(d -> redis.setUserData(user, snapshot, RedisKeyType.TTL_1_YEAR));
redis.sendUserDataUpdate(user, data);
});
}

}
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@

import net.william278.husksync.HuskSync;
import net.william278.husksync.data.DataSnapshot;
import net.william278.husksync.redis.RedisKeyType;
import net.william278.husksync.redis.RedisManager;
import net.william278.husksync.user.CommandUser;
import net.william278.husksync.user.User;
import net.william278.husksync.util.DataDumper;
Expand Down Expand Up @@ -152,11 +154,14 @@ public void execute(@NotNull CommandUser executor, @NotNull String[] args) {
);
}));

// Set the user's data and send a message
plugin.getDatabase().addSnapshot(user, data);
plugin.getRedisManager().sendUserDataUpdate(user, data);
plugin.getLocales().getLocale("data_restored", user.getUsername(), user.getUuid().toString(),
data.getShortId(), data.getId().toString()).ifPresent(executor::sendMessage);
// Save data
final RedisManager redis = plugin.getRedisManager();
plugin.getDataSyncer().saveData(user, data, (u, s) -> {
redis.getUserData(u).ifPresent(d -> redis.setUserData(u, s, RedisKeyType.TTL_1_YEAR));
redis.sendUserDataUpdate(u, s);
plugin.getLocales().getLocale("data_restored", u.getUsername(), u.getUuid().toString(),
s.getShortId(), s.getId().toString()).ifPresent(executor::sendMessage);
});
}

case "pin" -> {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -23,8 +23,6 @@
import net.william278.husksync.HuskSync;
import net.william278.husksync.config.Settings;
import net.william278.husksync.data.DataSnapshot;
import net.william278.husksync.data.DataSnapshot.SaveCause;
import net.william278.husksync.data.UserDataHolder;
import net.william278.husksync.user.User;
import org.jetbrains.annotations.Blocking;
import org.jetbrains.annotations.NotNull;
Expand All @@ -33,6 +31,7 @@
import java.nio.charset.StandardCharsets;
import java.time.OffsetDateTime;
import java.util.*;
import java.util.function.BiConsumer;

/**
* An abstract representation of the plugin database, storing player data.
Expand Down Expand Up @@ -156,42 +155,23 @@ protected final String formatStatementTables(@NotNull String sql) {
@Blocking
public abstract boolean deleteSnapshot(@NotNull User user, @NotNull UUID versionUuid);

/**
* Save user data to the database
* </p>
* This will remove the oldest data for the user if the amount of data exceeds the limit as configured
*
* @param user The user to add data for
* @param snapshot The {@link DataSnapshot} to set.
* The implementation should version it with a random UUID and the current timestamp during insertion.
* @see UserDataHolder#createSnapshot(SaveCause)
*/
@Blocking
public void addSnapshot(@NotNull User user, @NotNull DataSnapshot.Packed snapshot) {
if (snapshot.getSaveCause() != SaveCause.SERVER_SHUTDOWN) {
plugin.fireEvent(
plugin.getDataSaveEvent(user, snapshot),
(event) -> this.addAndRotateSnapshot(user, snapshot)
);
return;
}

this.addAndRotateSnapshot(user, snapshot);
}

/**
* <b>Internal</b> - Save user data to the database. This will:
* Save user data to the database, doing the following (in order):
* <ol>
* <li>Delete their most recent snapshot, if it was created before the backup frequency time</li>
* <li>Create the snapshot</li>
* <li>Rotate snapshot backups</li>
* </ol>
* This is an expensive blocking method and should be run off the main thread.
*
* @param user The user to add data for
* @param snapshot The {@link DataSnapshot} to set.
* @apiNote Prefer {@link net.william278.husksync.sync.DataSyncer#saveData(User, DataSnapshot.Packed, BiConsumer)}.
* </p>This method will not fire the {@link net.william278.husksync.event.DataSaveEvent}
*/
@Blocking
private void addAndRotateSnapshot(@NotNull User user, @NotNull DataSnapshot.Packed snapshot) {
public void addSnapshot(@NotNull User user, @NotNull DataSnapshot.Packed snapshot) {
final int backupFrequency = plugin.getSettings().getSynchronization().getSnapshotBackupFrequency();
if (!snapshot.isPinned() && backupFrequency > 0) {
this.rotateLatestSnapshot(user, snapshot.getTimestamp().minusHours(backupFrequency));
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -81,8 +81,8 @@ protected final void saveOnWorldSave(@NotNull List<OnlineUser> usersInWorld) {
}
usersInWorld.stream()
.filter(user -> !plugin.isLocked(user.getUuid()) && !user.isNpc())
.forEach(user -> plugin.getDatabase().addSnapshot(
user, user.createSnapshot(DataSnapshot.SaveCause.WORLD_SAVE)
.forEach(user -> plugin.getDataSyncer().saveData(
user, user.createSnapshot(DataSnapshot.SaveCause.WORLD_SAVE), null
));
}

Expand All @@ -101,7 +101,7 @@ protected void saveOnPlayerDeath(@NotNull OnlineUser user, @NotNull Data.Items i

final DataSnapshot.Packed snapshot = user.createSnapshot(DataSnapshot.SaveCause.DEATH);
snapshot.edit(plugin, (data -> data.getInventory().ifPresent(inventory -> inventory.setContents(items))));
plugin.getDatabase().addSnapshot(user, snapshot);
plugin.getDataSyncer().saveData(user, snapshot, (u, d) -> plugin.getRedisManager().sendUserDataUpdate(u, d));
}

/**
Expand All @@ -123,7 +123,9 @@ public final void handlePluginDisable() {
.filter(user -> !plugin.isLocked(user.getUuid()) && !user.isNpc())
.forEach(user -> {
plugin.lockPlayer(user.getUuid());
plugin.getDatabase().addSnapshot(user, user.createSnapshot(DataSnapshot.SaveCause.SERVER_SHUTDOWN));
plugin.getDataSyncer().saveData(
user, user.createSnapshot(DataSnapshot.SaveCause.SERVER_SHUTDOWN), null
);
});

// Close outstanding connections
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -22,14 +22,20 @@
import net.william278.husksync.HuskSync;
import net.william278.husksync.api.HuskSyncAPI;
import net.william278.husksync.data.DataSnapshot;
import net.william278.husksync.database.Database;
import net.william278.husksync.redis.RedisManager;
import net.william278.husksync.user.OnlineUser;
import net.william278.husksync.user.User;
import net.william278.husksync.util.Task;
import org.jetbrains.annotations.ApiStatus;
import org.jetbrains.annotations.Blocking;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;

import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;
import java.util.function.Function;
import java.util.function.Supplier;

Expand Down Expand Up @@ -87,6 +93,42 @@ public void terminate() {
*/
public abstract void saveUserData(@NotNull OnlineUser user);

/**
* Save a {@link DataSnapshot.Packed user's data snapshot} to the database,
* first firing the {@link net.william278.husksync.event.DataSaveEvent}. This will not update data on Redis.
*
* @param user the user to save the data for
* @param data the data to save
* @param after a consumer to run after data has been saved. Will be run async (off the main thread).
* @apiNote Data will not be saved if the {@link net.william278.husksync.event.DataSaveEvent} is cancelled.
* Note that this method can also edit the data before saving it.
* @implNote Note that the {@link net.william278.husksync.event.DataSaveEvent} will <b>not</b> be fired if the
* save cause is {@link DataSnapshot.SaveCause#SERVER_SHUTDOWN}.
* @since 3.3.2
*/
@Blocking
public void saveData(@NotNull User user, @NotNull DataSnapshot.Packed data,
@Nullable BiConsumer<User, DataSnapshot.Packed> after) {
if (data.getSaveCause() == DataSnapshot.SaveCause.SERVER_SHUTDOWN) {
addSnapshotToDatabase(user, data, after);
return;
}
plugin.fireEvent(
plugin.getDataSaveEvent(user, data),
(event) -> addSnapshotToDatabase(user, data, after)
);
}

// Adds a snapshot to the database and runs the after consumer
@Blocking
private void addSnapshotToDatabase(@NotNull User user, @NotNull DataSnapshot.Packed data,
@Nullable BiConsumer<User, DataSnapshot.Packed> after) {
getDatabase().addSnapshot(user, data);
if (after != null) {
after.accept(user, data);
}
}

// Calculates the max attempts the system should listen for user data for based on the latency value
private long getMaxListenAttempts() {
return BASE_LISTEN_ATTEMPTS + (
Expand All @@ -98,7 +140,7 @@ private long getMaxListenAttempts() {
// Set a user's data from the database, or set them as a new user
@ApiStatus.Internal
protected void setUserFromDatabase(@NotNull OnlineUser user) {
plugin.getDatabase().getLatestSnapshot(user).ifPresentOrElse(
getDatabase().getLatestSnapshot(user).ifPresentOrElse(
snapshot -> user.applySnapshot(snapshot, DataSnapshot.UpdateCause.SYNCHRONIZED),
() -> user.completeSync(true, DataSnapshot.UpdateCause.NEW_USER, plugin)
);
Expand Down Expand Up @@ -139,6 +181,16 @@ protected void listenForRedisData(@NotNull OnlineUser user, @NotNull Supplier<Bo
task.get().run();
}

@NotNull
protected RedisManager getRedis() {
return plugin.getRedisManager();
}

@NotNull
protected Database getDatabase() {
return plugin.getDatabase();
}

/**
* Represents the different available default modes of {@link DataSyncer}
*
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -39,15 +39,15 @@ public void setUserData(@NotNull OnlineUser user) {
plugin.runAsyncDelayed(
() -> {
// Fetch from the database if the user isn't changing servers
if (!plugin.getRedisManager().getUserServerSwitch(user)) {
if (!getRedis().getUserServerSwitch(user)) {
this.setUserFromDatabase(user);
return;
}

// Listen for the data to be updated
this.listenForRedisData(
user,
() -> plugin.getRedisManager().getUserData(user).map(data -> {
() -> getRedis().getUserData(user).map(data -> {
user.applySnapshot(data, DataSnapshot.UpdateCause.SYNCHRONIZED);
return true;
}).orElse(false)
Expand All @@ -58,12 +58,13 @@ public void setUserData(@NotNull OnlineUser user) {
}

@Override
public void saveUserData(@NotNull OnlineUser user) {
public void saveUserData(@NotNull OnlineUser onlineUser) {
plugin.runAsync(() -> {
plugin.getRedisManager().setUserServerSwitch(user);
final DataSnapshot.Packed data = user.createSnapshot(DataSnapshot.SaveCause.DISCONNECT);
plugin.getRedisManager().setUserData(user, data, RedisKeyType.TTL_10_SECONDS);
plugin.getDatabase().addSnapshot(user, data);
getRedis().setUserServerSwitch(onlineUser);
saveData(
onlineUser, onlineUser.createSnapshot(DataSnapshot.SaveCause.DISCONNECT),
(user, data) -> getRedis().setUserData(user, data, RedisKeyType.TTL_10_SECONDS)
);
});
}

Expand Down
Loading

0 comments on commit 12e2236

Please sign in to comment.