From b4309991ff7986b48fd7e802a4689b61da97fa74 Mon Sep 17 00:00:00 2001 From: Khaled Jaber Date: Thu, 3 Sep 2026 01:02:31 -0400 Subject: [PATCH 1/4] Add local cache support for replay sessions --- .../cache/store/LocalReplayDiskCacheStore.kt | 99 +++++++++++++++++++ .../store/LocalReplayDiskCacheStoreTest.kt | 32 ++++++ .../kotlin/net/rsprox/proxy/ProxyService.kt | 23 ++++- .../proxy/binary/BinaryHeaderExtensions.kt | 3 + .../proxy/server/ServerGameLoginDecoder.kt | 3 +- 5 files changed, 155 insertions(+), 5 deletions(-) create mode 100644 cache/src/main/kotlin/net/rsprox/cache/store/LocalReplayDiskCacheStore.kt create mode 100644 cache/src/test/kotlin/net/rsprox/cache/store/LocalReplayDiskCacheStoreTest.kt create mode 100644 proxy/src/main/kotlin/net/rsprox/proxy/binary/BinaryHeaderExtensions.kt diff --git a/cache/src/main/kotlin/net/rsprox/cache/store/LocalReplayDiskCacheStore.kt b/cache/src/main/kotlin/net/rsprox/cache/store/LocalReplayDiskCacheStore.kt new file mode 100644 index 000000000..db0fa9b63 --- /dev/null +++ b/cache/src/main/kotlin/net/rsprox/cache/store/LocalReplayDiskCacheStore.kt @@ -0,0 +1,99 @@ +package net.rsprox.cache.store + +import com.github.michaelbull.logging.InlineLogger +import io.netty.buffer.ByteBuf +import net.rsprox.cache.Js5MasterIndex +import org.openrs2.cache.DiskStore +import org.openrs2.cache.Store +import org.openrs2.cache.VersionTrailer +import java.nio.file.Files +import java.nio.file.LinkOption +import java.nio.file.Path + +public enum class ReplayCacheMatch { + MATCH, + MISMATCH, + UNVERIFIABLE, +} + +public class LocalReplayDiskCacheStore( + private val path: Path, +) : ReplayDiskCacheStore { + private val lock: Any = Any() + private var store: Store? = null + + override fun get( + archive: Int, + group: Int, + ): ByteBuf? { + return synchronized(lock) { + read(openStore(), archive, group) + } + } + + override fun open() { + synchronized(lock) { + openStore() + } + } + + public fun match(masterIndex: Js5MasterIndex): ReplayCacheMatch { + val buffer = get(Store.ARCHIVESET, Store.ARCHIVESET) ?: return ReplayCacheMatch.UNVERIFIABLE + try { + val actual = ByteArray(buffer.readableBytes()) + buffer.getBytes(buffer.readerIndex(), actual) + return if (masterIndex.data contentEquals actual) { + ReplayCacheMatch.MATCH + } else { + ReplayCacheMatch.MISMATCH + } + } finally { + buffer.release() + } + } + + override fun close() { + synchronized(lock) { + store?.close() + store = null + } + } + + private fun openStore(): Store { + val existing = store + if (existing != null) { + return existing + } + require(Files.isDirectory(path, LinkOption.NOFOLLOW_LINKS)) { + "Path does not point to a cache directory: $path" + } + require(Files.isRegularFile(path.resolve(DATA_FILE), LinkOption.NOFOLLOW_LINKS)) { + "Cache directory does not contain $DATA_FILE: $path" + } + val opened = DiskStore.open(path) + store = opened + return opened + } + + private fun read( + store: Store, + archive: Int, + group: Int, + ): ByteBuf? { + return try { + val buffer = store.read(archive, group) + if (archive != Store.ARCHIVESET) { + VersionTrailer.strip(buffer) + } + buffer + } catch (error: Exception) { + logger.debug(error) { "Unable to read local replay cache group $archive:$group from $path" } + null + } + } + + private companion object { + private const val DATA_FILE: String = "main_file_cache.dat2" + private val logger = InlineLogger() + } +} diff --git a/cache/src/test/kotlin/net/rsprox/cache/store/LocalReplayDiskCacheStoreTest.kt b/cache/src/test/kotlin/net/rsprox/cache/store/LocalReplayDiskCacheStoreTest.kt new file mode 100644 index 000000000..2d91c9fa5 --- /dev/null +++ b/cache/src/test/kotlin/net/rsprox/cache/store/LocalReplayDiskCacheStoreTest.kt @@ -0,0 +1,32 @@ +package net.rsprox.cache.store + +import io.netty.buffer.Unpooled +import net.rsprox.cache.Js5MasterIndex +import org.openrs2.cache.DiskStore +import org.openrs2.cache.Store +import java.nio.file.Files +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith + +class LocalReplayDiskCacheStoreTest { + @Test + fun `opens disk cache and compares its master index`() { + val cacheDirectory = Files.createTempDirectory("rsprox-local-replay-cache") + val expected = byteArrayOf(1, 2, 3, 4) + DiskStore.create(cacheDirectory).use { store -> + store.write(Store.ARCHIVESET, Store.ARCHIVESET, Unpooled.wrappedBuffer(expected)) + } + + LocalReplayDiskCacheStore(cacheDirectory).use { store -> + store.open() + assertEquals(ReplayCacheMatch.MATCH, store.match(Js5MasterIndex(240, expected))) + assertEquals(ReplayCacheMatch.MISMATCH, store.match(Js5MasterIndex(240, byteArrayOf(9)))) + } + + val invalidDirectory = Files.createTempDirectory("rsprox-invalid-replay-cache") + assertFailsWith { + LocalReplayDiskCacheStore(invalidDirectory).open() + } + } +} diff --git a/proxy/src/main/kotlin/net/rsprox/proxy/ProxyService.kt b/proxy/src/main/kotlin/net/rsprox/proxy/ProxyService.kt index e4f4ad6a0..d073d3bcc 100644 --- a/proxy/src/main/kotlin/net/rsprox/proxy/ProxyService.kt +++ b/proxy/src/main/kotlin/net/rsprox/proxy/ProxyService.kt @@ -16,6 +16,7 @@ import net.rsprox.patch.native.NativePatcher import net.rsprox.proxy.accounts.DefaultJagexAccountStore import net.rsprox.proxy.binary.BinaryBlob import net.rsprox.proxy.binary.BinaryHeader +import net.rsprox.proxy.binary.isOldSchoolRuneScape import net.rsprox.proxy.binary.credentials.BinaryCredentials import net.rsprox.proxy.binary.credentials.BinaryCredentialsStore import net.rsprox.proxy.bootstrap.BootstrapFactory @@ -716,7 +717,17 @@ public class ProxyService( ) } - public fun loadReplaySession(path: Path): ReplaySession { + public fun loadReplaySession(path: Path): ReplaySession = + checkNotNull( + loadReplaySession(path) { + error("A local disk cache is required for this replay.") + }, + ) + + public fun loadReplaySession( + path: Path, + manualCacheSelector: (Js5MasterIndex) -> ReplayDiskCacheStore?, + ): ReplaySession? { val binary = BinaryBlob.decode(path, filterSetStore, settingsStore) val masterIndex = Js5MasterIndex.trimmed( @@ -724,9 +735,13 @@ public class ProxyService( binary.header.js5MasterIndex, ) val cacheStore = - checkNotNull(ReplayDiskCacheProvider().get(masterIndex)) { - "Unable to locate RSProx Archive or OpenRS2 disk cache for replay revision " + - "${binary.header.revision}" + if (binary.header.isOldSchoolRuneScape()) { + checkNotNull(ReplayDiskCacheProvider().get(masterIndex)) { + "Unable to locate RSProx Archive or OpenRS2 disk cache for replay revision " + + "${binary.header.revision}" + } + } else { + manualCacheSelector(masterIndex) ?: return null } decoderLoader.load(ReplayCacheProvider, binary.header.revision) val decoder = decoderLoader.getDecoder(binary.header.revision, ReplayCacheProvider) diff --git a/proxy/src/main/kotlin/net/rsprox/proxy/binary/BinaryHeaderExtensions.kt b/proxy/src/main/kotlin/net/rsprox/proxy/binary/BinaryHeaderExtensions.kt new file mode 100644 index 000000000..0ba8118c4 --- /dev/null +++ b/proxy/src/main/kotlin/net/rsprox/proxy/binary/BinaryHeaderExtensions.kt @@ -0,0 +1,3 @@ +package net.rsprox.proxy.binary + +internal fun BinaryHeader.isOldSchoolRuneScape(): Boolean = worldHost.endsWith(".runescape.com") diff --git a/proxy/src/main/kotlin/net/rsprox/proxy/server/ServerGameLoginDecoder.kt b/proxy/src/main/kotlin/net/rsprox/proxy/server/ServerGameLoginDecoder.kt index d4ea7d24c..d5df26474 100644 --- a/proxy/src/main/kotlin/net/rsprox/proxy/server/ServerGameLoginDecoder.kt +++ b/proxy/src/main/kotlin/net/rsprox/proxy/server/ServerGameLoginDecoder.kt @@ -15,6 +15,7 @@ import net.rsprox.cache.api.CacheProvider import net.rsprox.proxy.attributes.BINARY_BLOB import net.rsprox.proxy.attributes.BINARY_HEADER_BUILDER import net.rsprox.proxy.binary.BinaryBlob +import net.rsprox.proxy.binary.isOldSchoolRuneScape import net.rsprox.proxy.binary.BinaryStream import net.rsprox.proxy.channel.getAndDropEncodeSeed import net.rsprox.proxy.channel.getBinaryHeaderBuilder @@ -341,7 +342,7 @@ public class ServerGameLoginDecoder( sessionMonitor, filters, settings, - header.worldHost.endsWith(".runescape.com"), + header.isOldSchoolRuneScape(), ) blob.setServerChannel(ctx.channel()) if (LATEST_SUPPORTED_PLUGIN >= target.revisionNum(clientChannel)) { From 4788f595c8993f104e6366950167c72de5681bb5 Mon Sep 17 00:00:00 2001 From: Khaled Jaber Date: Thu, 3 Sep 2026 01:02:35 -0400 Subject: [PATCH 2/4] Track recent replay cache directories --- .../net/rsprox/gui/RecentReplayCaches.kt | 63 +++++++++++++++++++ .../net/rsprox/gui/RecentReplayCachesTest.kt | 34 ++++++++++ 2 files changed, 97 insertions(+) create mode 100644 gui/proxy-tool/src/main/kotlin/net/rsprox/gui/RecentReplayCaches.kt create mode 100644 gui/proxy-tool/src/test/kotlin/net/rsprox/gui/RecentReplayCachesTest.kt diff --git a/gui/proxy-tool/src/main/kotlin/net/rsprox/gui/RecentReplayCaches.kt b/gui/proxy-tool/src/main/kotlin/net/rsprox/gui/RecentReplayCaches.kt new file mode 100644 index 000000000..e16acf13f --- /dev/null +++ b/gui/proxy-tool/src/main/kotlin/net/rsprox/gui/RecentReplayCaches.kt @@ -0,0 +1,63 @@ +package net.rsprox.gui + +import java.nio.file.Files +import java.nio.file.LinkOption +import java.nio.file.Path +import java.util.prefs.Preferences + +internal class RecentReplayCaches( + private val preferences: Preferences = + Preferences.userNodeForPackage(App::class.java).node("recentReplayCaches"), +) { + fun record(path: Path) { + val normalized = normalize(path) + save( + buildList { + add(normalized) + addAll(load().filterNot { it == normalized }) + }.take(MAX_RECENTS), + ) + } + + fun remove(path: Path) { + val normalized = normalize(path) + save(load().filterNot { it == normalized }) + } + + fun list(): List { + val valid = load().filter(::isCacheDirectory).take(MAX_RECENTS) + save(valid) + return valid + } + + private fun load(): List { + return preferences + .get(RECENTS_KEY, "") + .lineSequence() + .map(String::trim) + .filter(String::isNotEmpty) + .map(Path::of) + .map(::normalize) + .distinct() + .toList() + } + + private fun save(paths: List) { + preferences.put(RECENTS_KEY, paths.joinToString("\n")) + runCatching { preferences.flush() } + } + + private fun normalize(path: Path): Path = + runCatching { path.toRealPath() } + .getOrElse { path.toAbsolutePath().normalize() } + + private fun isCacheDirectory(path: Path): Boolean = + Files.isDirectory(path, LinkOption.NOFOLLOW_LINKS) && + Files.isRegularFile(path.resolve(DATA_FILE), LinkOption.NOFOLLOW_LINKS) + + private companion object { + private const val RECENTS_KEY: String = "paths" + private const val MAX_RECENTS: Int = 5 + private const val DATA_FILE: String = "main_file_cache.dat2" + } +} diff --git a/gui/proxy-tool/src/test/kotlin/net/rsprox/gui/RecentReplayCachesTest.kt b/gui/proxy-tool/src/test/kotlin/net/rsprox/gui/RecentReplayCachesTest.kt new file mode 100644 index 000000000..620c8c376 --- /dev/null +++ b/gui/proxy-tool/src/test/kotlin/net/rsprox/gui/RecentReplayCachesTest.kt @@ -0,0 +1,34 @@ +package net.rsprox.gui + +import java.nio.file.Files +import java.util.UUID +import java.util.prefs.Preferences +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +class RecentReplayCachesTest { + @Test + fun `keeps five unique cache directories in most-recent order`() { + val preferences = Preferences.userRoot().node("rsprox-test/${UUID.randomUUID()}") + try { + val store = RecentReplayCaches(preferences) + val paths = + List(6) { + Files.createTempDirectory("rsprox-recent-cache-$it").also { path -> + Files.createFile(path.resolve("main_file_cache.dat2")) + } + } + + paths.forEach(store::record) + store.record(paths[2]) + + val expected = listOf(paths[2], paths[5], paths[4], paths[3], paths[1]) + val actual = store.list() + assertEquals(5, actual.size) + assertTrue(expected.zip(actual).all { (left, right) -> Files.isSameFile(left, right) }) + } finally { + preferences.removeNode() + } + } +} From 49325e873f9cb8b3f300422bdecafab2b693d4e1 Mon Sep 17 00:00:00 2001 From: Khaled Jaber Date: Thu, 3 Sep 2026 01:02:40 -0400 Subject: [PATCH 3/4] Prompt for caches when loading replays --- .../main/kotlin/net/rsprox/gui/ReplayPanel.kt | 154 +++++++++++++++++- 1 file changed, 152 insertions(+), 2 deletions(-) diff --git a/gui/proxy-tool/src/main/kotlin/net/rsprox/gui/ReplayPanel.kt b/gui/proxy-tool/src/main/kotlin/net/rsprox/gui/ReplayPanel.kt index 81a4e2242..46d4d4ab0 100644 --- a/gui/proxy-tool/src/main/kotlin/net/rsprox/gui/ReplayPanel.kt +++ b/gui/proxy-tool/src/main/kotlin/net/rsprox/gui/ReplayPanel.kt @@ -11,6 +11,10 @@ import com.formdev.flatlaf.util.ColorFunctions import com.formdev.flatlaf.util.SystemFileChooser import com.github.michaelbull.logging.InlineLogger import net.miginfocom.swing.MigLayout +import net.rsprox.cache.Js5MasterIndex +import net.rsprox.cache.store.LocalReplayDiskCacheStore +import net.rsprox.cache.store.ReplayCacheMatch +import net.rsprox.cache.store.ReplayDiskCacheStore import net.rsprox.gui.sessions.SessionType import net.rsprox.proxy.binary.BinaryHeader import net.rsprox.proxy.config.BINARY_PATH @@ -41,6 +45,7 @@ import java.nio.file.Path import java.text.DateFormat import java.util.Date import java.util.concurrent.ForkJoinPool +import java.util.concurrent.FutureTask import javax.swing.BorderFactory import javax.swing.DefaultComboBoxModel import javax.swing.JOptionPane @@ -61,6 +66,7 @@ public class ReplayPanel( ) : JPanel() { private var replaySession: ReplaySession? = null private var selectedPath: Path? = null + private val recentReplayCaches = RecentReplayCaches() private val titleLabel = FlatLabel() private val detailsLabel = FlatLabel() private val statusLabel = FlatLabel() @@ -178,8 +184,14 @@ public class ReplayPanel( ForkJoinPool.commonPool().submit { var session: ReplaySession? = null try { - session = App.service.loadReplaySession(path) - val transcript = App.service.transcribeReplaySession(session) + val loadedSession = + App.service.loadReplaySession(path, ::selectManualReplayCache) + ?: run { + finishCancelledReplayLoad(generation) + return@submit + } + session = loadedSession + val transcript = App.service.transcribeReplaySession(loadedSession) SwingUtilities.invokeLater { if (generation != loadGeneration) { session.close() @@ -221,6 +233,143 @@ public class ReplayPanel( } } + private fun selectManualReplayCache(masterIndex: Js5MasterIndex): ReplayDiskCacheStore? { + while (true) { + val recents = recentReplayCaches.list() + val path = + onEventDispatchThread { + chooseManualReplayCache(masterIndex.revision, recents) + } ?: return null + val store = LocalReplayDiskCacheStore(path) + try { + store.open() + } catch (error: Exception) { + store.close() + recentReplayCaches.remove(path) + onEventDispatchThread { + JOptionPane.showMessageDialog( + this, + error.message ?: "Unable to open the selected cache directory.", + "Invalid Replay Cache", + JOptionPane.ERROR_MESSAGE, + ) + } + continue + } + + when (val match = store.match(masterIndex)) { + ReplayCacheMatch.MATCH -> { + recentReplayCaches.record(path) + return store + } + else -> { + when (confirmManualReplayCache(match, path)) { + true -> { + recentReplayCaches.record(path) + return store + } + false -> store.close() + null -> { + store.close() + return null + } + } + } + } + } + } + + private fun confirmManualReplayCache( + match: ReplayCacheMatch, + path: Path, + ): Boolean? { + return onEventDispatchThread { + val detail = + if (match == ReplayCacheMatch.MISMATCH) "does not match" + else "could not be verified against" + val options = arrayOf("Use Anyway", "Choose Another", "Cancel") + when ( + JOptionPane.showOptionDialog( + this, + "The cache at:\n$path\n\n$detail the master index embedded in this replay.", + "Replay Cache Mismatch", + JOptionPane.DEFAULT_OPTION, + JOptionPane.WARNING_MESSAGE, + null, + options, + options[1], + ) + ) { + 0 -> true + 1 -> false + else -> null + } + } + } + + private fun chooseManualReplayCache( + revision: Int, + recents: List, + ): Path? { + if (recents.isEmpty()) { + return browseForManualReplayCache(revision) + } + val browse = "Browse..." + val choices = (recents + browse).toTypedArray() + val selected = + JOptionPane.showInputDialog( + this, + "Select the disk cache used to record this replay.", + "Select Replay Cache (Revision $revision)", + JOptionPane.QUESTION_MESSAGE, + null, + choices, + choices.first(), + ) ?: return null + return if (selected == browse) { + browseForManualReplayCache(revision) + } else { + selected as Path + } + } + + private fun browseForManualReplayCache(revision: Int): Path? { + val chooser = + SystemFileChooser().apply { + stateStoreID = REPLAY_CACHE_FILE_CHOOSER_STATE_ID + dialogTitle = "Select Replay Cache (Revision $revision)" + fileSelectionMode = SystemFileChooser.DIRECTORIES_ONLY + isAcceptAllFileFilterUsed = false + } + if (chooser.showOpenDialog(this) != SystemFileChooser.APPROVE_OPTION) { + return null + } + return chooser.selectedFile?.toPath() + } + + private fun finishCancelledReplayLoad(generation: Int) { + SwingUtilities.invokeLater { + if (generation != loadGeneration) { + return@invokeLater + } + loadingReplay = false + launchingReplayClient = false + loadingProgress.isVisible = false + selectedPath = null + updateEmptyState() + refreshControls() + } + } + + private fun onEventDispatchThread(action: () -> T): T { + if (SwingUtilities.isEventDispatchThread()) { + return action() + } + val task = FutureTask { action() } + SwingUtilities.invokeAndWait(task) + return task.get() + } + private fun updateInitialWindowMode( windowMode: Int, gameRevision: Int, @@ -1202,6 +1351,7 @@ public class ReplayPanel( private const val GAME_TICK_MILLIS = 600 private const val DEFAULT_REPLAY_BINARY_FOLDER = "Old School RuneScape" private const val REPLAY_FILE_CHOOSER_STATE_ID = "replayFile" + private const val REPLAY_CACHE_FILE_CHOOSER_STATE_ID = "replayCacheDirectory" private const val SKIP_TICKS = 10 } From 4851058dc957cd6716711cf3fa571bb888b35acd Mon Sep 17 00:00:00 2001 From: Khaled Jaber Date: Thu, 3 Sep 2026 01:25:35 -0400 Subject: [PATCH 4/4] Stream replay events to RuneLite plugins --- .../kotlin/net/rsprox/proxy/ProxyService.kt | 19 ++-- .../net/rsprox/proxy/replay/ReplaySession.kt | 29 +++++- .../rsprox/proxy/replay/ReplayTranscriber.kt | 7 +- .../rsprox/proxy/unix/UnixSocketConnection.kt | 89 +++++++++---------- 4 files changed, 85 insertions(+), 59 deletions(-) diff --git a/proxy/src/main/kotlin/net/rsprox/proxy/ProxyService.kt b/proxy/src/main/kotlin/net/rsprox/proxy/ProxyService.kt index d073d3bcc..569c6fe18 100644 --- a/proxy/src/main/kotlin/net/rsprox/proxy/ProxyService.kt +++ b/proxy/src/main/kotlin/net/rsprox/proxy/ProxyService.kt @@ -770,6 +770,7 @@ public class ProxyService( try { (replaySession.cacheStore as? ReplayDiskCacheStore)?.open() val target = initializeReplayHttpServer(port, replaySession) + initializeUnixSocketConnection(target.httpPort)?.let(replaySession::attachUnixSocketConnection) launchReplayServer(replaySession, port) // Clear out existing trackers ClientTypeDictionary.remove(port) @@ -837,12 +838,7 @@ public class ProxyService( sessionId, ) target.load(properties, bootstrapFactory) - val establishConnection = properties.getPropertyOrNull(RUNELITE_RSPROX_CONNECTION) == true - if (establishConnection) { - val connection = initializeUnixSocketListener(target.httpPort) - connection.start() - connections.addUnixConnection(target.httpPort, connection) - } + initializeUnixSocketConnection(target.httpPort) return target } @@ -881,6 +877,17 @@ public class ProxyService( return UnixSocketConnection(port) } + private fun initializeUnixSocketConnection(port: Int): UnixSocketConnection? { + if (properties.getPropertyOrNull(RUNELITE_RSPROX_CONNECTION) != true) { + return null + } + return connections.getUnixConnectionOrNull(port) + ?: initializeUnixSocketListener(port).also { connection -> + connection.start() + connections.addUnixConnection(port, connection) + } + } + public fun launchNativeClient( sessionMonitor: SessionMonitor, character: JagexCharacter?, diff --git a/proxy/src/main/kotlin/net/rsprox/proxy/replay/ReplaySession.kt b/proxy/src/main/kotlin/net/rsprox/proxy/replay/ReplaySession.kt index 83af1110a..03e8b5226 100644 --- a/proxy/src/main/kotlin/net/rsprox/proxy/replay/ReplaySession.kt +++ b/proxy/src/main/kotlin/net/rsprox/proxy/replay/ReplaySession.kt @@ -3,9 +3,11 @@ package net.rsprox.proxy.replay import com.github.michaelbull.logging.InlineLogger import io.netty.channel.Channel import net.rsprot.protocol.Prot +import net.rsprot.protocol.message.IncomingMessage import net.rsprox.cache.store.GroupStore import net.rsprox.proxy.channel.getServerToClientStreamCipher import net.rsprox.proxy.plugin.RevisionDecoder +import net.rsprox.proxy.unix.UnixSocketConnection import net.rsprox.shared.StreamDirection import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.atomic.AtomicReference @@ -17,6 +19,8 @@ public class ReplaySession( private val scheduler: ReplayScheduler = ExecutorReplayScheduler(), ) : AutoCloseable { private val clientChannel: AtomicReference = AtomicReference(null) + private val unixSocketConnection: AtomicReference = AtomicReference(null) + private val decodedMessages: Array?> = arrayOfNulls(timeline.frames.size) private val clientLaunchStarted: AtomicBoolean = AtomicBoolean(false) private val clientDisconnected: AtomicBoolean = AtomicBoolean(false) private val rebuildInProgress: AtomicBoolean = AtomicBoolean(false) @@ -72,6 +76,18 @@ public class ReplaySession( clientDisconnected.set(false) } + internal fun attachUnixSocketConnection(connection: UnixSocketConnection) { + connection.reset() + unixSocketConnection.set(connection) + } + + internal fun cacheDecodedMessages( + frame: ReplayFrame, + messages: List, + ) { + decodedMessages[frame.index] = messages + } + public fun handleLaunchedClientProcessExit() { if (hasActiveClient()) { return @@ -263,15 +279,16 @@ public class ReplaySession( } private fun sendReplayFrame(frame: ReplayFrame) { - if (frame.direction != StreamDirection.SERVER_TO_CLIENT) { - return - } val channel = clientChannel.get() ?: return if (!channel.isActive) { return } + pushDecodedMessages(frame) + if (frame.direction != StreamDirection.SERVER_TO_CLIENT) { + return + } val encoded = ReplayPacketEncoder.encode(channel.alloc(), channel.getServerToClientStreamCipher(), frame) channel.writeAndFlush(encoded).addListener { future -> if (!future.isSuccess) { @@ -289,6 +306,7 @@ public class ReplaySession( if (!channel.isActive) { return } + frames.forEach(::pushDecodedMessages) val cipher = channel.getServerToClientStreamCipher() val allocator = channel.alloc() var groupBytes = 0 @@ -369,6 +387,11 @@ public class ReplaySession( flushGroup() } + private fun pushDecodedMessages(frame: ReplayFrame) { + val connection = unixSocketConnection.get() ?: return + decodedMessages[frame.index]?.forEach(connection::push) + } + private fun packetGroupStartPayload(length: Int): ByteArray { require(length in 0..Short.MAX_VALUE) { "Packet group length out of range: $length" diff --git a/proxy/src/main/kotlin/net/rsprox/proxy/replay/ReplayTranscriber.kt b/proxy/src/main/kotlin/net/rsprox/proxy/replay/ReplayTranscriber.kt index 2c01c9c34..d2ebf5769 100644 --- a/proxy/src/main/kotlin/net/rsprox/proxy/replay/ReplayTranscriber.kt +++ b/proxy/src/main/kotlin/net/rsprox/proxy/replay/ReplayTranscriber.kt @@ -94,9 +94,10 @@ public class ReplayTranscriber( try { val packets = decodeReplayFrame(frame, revisionDecoder, protocolSession) - .map { Packet(it.direction, it.prot, it.message) } - packetList += packets - if (packets.any { it.prot.toString() == "SERVER_TICK_END" }) { + session.cacheDecodedMessages(frame, packets.map { it.message }) + val transcriberPackets = packets.map { Packet(it.direction, it.prot, it.message) } + packetList += transcriberPackets + if (transcriberPackets.any { it.prot.toString() == "SERVER_TICK_END" }) { executeRunner(runner, tracker, packetList, session.timeline.header.revision) packetList.clear() } diff --git a/proxy/src/main/kotlin/net/rsprox/proxy/unix/UnixSocketConnection.kt b/proxy/src/main/kotlin/net/rsprox/proxy/unix/UnixSocketConnection.kt index 3afb986bb..36f11b02a 100644 --- a/proxy/src/main/kotlin/net/rsprox/proxy/unix/UnixSocketConnection.kt +++ b/proxy/src/main/kotlin/net/rsprox/proxy/unix/UnixSocketConnection.kt @@ -19,6 +19,7 @@ import java.nio.file.FileSystems import java.nio.file.Path import java.nio.file.StandardWatchEventKinds import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.LinkedBlockingQueue import java.util.concurrent.atomic.AtomicInteger import java.util.concurrent.locks.ReentrantLock import kotlin.concurrent.thread @@ -114,6 +115,7 @@ public class UnixSocketConnection( conFile: Path, socketFile: Path, ) { + socketFile.deleteIfExists() val address = AFUNIXSocketAddress.of(socketFile) val server = AFUNIXServerSocket.newInstance() @@ -122,35 +124,27 @@ public class UnixSocketConnection( logger.debug { "Waiting for RSProx-Connection to establish a connection to $socketFile." } thread(start = true, isDaemon = true) { server.use { srv -> - val socket = srv.accept() - logger.debug { "RSProx-Connection has successfully established a connection to $socketFile." } - val outputStream = DataOutputStream(socket.outputStream) - val msgOutput = Output(4096, -1) - val key = eventLog.consumerKey() - eventLog.addConsumer( - key, - syncBlock = { - try { - val snapshot = eventLog.snapshot(0) - outputStream.writeIndicator(FLAG_BEGIN_SYNC) - for (packet in snapshot) { - outputStream.writeMessage(msgOutput, packet) + srv.accept().use { socket -> + logger.debug { "RSProx-Connection has successfully established a connection to $socketFile." } + val outputStream = DataOutputStream(socket.outputStream) + val msgOutput = Output(4096, -1) + val queue = LinkedBlockingQueue() + val key = eventLog.consumerKey() + try { + eventLog.addConsumer(key, queue::offer) + while (true) { + when (val event = queue.take()) { + is UnixPacket -> outputStream.writeMessage(msgOutput, event) + is UnixIndicator -> outputStream.writeIndicator(event.value) } - outputStream.writeIndicator(FLAG_END_SYNC) - } catch (_: SocketException) { - eventLog.removeConsumer(key) - uniqueConnections.remove(socketFile) } - }, - consumer = { event -> - try { - outputStream.writeMessage(msgOutput, event) - } catch (_: SocketException) { - eventLog.removeConsumer(key) - uniqueConnections.remove(socketFile) - } - }, - ) + } catch (_: SocketException) { + // The RuneLite consumer disconnected. + } finally { + eventLog.removeConsumer(key) + uniqueConnections.remove(socketFile) + } + } } } } @@ -162,6 +156,11 @@ public class UnixSocketConnection( eventLog.append(UnixPacket(payload)) } + internal fun reset() { + uniqueConnections.clear() + eventLog.reset() + } + private fun DataOutputStream.writeMessage( msgOutput: Output, unixPacket: UnixPacket, @@ -189,10 +188,14 @@ public class UnixSocketConnection( val message: IncomingMessage, ) : Event + private class UnixIndicator( + val value: Int, + ) : Event + private class EventLog { private val lock = ReentrantLock() private val events = mutableListOf() - private val consumers = ConcurrentHashMap Unit>() + private val consumers = ConcurrentHashMap Unit>() private val consumerKeyCount: AtomicInteger = AtomicInteger(0) fun consumerKey(): Int { @@ -201,11 +204,12 @@ public class UnixSocketConnection( fun addConsumer( key: Int, - syncBlock: () -> Unit, - consumer: (event: UnixPacket) -> Unit, + consumer: (event: Event) -> Unit, ) { withLock { - syncBlock() + consumer(UnixIndicator(FLAG_BEGIN_SYNC)) + events.forEach(consumer) + consumer(UnixIndicator(FLAG_END_SYNC)) consumers.put(key, consumer) } } @@ -222,6 +226,13 @@ public class UnixSocketConnection( } } + fun reset() { + withLock { + events.clear() + consumers.clear() + } + } + fun append(event: UnixPacket) { return withLock { events.add(event) @@ -231,22 +242,6 @@ public class UnixSocketConnection( } } - fun snapshot(fromIndex: Int = 0): List { - return withLock { - if (fromIndex >= events.size) { - emptyList() - } else { - events.subList(fromIndex, events.size) - } - } - } - - fun size(): Int { - return withLock { - events.size - } - } - private inline fun withLock(block: () -> T): T { lock.lock() return try {