From 606ee48724fea7447e7fc4c34dd4591ef4a905be Mon Sep 17 00:00:00 2001 From: LossyDragon Date: Thu, 22 May 2025 15:39:33 -0500 Subject: [PATCH 01/11] Optimize gzip in HandleMulti --- .../dragonbra/javasteam/steam/CMClient.java | 46 ++++++++----------- 1 file changed, 19 insertions(+), 27 deletions(-) diff --git a/src/main/java/in/dragonbra/javasteam/steam/CMClient.java b/src/main/java/in/dragonbra/javasteam/steam/CMClient.java index 278e53c6..ea67fc25 100644 --- a/src/main/java/in/dragonbra/javasteam/steam/CMClient.java +++ b/src/main/java/in/dragonbra/javasteam/steam/CMClient.java @@ -28,8 +28,8 @@ import org.jetbrains.annotations.Nullable; import java.io.ByteArrayInputStream; -import java.io.ByteArrayOutputStream; import java.io.IOException; +import java.io.InputStream; import java.net.InetAddress; import java.net.InetSocketAddress; import java.util.EnumSet; @@ -39,6 +39,7 @@ * This base client handles the underlying connection to a CM server. This class should not be use directly, but through * the {@link in.dragonbra.javasteam.steam.steamclient.SteamClient SteamClient} class. */ +@SuppressWarnings("unused") public abstract class CMClient { private static final Logger logger = LogManager.getLogger(CMClient.class); @@ -365,45 +366,34 @@ public static IPacketMsg getPacketMsg(byte[] data) { } } + // region ClientMsg Handlers + private void handleMulti(IPacketMsg packetMsg) { if (!packetMsg.isProto()) { logger.debug("HandleMulti got non-proto MsgMulti!!"); return; } - ClientMsgProtobuf msgMulti = new ClientMsgProtobuf<>(CMsgMulti.class, packetMsg); + var msgMulti = new ClientMsgProtobuf(CMsgMulti.class, packetMsg); - byte[] payload = msgMulti.getBody().getMessageBody().toByteArray(); + try (var payloadStream = msgMulti.getBody().getMessageBody().newInput()) { + InputStream stream = payloadStream; - if (msgMulti.getBody().getSizeUnzipped() > 0) { - try (var gzin = new GZIPInputStream(new ByteArrayInputStream(payload)); - var baos = new ByteArrayOutputStream()) { - int res = 0; - byte[] buf = new byte[1024]; - while (res >= 0) { - res = gzin.read(buf, 0, buf.length); - if (res > 0) { - baos.write(buf, 0, res); - } - } - payload = baos.toByteArray(); - } catch (IOException e) { - logger.debug("HandleMulti encountered an exception when decompressing.", e); - return; + if (msgMulti.getBody().getSizeUnzipped() > 0) { + stream = new GZIPInputStream(msgMulti.getBody().getMessageBody().newInput()); } - } - try (var bais = new ByteArrayInputStream(payload); - var br = new BinaryReader(bais)) { - while (br.available() > 0) { - int subSize = br.readInt(); - byte[] subData = br.readBytes(subSize); + try (var br = new BinaryReader(stream)) { + while (br.available() > 0) { + int subSize = br.readInt(); + var subData = br.readBytes(subSize); - if (!onClientMsgReceived(getPacketMsg(subData))) { - break; + if (!onClientMsgReceived(getPacketMsg(subData))) { + break; + } } } - } catch (IOException e) { + } catch (Exception e) { logger.error("error in handleMulti()", e); } } @@ -490,6 +480,8 @@ private void handleSessionToken(IPacketMsg packetMsg) { sessionToken = sessToken.getBody().getToken(); } + // endregion + public SteamConfiguration getConfiguration() { return configuration; } From 00bcba9b60cd0d534fca0d417c3ed8f411f544b3 Mon Sep 17 00:00:00 2001 From: LossyDragon Date: Thu, 22 May 2025 15:46:09 -0500 Subject: [PATCH 02/11] Fix accidental resource leak --- src/main/java/in/dragonbra/javasteam/steam/CMClient.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main/java/in/dragonbra/javasteam/steam/CMClient.java b/src/main/java/in/dragonbra/javasteam/steam/CMClient.java index ea67fc25..8675cf9a 100644 --- a/src/main/java/in/dragonbra/javasteam/steam/CMClient.java +++ b/src/main/java/in/dragonbra/javasteam/steam/CMClient.java @@ -380,7 +380,7 @@ private void handleMulti(IPacketMsg packetMsg) { InputStream stream = payloadStream; if (msgMulti.getBody().getSizeUnzipped() > 0) { - stream = new GZIPInputStream(msgMulti.getBody().getMessageBody().newInput()); + stream = new GZIPInputStream(payloadStream); } try (var br = new BinaryReader(stream)) { From 083e73de53234f10ca5f1715aafc19612f5b825e Mon Sep 17 00:00:00 2001 From: LossyDragon Date: Sat, 31 May 2025 23:30:35 -0500 Subject: [PATCH 03/11] Rename .java to .kt --- .../javasteam/util/stream/{BinaryReader.java => BinaryReader.kt} | 0 .../javasteam/util/stream/{BinaryWriter.java => BinaryWriter.kt} | 0 2 files changed, 0 insertions(+), 0 deletions(-) rename src/main/java/in/dragonbra/javasteam/util/stream/{BinaryReader.java => BinaryReader.kt} (100%) rename src/main/java/in/dragonbra/javasteam/util/stream/{BinaryWriter.java => BinaryWriter.kt} (100%) diff --git a/src/main/java/in/dragonbra/javasteam/util/stream/BinaryReader.java b/src/main/java/in/dragonbra/javasteam/util/stream/BinaryReader.kt similarity index 100% rename from src/main/java/in/dragonbra/javasteam/util/stream/BinaryReader.java rename to src/main/java/in/dragonbra/javasteam/util/stream/BinaryReader.kt diff --git a/src/main/java/in/dragonbra/javasteam/util/stream/BinaryWriter.java b/src/main/java/in/dragonbra/javasteam/util/stream/BinaryWriter.kt similarity index 100% rename from src/main/java/in/dragonbra/javasteam/util/stream/BinaryWriter.java rename to src/main/java/in/dragonbra/javasteam/util/stream/BinaryWriter.kt From 9f64e558243f04bc38aa3500ad538acf6d18ef46 Mon Sep 17 00:00:00 2001 From: LossyDragon Date: Sat, 31 May 2025 23:30:35 -0500 Subject: [PATCH 04/11] Port BinaryReader and BinaryWriter to koltin. Tweaking them to be a bit more performant. --- .../util/compat/InputStreamCompat.kt | 4 +- .../javasteam/util/stream/BinaryReader.kt | 221 ++++++++++-------- .../javasteam/util/stream/BinaryWriter.kt | 81 +++---- 3 files changed, 166 insertions(+), 140 deletions(-) diff --git a/src/main/java/in/dragonbra/javasteam/util/compat/InputStreamCompat.kt b/src/main/java/in/dragonbra/javasteam/util/compat/InputStreamCompat.kt index 5f9ff54a..40a85ce3 100644 --- a/src/main/java/in/dragonbra/javasteam/util/compat/InputStreamCompat.kt +++ b/src/main/java/in/dragonbra/javasteam/util/compat/InputStreamCompat.kt @@ -42,7 +42,7 @@ fun InputStream.readNBytesCompat(len: Int): ByteArray { var n: Int do { - var buf = ByteArray(min(remaining, 8192)) + val buf = ByteArray(min(remaining, 8192)) var nread = 0 // read to EOF which may read more or less than buffer size @@ -82,7 +82,7 @@ fun InputStream.readNBytesCompat(len: Int): ByteArray { remaining = total bufs.forEach { b -> - var count = min(b.size, remaining) + val count = min(b.size, remaining) System.arraycopy(b, 0, result, offset, count) offset += count remaining -= count diff --git a/src/main/java/in/dragonbra/javasteam/util/stream/BinaryReader.kt b/src/main/java/in/dragonbra/javasteam/util/stream/BinaryReader.kt index 3ed59ab0..a638aca0 100644 --- a/src/main/java/in/dragonbra/javasteam/util/stream/BinaryReader.kt +++ b/src/main/java/in/dragonbra/javasteam/util/stream/BinaryReader.kt @@ -1,158 +1,181 @@ -package in.dragonbra.javasteam.util.stream; - -import in.dragonbra.javasteam.util.compat.ByteArrayOutputStreamCompat; - -import java.io.*; -import java.nio.charset.Charset; -import java.nio.charset.StandardCharsets; +package `in`.dragonbra.javasteam.util.stream + +import `in`.dragonbra.javasteam.util.compat.ByteArrayOutputStreamCompat +import `in`.dragonbra.javasteam.util.compat.readNBytesCompat +import java.io.ByteArrayOutputStream +import java.io.EOFException +import java.io.FilterInputStream +import java.io.IOException +import java.io.InputStream +import java.nio.charset.Charset +import java.nio.charset.StandardCharsets /** * Basically DataInputStream, but the bytes are parsed in reverse order */ -public class BinaryReader extends FilterInputStream { +class BinaryReader(inputStream: InputStream) : FilterInputStream(inputStream) { - private final byte[] readBuffer = new byte[8]; + private val readBuffer = ByteArray(16) - private int position = 0; + var position: Int = 0 + private set - public BinaryReader(InputStream in) { - super(in); - } + @Throws(IOException::class) + fun readInt(): Int { + val bytesRead = `in`.read(readBuffer, 0, 4) - public int readInt() throws IOException { - int ch1 = in.read(); - int ch2 = in.read(); - int ch3 = in.read(); - int ch4 = in.read(); - position += 4; - if ((ch1 | ch2 | ch3 | ch4) < 0) { - throw new EOFException(); + if (bytesRead != 4) { + throw EOFException() } - return ((ch4 << 24) + (ch3 << 16) + (ch2 << 8) + ch1); + + position += 4 + + return ((readBuffer[3].toInt() and 0xFF) shl 24) or + ((readBuffer[2].toInt() and 0xFF) shl 16) or + ((readBuffer[1].toInt() and 0xFF) shl 8) or + (readBuffer[0].toInt() and 0xFF) } - public byte[] readBytes(int len) throws IOException { + @Throws(IOException::class) + fun readBytes(len: Int): ByteArray { if (len < 0) { - throw new IOException("negative length"); + throw IOException("negative length") } - byte[] bytes = new byte[len]; + val bytes = `in`.readNBytesCompat(len) - for (int i = 0; i < bytes.length; i++) { - bytes[i] = readByte(); + if (bytes.size != len) { + throw EOFException("Unexpected end of stream") } - return bytes; + position += len + + return bytes } - public byte readByte() throws IOException { - int ch = in.read(); + @Throws(IOException::class) + fun readByte(): Byte { + val ch = `in`.read() + if (ch < 0) { - throw new EOFException(); + throw EOFException() } - position += 1; - return (byte) ch; + + position += 1 + + return ch.toByte() } - public short readShort() throws IOException { - int ch1 = in.read(); - int ch2 = in.read(); - if ((ch1 | ch2) < 0) { - throw new EOFException(); + @Throws(IOException::class) + fun readShort(): Short { + val bytesRead = `in`.read(readBuffer, 0, 2) + if (bytesRead != 2) { + throw EOFException() } - position += 2; - return (short) ((ch2 << 8) + ch1); + + position += 2 + + return (((readBuffer[1].toInt() and 0xFF) shl 8) or (readBuffer[0].toInt() and 0xFF)).toShort() } - public long readLong() throws IOException { - in.read(readBuffer, 0, 8); - position += 8; - return (((long) readBuffer[7] << 56) + - ((long) (readBuffer[6] & 255) << 48) + - ((long) (readBuffer[5] & 255) << 40) + - ((long) (readBuffer[4] & 255) << 32) + - ((long) (readBuffer[3] & 255) << 24) + - ((readBuffer[2] & 255) << 16) + - ((readBuffer[1] & 255) << 8) + - (readBuffer[0] & 255)); + @Throws(IOException::class) + fun readLong(): Long { + val bytesRead = `in`.read(readBuffer, 0, 8) + + if (bytesRead != 8) { + throw EOFException() + } + + position += 8 + + return ( + (readBuffer[7].toLong() shl 56) + + ((readBuffer[6].toInt() and 255).toLong() shl 48) + + ((readBuffer[5].toInt() and 255).toLong() shl 40) + + ((readBuffer[4].toInt() and 255).toLong() shl 32) + + ((readBuffer[3].toInt() and 255).toLong() shl 24) + + ((readBuffer[2].toInt() and 255) shl 16) + + ((readBuffer[1].toInt() and 255) shl 8) + + (readBuffer[0].toInt() and 255) + ) } - public char readChar() throws IOException { - int ch1 = in.read(); + @Throws(IOException::class) + fun readChar(): Char { + val ch1 = `in`.read() + if (ch1 < 0) { - throw new EOFException(); + throw EOFException() } - position += 1; - return (char) ch1; - } - public float readFloat() throws IOException { - return Float.intBitsToFloat(readInt()); - } + position += 1 - public double readDouble() throws IOException { - return Double.longBitsToDouble(readLong()); + return ch1.toChar() } - public boolean readBoolean() throws IOException { - int ch = in.read(); + @Throws(IOException::class) + fun readFloat(): Float = Float.fromBits(readInt()) + + @Throws(IOException::class) + fun readDouble(): Double = Double.fromBits(readLong()) + + @Throws(IOException::class) + fun readBoolean(): Boolean { + val ch = `in`.read() + if (ch < 0) { - throw new EOFException(); + throw EOFException() } - position += 1; - return ch != 0; - } - public String readNullTermString() throws IOException { - return readNullTermString(StandardCharsets.UTF_8); - } + position += 1 - public String readNullTermString(Charset charset) throws IOException { - if (charset == null) { - throw new IOException("charset is null"); - } + return ch != 0 + } - if (charset.equals(StandardCharsets.UTF_8)) { - return readNullTermUtf8String(); + @JvmOverloads + @Throws(IOException::class) + fun readNullTermString(charset: Charset = StandardCharsets.UTF_8): String { + if (charset == StandardCharsets.UTF_8) { + return readNullTermUtf8String() } - ByteArrayOutputStream buffer = new ByteArrayOutputStream(0); - BinaryWriter bw = new BinaryWriter(buffer); + val buffer = ByteArrayOutputStream(0) + val bw = BinaryWriter(buffer) while (true) { - char ch = readChar(); + val ch = readChar() - if (ch == 0) { - break; + if (ch.code == 0) { + break } - bw.writeChar(ch); + bw.writeChar(ch) } - byte[] bytes = buffer.toByteArray(); - position += bytes.length; + val bytes = buffer.toByteArray() - return new String(bytes, charset); + position += bytes.size + + return String(bytes, charset) } - private String readNullTermUtf8String() throws IOException { - ByteArrayOutputStream baos = new ByteArrayOutputStream(); - int b; + @Throws(IOException::class) + private fun readNullTermUtf8String(): String { + val baos = ByteArrayOutputStream() + var b: Int - while ((b = in.read()) != 0) { + while ((`in`.read().also { b = it }) != 0) { if (b <= 0) { - break; + break } - baos.write(b); - position++; - } - position++; // Increment for the null terminator + baos.write(b) - return ByteArrayOutputStreamCompat.toString(baos); - } + position++ + } + + position++ // Increment for the null terminator - public int getPosition() { - return position; + return ByteArrayOutputStreamCompat.toString(baos) } } diff --git a/src/main/java/in/dragonbra/javasteam/util/stream/BinaryWriter.kt b/src/main/java/in/dragonbra/javasteam/util/stream/BinaryWriter.kt index c4d56a32..c97ae203 100644 --- a/src/main/java/in/dragonbra/javasteam/util/stream/BinaryWriter.kt +++ b/src/main/java/in/dragonbra/javasteam/util/stream/BinaryWriter.kt @@ -1,61 +1,64 @@ -package in.dragonbra.javasteam.util.stream; - -import java.io.FilterOutputStream; -import java.io.IOException; -import java.io.OutputStream; +package `in`.dragonbra.javasteam.util.stream +import java.io.FilterOutputStream +import java.io.IOException +import java.io.OutputStream /** * Basically DataOutputStream, but the bytes are parsed in reverse order */ -public class BinaryWriter extends FilterOutputStream { - - private final byte[] writeBuffer = new byte[8]; +class BinaryWriter(out: OutputStream) : FilterOutputStream(out) { - public BinaryWriter(OutputStream out) { - super(out); - } + private val writeBuffer = ByteArray(8) - public void writeInt(int v) throws IOException { - out.write(v & 0xFF); - out.write((v >>> 8) & 0xFF); - out.write((v >>> 16) & 0xFF); - out.write((v >>> 24) & 0xFF); + @Throws(IOException::class) + fun writeInt(v: Int) { + out.write(v and 0xFF) + out.write((v ushr 8) and 0xFF) + out.write((v ushr 16) and 0xFF) + out.write((v ushr 24) and 0xFF) } - public void writeShort(short v) throws IOException { - out.write(v & 0xFF); - out.write((v >>> 8) & 0xFF); + @Throws(IOException::class) + fun writeShort(v: Short) { + out.write(v.toInt() and 0xFF) + out.write((v.toInt() ushr 8) and 0xFF) } - public void writeLong(long v) throws IOException { - writeBuffer[7] = (byte) (v >>> 56); - writeBuffer[6] = (byte) (v >>> 48); - writeBuffer[5] = (byte) (v >>> 40); - writeBuffer[4] = (byte) (v >>> 32); - writeBuffer[3] = (byte) (v >>> 24); - writeBuffer[2] = (byte) (v >>> 16); - writeBuffer[1] = (byte) (v >>> 8); - writeBuffer[0] = (byte) v; - out.write(writeBuffer, 0, 8); + @Throws(IOException::class) + fun writeLong(v: Long) { + writeBuffer[7] = (v ushr 56).toByte() + writeBuffer[6] = (v ushr 48).toByte() + writeBuffer[5] = (v ushr 40).toByte() + writeBuffer[4] = (v ushr 32).toByte() + writeBuffer[3] = (v ushr 24).toByte() + writeBuffer[2] = (v ushr 16).toByte() + writeBuffer[1] = (v ushr 8).toByte() + writeBuffer[0] = v.toByte() + out.write(writeBuffer, 0, 8) } - public void writeFloat(float v) throws IOException { - writeInt(Float.floatToIntBits(v)); + @Throws(IOException::class) + fun writeFloat(v: Float) { + writeInt(v.toBits()) } - public void writeDouble(double v) throws IOException { - writeLong(Double.doubleToLongBits(v)); + @Throws(IOException::class) + fun writeDouble(v: Double) { + writeLong(v.toBits()) } - public void writeBoolean(boolean v) throws IOException { - out.write(v ? 1 : 0); + @Throws(IOException::class) + fun writeBoolean(v: Boolean) { + out.write(if (v) 1 else 0) } - public void writeByte(byte v) throws IOException { - out.write(v); + @Throws(IOException::class) + fun writeByte(v: Byte) { + out.write(v.toInt()) } - public void writeChar(char v) throws IOException { - out.write(v); + @Throws(IOException::class) + fun writeChar(v: Char) { + out.write(v.code) } } From 4c6a77def00d9738041bdadbe9bbf8944df0d471 Mon Sep 17 00:00:00 2001 From: LossyDragon Date: Sat, 31 May 2025 23:31:09 -0500 Subject: [PATCH 05/11] Increase GZIPInputStream buffer size --- src/main/java/in/dragonbra/javasteam/steam/CMClient.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/main/java/in/dragonbra/javasteam/steam/CMClient.java b/src/main/java/in/dragonbra/javasteam/steam/CMClient.java index 8675cf9a..079240ad 100644 --- a/src/main/java/in/dragonbra/javasteam/steam/CMClient.java +++ b/src/main/java/in/dragonbra/javasteam/steam/CMClient.java @@ -25,6 +25,7 @@ import in.dragonbra.javasteam.util.log.LogManager; import in.dragonbra.javasteam.util.log.Logger; import in.dragonbra.javasteam.util.stream.BinaryReader; +import in.dragonbra.javasteam.util.stream.MemoryStream; import org.jetbrains.annotations.Nullable; import java.io.ByteArrayInputStream; @@ -376,11 +377,11 @@ private void handleMulti(IPacketMsg packetMsg) { var msgMulti = new ClientMsgProtobuf(CMsgMulti.class, packetMsg); - try (var payloadStream = msgMulti.getBody().getMessageBody().newInput()) { + try (var payloadStream = new MemoryStream(msgMulti.getBody().getMessageBody())) { InputStream stream = payloadStream; if (msgMulti.getBody().getSizeUnzipped() > 0) { - stream = new GZIPInputStream(payloadStream); + stream = new GZIPInputStream(payloadStream, 8192); } try (var br = new BinaryReader(stream)) { From 8554eed69ccac590ea4d0f3dc8c30a25434de122 Mon Sep 17 00:00:00 2001 From: LossyDragon Date: Sat, 31 May 2025 23:43:07 -0500 Subject: [PATCH 06/11] Cleanup --- .../javasteam/util/stream/BinaryReader.kt | 6 ++--- .../javasteam/util/stream/BinaryWriter.kt | 22 +++++++++---------- 2 files changed, 14 insertions(+), 14 deletions(-) diff --git a/src/main/java/in/dragonbra/javasteam/util/stream/BinaryReader.kt b/src/main/java/in/dragonbra/javasteam/util/stream/BinaryReader.kt index a638aca0..e5baa7cc 100644 --- a/src/main/java/in/dragonbra/javasteam/util/stream/BinaryReader.kt +++ b/src/main/java/in/dragonbra/javasteam/util/stream/BinaryReader.kt @@ -162,15 +162,15 @@ class BinaryReader(inputStream: InputStream) : FilterInputStream(inputStream) { @Throws(IOException::class) private fun readNullTermUtf8String(): String { val baos = ByteArrayOutputStream() - var b: Int - while ((`in`.read().also { b = it }) != 0) { + while (true) { + val b = `in`.read() + if (b <= 0) { break } baos.write(b) - position++ } diff --git a/src/main/java/in/dragonbra/javasteam/util/stream/BinaryWriter.kt b/src/main/java/in/dragonbra/javasteam/util/stream/BinaryWriter.kt index c97ae203..b28af24d 100644 --- a/src/main/java/in/dragonbra/javasteam/util/stream/BinaryWriter.kt +++ b/src/main/java/in/dragonbra/javasteam/util/stream/BinaryWriter.kt @@ -13,26 +13,26 @@ class BinaryWriter(out: OutputStream) : FilterOutputStream(out) { @Throws(IOException::class) fun writeInt(v: Int) { out.write(v and 0xFF) - out.write((v ushr 8) and 0xFF) - out.write((v ushr 16) and 0xFF) - out.write((v ushr 24) and 0xFF) + out.write((v shr 8) and 0xFF) + out.write((v shr 16) and 0xFF) + out.write((v shr 24) and 0xFF) } @Throws(IOException::class) fun writeShort(v: Short) { out.write(v.toInt() and 0xFF) - out.write((v.toInt() ushr 8) and 0xFF) + out.write((v.toInt() shr 8) and 0xFF) } @Throws(IOException::class) fun writeLong(v: Long) { - writeBuffer[7] = (v ushr 56).toByte() - writeBuffer[6] = (v ushr 48).toByte() - writeBuffer[5] = (v ushr 40).toByte() - writeBuffer[4] = (v ushr 32).toByte() - writeBuffer[3] = (v ushr 24).toByte() - writeBuffer[2] = (v ushr 16).toByte() - writeBuffer[1] = (v ushr 8).toByte() + writeBuffer[7] = (v shr 56).toByte() + writeBuffer[6] = (v shr 48).toByte() + writeBuffer[5] = (v shr 40).toByte() + writeBuffer[4] = (v shr 32).toByte() + writeBuffer[3] = (v shr 24).toByte() + writeBuffer[2] = (v shr 16).toByte() + writeBuffer[1] = (v shr 8).toByte() writeBuffer[0] = v.toByte() out.write(writeBuffer, 0, 8) } From 4c1d86dbbf7b2f1ab3d10703f1af627bba599f91 Mon Sep 17 00:00:00 2001 From: LossyDragon Date: Fri, 6 Jun 2025 00:05:35 -0500 Subject: [PATCH 07/11] Rename .java to .kt --- .../in/dragonbra/javasteam/steam/{CMClient.java => CMClient.kt} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename src/main/java/in/dragonbra/javasteam/steam/{CMClient.java => CMClient.kt} (100%) diff --git a/src/main/java/in/dragonbra/javasteam/steam/CMClient.java b/src/main/java/in/dragonbra/javasteam/steam/CMClient.kt similarity index 100% rename from src/main/java/in/dragonbra/javasteam/steam/CMClient.java rename to src/main/java/in/dragonbra/javasteam/steam/CMClient.kt From 870cb5e08816d2ebb2d748e599c6cac013088f84 Mon Sep 17 00:00:00 2001 From: LossyDragon Date: Fri, 6 Jun 2025 00:05:36 -0500 Subject: [PATCH 08/11] Port CMClient to kotlin, use okio buffers for multi messages. --- .../in/dragonbra/javasteam/steam/CMClient.kt | 926 +++++++++--------- .../steam/handlers/ClientMsgHandler.kt | 4 +- .../steam/steamclient/SteamClient.kt | 11 +- .../javasteam/steam/DummyClient.java | 2 +- 4 files changed, 455 insertions(+), 488 deletions(-) diff --git a/src/main/java/in/dragonbra/javasteam/steam/CMClient.kt b/src/main/java/in/dragonbra/javasteam/steam/CMClient.kt index 079240ad..30e94817 100644 --- a/src/main/java/in/dragonbra/javasteam/steam/CMClient.kt +++ b/src/main/java/in/dragonbra/javasteam/steam/CMClient.kt @@ -1,206 +1,287 @@ -package in.dragonbra.javasteam.steam; - -import in.dragonbra.javasteam.base.*; -import in.dragonbra.javasteam.enums.EMsg; -import in.dragonbra.javasteam.enums.EResult; -import in.dragonbra.javasteam.enums.EUniverse; -import in.dragonbra.javasteam.generated.MsgClientLogon; -import in.dragonbra.javasteam.generated.MsgClientServerUnavailable; -import in.dragonbra.javasteam.networking.steam3.*; -import in.dragonbra.javasteam.protobufs.steamclient.SteammessagesBase.CMsgMulti; -import in.dragonbra.javasteam.protobufs.steamclient.SteammessagesClientserver.CMsgClientSessionToken; -import in.dragonbra.javasteam.protobufs.steamclient.SteammessagesClientserverLogin.CMsgClientHeartBeat; -import in.dragonbra.javasteam.protobufs.steamclient.SteammessagesClientserverLogin.CMsgClientHello; -import in.dragonbra.javasteam.protobufs.steamclient.SteammessagesClientserverLogin.CMsgClientLoggedOff; -import in.dragonbra.javasteam.protobufs.steamclient.SteammessagesClientserverLogin.CMsgClientLogonResponse; -import in.dragonbra.javasteam.steam.discovery.ServerQuality; -import in.dragonbra.javasteam.steam.discovery.ServerRecord; -import in.dragonbra.javasteam.steam.discovery.SmartCMServerList; -import in.dragonbra.javasteam.steam.steamclient.configuration.SteamConfiguration; -import in.dragonbra.javasteam.types.SteamID; -import in.dragonbra.javasteam.util.*; -import in.dragonbra.javasteam.util.event.EventArgs; -import in.dragonbra.javasteam.util.event.EventHandler; -import in.dragonbra.javasteam.util.event.ScheduledFunction; -import in.dragonbra.javasteam.util.log.LogManager; -import in.dragonbra.javasteam.util.log.Logger; -import in.dragonbra.javasteam.util.stream.BinaryReader; -import in.dragonbra.javasteam.util.stream.MemoryStream; -import org.jetbrains.annotations.Nullable; - -import java.io.ByteArrayInputStream; -import java.io.IOException; -import java.io.InputStream; -import java.net.InetAddress; -import java.net.InetSocketAddress; -import java.util.EnumSet; -import java.util.zip.GZIPInputStream; +package `in`.dragonbra.javasteam.steam + +import `in`.dragonbra.javasteam.base.ClientMsg +import `in`.dragonbra.javasteam.base.ClientMsgProtobuf +import `in`.dragonbra.javasteam.base.IClientMsg +import `in`.dragonbra.javasteam.base.IPacketMsg +import `in`.dragonbra.javasteam.base.PacketClientMsg +import `in`.dragonbra.javasteam.base.PacketClientMsgProtobuf +import `in`.dragonbra.javasteam.base.PacketMsg +import `in`.dragonbra.javasteam.enums.EMsg +import `in`.dragonbra.javasteam.enums.EResult +import `in`.dragonbra.javasteam.enums.EUniverse +import `in`.dragonbra.javasteam.generated.MsgClientLogon +import `in`.dragonbra.javasteam.generated.MsgClientServerUnavailable +import `in`.dragonbra.javasteam.networking.steam3.Connection +import `in`.dragonbra.javasteam.networking.steam3.DisconnectedEventArgs +import `in`.dragonbra.javasteam.networking.steam3.IConnectionFactory +import `in`.dragonbra.javasteam.networking.steam3.NetMsgEventArgs +import `in`.dragonbra.javasteam.networking.steam3.ProtocolTypes +import `in`.dragonbra.javasteam.protobufs.steamclient.SteammessagesBase.CMsgMulti +import `in`.dragonbra.javasteam.protobufs.steamclient.SteammessagesClientserver.CMsgClientSessionToken +import `in`.dragonbra.javasteam.protobufs.steamclient.SteammessagesClientserverLogin.CMsgClientHeartBeat +import `in`.dragonbra.javasteam.protobufs.steamclient.SteammessagesClientserverLogin.CMsgClientHello +import `in`.dragonbra.javasteam.protobufs.steamclient.SteammessagesClientserverLogin.CMsgClientLoggedOff +import `in`.dragonbra.javasteam.protobufs.steamclient.SteammessagesClientserverLogin.CMsgClientLogonResponse +import `in`.dragonbra.javasteam.steam.discovery.ServerQuality +import `in`.dragonbra.javasteam.steam.discovery.ServerRecord +import `in`.dragonbra.javasteam.steam.discovery.SmartCMServerList +import `in`.dragonbra.javasteam.steam.steamclient.SteamClient +import `in`.dragonbra.javasteam.steam.steamclient.configuration.SteamConfiguration +import `in`.dragonbra.javasteam.types.SteamID +import `in`.dragonbra.javasteam.util.IDebugNetworkListener +import `in`.dragonbra.javasteam.util.MsgUtil +import `in`.dragonbra.javasteam.util.NetHelpers +import `in`.dragonbra.javasteam.util.NetHookNetworkListener +import `in`.dragonbra.javasteam.util.Strings +import `in`.dragonbra.javasteam.util.event.EventArgs +import `in`.dragonbra.javasteam.util.event.EventHandler +import `in`.dragonbra.javasteam.util.event.ScheduledFunction +import `in`.dragonbra.javasteam.util.log.LogManager +import `in`.dragonbra.javasteam.util.log.Logger +import `in`.dragonbra.javasteam.util.stream.BinaryReader +import okio.Buffer +import okio.Source +import okio.buffer +import okio.gzip +import java.io.ByteArrayInputStream +import java.io.IOException +import java.net.InetAddress +import java.net.InetSocketAddress +import java.util.EnumSet /** - * This base client handles the underlying connection to a CM server. This class should not be use directly, but through - * the {@link in.dragonbra.javasteam.steam.steamclient.SteamClient SteamClient} class. + * This base client handles the underlying connection to a CM server. This class should not be use directly, + * but through the [SteamClient] class. + * + * @constructor Initializes a new instance of the [CMClient] class with a specific configuration. + * @param configuration The configuration to use for this client. + * @param identifier A specific identifier to be used to uniquely identify this instance. + * @throws IllegalArgumentException The identifier is an empty string */ -@SuppressWarnings("unused") -public abstract class CMClient { - - private static final Logger logger = LogManager.getLogger(CMClient.class); - - private final SteamConfiguration configuration; - - @Nullable - private InetAddress publicIP; - - @Nullable - private String ipCountryCode; - - private boolean isConnected; - - private long sessionToken; - - @Nullable - private Integer cellID; - - @Nullable - private Integer sessionID; - - @Nullable - private SteamID steamID; - - private IDebugNetworkListener debugNetworkListener; - - private boolean expectDisconnection; - - // connection lock around the setup and tear down of the connection task - private final Object connectionLock = new Object(); - - @Nullable - private Connection connection; - - private final ScheduledFunction heartBeatFunc; +@Suppress("unused") +abstract class CMClient +@Throws(IllegalArgumentException::class) +constructor( + val configuration: SteamConfiguration, + val identifier: String, +) { + companion object { + private val logger: Logger = LogManager.getLogger(CMClient::class.java) + + // TODO move to Utils once that is ported to Kotlin + fun Long.toSteamID(): SteamID = SteamID(this) + + @JvmStatic + fun getPacketMsg(data: ByteArray): IPacketMsg? { + if (data.size < 4) { + logger.debug( + "PacketMsg too small to contain a message, was only ${data.size} bytes. " + + "Message: " + Strings.toHex(data) + ) + return null + } - private final EventHandler netMsgReceived = (sender, e) -> onClientMsgReceived(getPacketMsg(e.getData())); + var rawEMsg = 0 + BinaryReader(ByteArrayInputStream(data)).use { reader -> + rawEMsg = reader.readInt() + } + val eMsg: EMsg = MsgUtil.getMsg(rawEMsg) + + when (eMsg) { + EMsg.ChannelEncryptRequest, + EMsg.ChannelEncryptResponse, + EMsg.ChannelEncryptResult, + -> try { + return PacketMsg(eMsg, data) + } catch (e: IOException) { + logger.debug("Exception deserializing emsg " + eMsg + " (" + MsgUtil.isProtoBuf(rawEMsg) + ").", e) + } - private final EventHandler connected = (sender, e) -> { - logger.debug("EventHandler `connected` called"); + else -> Unit + } - getServers().tryMark(connection.getCurrentEndPoint(), connection.getProtocolTypes(), ServerQuality.GOOD); + try { + return if (MsgUtil.isProtoBuf(rawEMsg)) { + // if the emsg is flagged, we're a proto message + PacketClientMsgProtobuf(eMsg, data) + } else { + PacketClientMsg(eMsg, data) + } + } catch (e: IOException) { + logger.debug("Exception deserializing emsg $eMsg (${MsgUtil.isProtoBuf(rawEMsg)}).", e) + return null + } + } + } - isConnected = true; + /** + * Bootstrap list of CM servers. + */ + val servers: SmartCMServerList + get() = configuration.serverList - try { - onClientConnected(); - } catch (Exception ex) { - logger.error("Unhandled exception after connecting: ", ex); - disconnect(false); - } - }; + /** + * Returns the local IP of this client. + */ + val localIP: InetAddress? + get() = connection?.localIP - private final EventHandler disconnected = new EventHandler<>() { - @Override - public void handleEvent(Object sender, DisconnectedEventArgs e) { - logger.debug("EventHandler `disconnected` called. User Initiated: " + e.isUserInitiated() + - ", Expected Disconnection: " + expectDisconnection); + /** + * Returns the current endpoint this client is connected to. + */ + val currentEndPoint: InetSocketAddress? + get() = connection?.currentEndPoint - isConnected = false; + /** + * Gets the public IP address of this client. This value is assigned after a logon attempt has succeeded. + * This value will be null if the client is logged off of Steam. + */ + var publicIP: InetAddress? = null + private set - if (!e.isUserInitiated() && !expectDisconnection) { - getServers().tryMark(connection.getCurrentEndPoint(), connection.getProtocolTypes(), ServerQuality.BAD); - } + /** + * Gets the country code of our public IP address according to Steam. This value is assigned after a logon attempt has succeeded. + * This value will be null if the client is logged off of Steam. + */ + var ipCountryCode: String? = null + private set - sessionID = null; - steamID = null; + /** + * Gets the universe of this client. + */ + val universe: EUniverse + get() = configuration.universe - connection.getNetMsgReceived().removeEventHandler(netMsgReceived); - connection.getConnected().removeEventHandler(connected); - connection.getDisconnected().removeEventHandler(this); - connection = null; + /** + * Gets a value indicating whether this instance is connected to the remote CM server. + * true if this instance is connected; otherwise, false. + */ + var isConnected: Boolean = false + private set - heartBeatFunc.stop(); + /** + * Gets a value indicating whether isConnected and connection is not connected to the remote CM server. + * Inverse alternative to [CMClient.isConnected] + * @return **true** is this instance is disconnected, otherwise, **false**. + */ + // JavaSteam addition: "since the client can technically not be connected but still have a connection" + fun isDisconnected(): Boolean = !isConnected && connection == null - onClientDisconnected(e.isUserInitiated() || expectDisconnection); - } - }; + /** + * Gets the session token assigned to this client from the AM. + */ + var sessionToken: Long = 0L + private set - public CMClient(SteamConfiguration configuration) { - if (configuration == null) { - throw new IllegalArgumentException("configuration is null"); - } + /** + * Gets the Steam recommended Cell ID of this client. This value is assigned after a logon attempt has succeeded. + * This value will be null if the client is logged off of Steam. + */ + var cellID: Int? = null + private set - this.configuration = configuration; + /** + * Gets the session ID of this client. This value is assigned after a logon attempt has succeeded. + * This value will be null if the client is logged off of Steam. + */ + var sessionID: Int? = null + private set - heartBeatFunc = new ScheduledFunction(() -> { - var heartbeat = new ClientMsgProtobuf( - CMsgClientHeartBeat.class, EMsg.ClientHeartBeat); - heartbeat.getBody().setSendReply(true); // Ping Pong - send(heartbeat); - }, 5000); - } + /** + * Gets the SteamID of this client. This value is assigned after a logon attempt has succeeded. + * This value will be null if the client is logged off of Steam. + */ + var steamID: SteamID? = null + private set /** - * Debugging only method: - * Do not use this directly. + * Gets or sets the connection timeout used when connecting to the Steam server. */ - public void receiveTestPacketMsg(IPacketMsg packetMsg) { - onClientMsgReceived(packetMsg); - } + val connectionTimeout: Long + get() = configuration.connectionTimeout /** - * Debugging only method: - * Do not use this directly. + * Gets or sets the network listening interface. Use this for debugging only. + * For your convenience, you can use [NetHookNetworkListener] class. */ - public void setIsConnected(boolean value) { - isConnected = value; - } + var debugNetworkListener: IDebugNetworkListener? = null /** - * Connects this client to a Steam3 server. This begins the process of connecting and encrypting the data channel - * between the client and the server. Results are returned asynchronously in a {@link in.dragonbra.javasteam.steam.steamclient.callbacks.ConnectedCallback ConnectedCallback}. If the - * server that SteamKit attempts to connect to is down, a {@link in.dragonbra.javasteam.steam.steamclient.callbacks.DisconnectedCallback DisconnectedCallback} will be posted instead. - * SteamKit will not attempt to reconnect to Steam, you must handle this callback and call Connect again preferably - * after a short delay. SteamKit will randomly select a CM server from its internal list. + * */ - public void connect() { - connect(null); + var expectDisconnection: Boolean = false + + // connection lock around the setup and tear down of the connection task + private val connectionLock: Any = Any() + + @Volatile + private var connection: Connection? = null + + val heartBeatFunc: ScheduledFunction + + init { + if (identifier.isBlank()) { + throw IllegalArgumentException("identifier must not be empty") + } + + heartBeatFunc = ScheduledFunction({ + val heartbeat = ClientMsgProtobuf( + CMsgClientHeartBeat::class.java, + EMsg.ClientHeartBeat + ).apply { + body.sendReply = true // Ping Pong + } + send(heartbeat) + }, 5000) } /** * Connects this client to a Steam3 server. This begins the process of connecting and encrypting the data channel - * between the client and the server. Results are returned asynchronously in a {@link in.dragonbra.javasteam.steam.steamclient.callbacks.ConnectedCallback ConnectedCallback}. If the - * server that SteamKit attempts to connect to is down, a {@link in.dragonbra.javasteam.steam.steamclient.callbacks.DisconnectedCallback DisconnectedCallback} will be posted instead. + * between the client and the server. Results are returned asynchronously in a [ConnectedCallback][in.dragonbra.javasteam.steam.steamclient.callbacks.ConnectedCallback]. If the + * server that SteamKit attempts to connect to is down, a [DisconnectedCallback][in.dragonbra.javasteam.steam.steamclient.callbacks.DisconnectedCallback] will be posted instead. * SteamKit will not attempt to reconnect to Steam, you must handle this callback and call Connect again preferably * after a short delay. * - * @param cmServer The {@link ServerRecord} of the CM server to connect to. + * @param cmServer The [ServerRecord] of the CM server to connect to. */ - public void connect(ServerRecord cmServer) { - synchronized (connectionLock) { + @JvmOverloads + fun connect(cmServer: ServerRecord? = null) { + var cmServer = cmServer + synchronized(connectionLock) { try { - disconnect(true); + disconnect(true) - assert connection == null; + assert(connection == null) - expectDisconnection = false; + expectDisconnection = false if (cmServer == null) { - cmServer = getServers().getNextServerCandidate(configuration.getProtocolTypes()); + cmServer = servers.getNextServerCandidate(configuration.protocolTypes) } if (cmServer == null) { - logger.error("No CM servers available to connect to"); - onClientDisconnected(false); - return; + logger.error("No CM servers available to connect to") + onClientDisconnected(false) + return } - connection = createConnection(cmServer.getProtocolTypes()); - connection.getNetMsgReceived().addEventHandler(netMsgReceived); - connection.getConnected().addEventHandler(connected); - connection.getDisconnected().addEventHandler(disconnected); - logger.debug(String.format("Connecting to %s with protocol %s, and with connection impl %s", - cmServer.getEndpoint(), cmServer.getProtocolTypes(), connection.getClass().getSimpleName())); - connection.connect(cmServer.getEndpoint()); - } catch (Exception e) { - logger.debug("Failed to connect to Steam network", e); - onClientDisconnected(false); + connection = createConnection(cmServer.protocolTypes) + connection!!.getNetMsgReceived().addEventHandler(netMsgReceived) + connection!!.getConnected().addEventHandler(connected) + connection!!.getDisconnected().addEventHandler(disconnected) + logger.debug( + String.format( + "Connecting to %s with protocol %s, and with connection impl %s", + cmServer.endpoint, + cmServer.protocolTypes, + connection!!.javaClass.getSimpleName() + ) + ) + connection!!.connect(cmServer.endpoint) + } catch (e: Exception) { + logger.debug("Failed to connect to Steam network", e) + onClientDisconnected(false) } } } @@ -208,434 +289,315 @@ public abstract class CMClient { /** * Disconnects this client. */ - public void disconnect() { - disconnect(true); - } - - private void disconnect(boolean userInitiated) { - synchronized (connectionLock) { - heartBeatFunc.stop(); - + @JvmOverloads + fun disconnect(userInitiated: Boolean = true) { + synchronized(connectionLock) { + heartBeatFunc.stop() if (connection != null) { - connection.disconnect(userInitiated); + connection!!.disconnect(userInitiated) } } } /** - * Sends the specified client message to the server. This method automatically assigns the correct SessionID and - * SteamID of the message. - * + * Sends the specified client message to the server. + * This method automatically assigns the correct [sessionID] and [SteamID] of the message. * @param msg The client message to send. */ - public void send(IClientMsg msg) { - if (msg == null) { - throw new IllegalArgumentException("A value for 'msg' must be supplied"); - } - - Integer _sessionID = this.sessionID; - - if (_sessionID != null) { - msg.setSessionID(_sessionID); + fun send(msg: IClientMsg) { + if (!isConnected) { + logger.error("Send() was called while not connected to Steam.") } - SteamID _steamID = this.steamID; - - if (_steamID != null) { - msg.setSteamID(_steamID); - } + sessionID?.let { msg.setSessionID(it) } + steamID?.let { msg.setSteamID(it) } try { - if (debugNetworkListener != null) { - debugNetworkListener.onOutgoingNetworkMessage(msg.getMsgType(), msg.serialize()); - } - } catch (Exception e) { - logger.debug("DebugNetworkListener threw an exception", e); + debugNetworkListener?.onOutgoingNetworkMessage(msg.getMsgType(), msg.serialize()) + } catch (e: Exception) { + logger.debug("DebugNetworkListener threw an exception", e) } // we'll swallow any network failures here because they will be thrown later // on the network thread, and that will lead to a disconnect callback // down the line - - if (connection != null) { - connection.send(msg.serialize()); - } + connection?.send(msg.serialize()) } - protected boolean onClientMsgReceived(IPacketMsg packetMsg) { + // TODO: override fun logDebug() + + /** + * Called when a client message is received from the network. + * @param packetMsg The packet message. + */ + protected open fun onClientMsgReceived(packetMsg: IPacketMsg?): Boolean { if (packetMsg == null) { - logger.debug("Packet message failed to parse, shutting down connection"); - disconnect(false); - return false; + logger.debug("Packet message failed to parse, shutting down connection") + disconnect(userInitiated = false) + return false } // Multi message gets logged down the line after it's decompressed if (packetMsg.getMsgType() != EMsg.Multi) { try { - if (debugNetworkListener != null) { - debugNetworkListener.onIncomingNetworkMessage(packetMsg.getMsgType(), packetMsg.getData()); - } - } catch (Exception e) { - logger.debug("debugNetworkListener threw an exception", e); + debugNetworkListener?.onIncomingNetworkMessage(packetMsg.getMsgType(), packetMsg.getData()) + } catch (e: Exception) { + logger.debug("debugNetworkListener threw an exception", e) } } - switch (packetMsg.getMsgType()) { - case Multi: - handleMulti(packetMsg); - break; - case ClientLogOnResponse: // we handle this to get the SteamID/SessionID and to set up heart beating - handleLogOnResponse(packetMsg); - break; - case ClientLoggedOff: // to stop heart beating when we get logged off - handleLoggedOff(packetMsg); - break; - case ClientServerUnavailable: - handleServerUnavailable(packetMsg); - break; - case ClientSessionToken: // am session token - handleSessionToken(packetMsg); - break; + when (packetMsg.getMsgType()) { + EMsg.Multi -> handleMulti(packetMsg) + EMsg.ClientLogOnResponse -> handleLogOnResponse(packetMsg) // we handle this to get the SteamID/SessionID and to setup heartbeating + EMsg.ClientLoggedOff -> handleLoggedOff(packetMsg) // to stop heartbeating when we get logged off + EMsg.ClientServerUnavailable -> handleServerUnavailable(packetMsg) + EMsg.ClientSessionToken -> handleSessionToken(packetMsg) // am session token + else -> Unit } - return true; + return true } /** * Called when the client is securely isConnected to Steam3. */ - protected void onClientConnected() { - ClientMsgProtobuf request = new ClientMsgProtobuf<>(CMsgClientHello.class, EMsg.ClientHello); - request.getBody().setProtocolVersion(MsgClientLogon.CurrentProtocol); + protected open fun onClientConnected() { + val request = ClientMsgProtobuf( + CMsgClientHello::class.java, + EMsg.ClientHello + ).apply { + body.protocolVersion = MsgClientLogon.CurrentProtocol + } - send(request); + send(request) } /** * Called when the client is physically disconnected from Steam3. - * * @param userInitiated whether the disconnect was initialized by the client */ - protected void onClientDisconnected(boolean userInitiated) { - } + protected open fun onClientDisconnected(userInitiated: Boolean) {} - private Connection createConnection(EnumSet protocol) { - IConnectionFactory connectionFactory = configuration.getConnectionFactory(); - Connection connection = connectionFactory.createConnection(configuration, protocol); + private fun createConnection(protocol: EnumSet): Connection { + val connectionFactory: IConnectionFactory = configuration.connectionFactory + val connection = connectionFactory.createConnection(configuration, protocol) if (connection == null) { - logger.error(String.format("Connection factory returned null connection for protocols %s", protocol)); - throw new IllegalArgumentException("Connection factory returned null connection."); + logger.error(String.format("Connection factory returned null connection for protocols %s", protocol)) + throw IllegalArgumentException("Connection factory returned null connection.") } - return connection; + return connection } - public static IPacketMsg getPacketMsg(byte[] data) { - if (data.length < 4) { - logger.debug("PacketMsg too small to contain a message, was only " + data.length + " bytes. Message: " + Strings.toHex(data)); - return null; - } + /** + * Debugging: Do not use this directly. + */ + fun receiveTestPacketMsg(packetMsg: IPacketMsg) { + onClientMsgReceived(packetMsg) + } + + /** + * Debugging: Do not use this directly. + */ + fun setIsConnected(value: Boolean) { + isConnected = value + } - int rawEMsg = 0; - try (var reader = new BinaryReader(new ByteArrayInputStream(data))) { - rawEMsg = reader.readInt(); - } catch (IOException e) { - logger.debug("Exception while getting EMsg code", e); + private val netMsgReceived = EventHandler { _, e -> + onClientMsgReceived(getPacketMsg(e.data)) + } + + private val connected = EventHandler { _, e -> + logger.debug("EventHandler `connected` called") + + if (connection == null) { + logger.error("No connection object after connecting.") } - EMsg eMsg = MsgUtil.getMsg(rawEMsg); - - switch (eMsg) { - case ChannelEncryptRequest: - case ChannelEncryptResponse: - case ChannelEncryptResult: - try { - return new PacketMsg(eMsg, data); - } catch (IOException e) { - logger.debug("Exception deserializing emsg " + eMsg + " (" + MsgUtil.isProtoBuf(rawEMsg) + ").", e); - } + + if (connection?.currentEndPoint == null) { + logger.error("No connection endpoint after connecting - cannot update server list") } + servers.tryMark( + endPoint = connection!!.getCurrentEndPoint(), + protocolTypes = connection!!.getProtocolTypes(), + quality = ServerQuality.GOOD + ) + + isConnected = true + try { - if (MsgUtil.isProtoBuf(rawEMsg)) { - // if the emsg is flagged, we're a proto message - return new PacketClientMsgProtobuf(eMsg, data); - } else { - return new PacketClientMsg(eMsg, data); + onClientConnected() + } catch (ex: Exception) { + logger.error("Unhandled exception after connecting: ", ex) + disconnect(userInitiated = false) + } + } + + private val disconnected = object : EventHandler { + override fun handleEvent( + sender: Any?, + e: DisconnectedEventArgs, + ) { + logger.debug( + "EventHandler `disconnected` called. User Initiated: ${e.isUserInitiated}," + + "Expected Disconnection: $expectDisconnection" + ) + isConnected = false + + if (!e.isUserInitiated && !expectDisconnection) { + servers.tryMark( + endPoint = connection!!.currentEndPoint, + protocolTypes = connection!!.protocolTypes, + quality = ServerQuality.BAD + ) } - } catch (IOException e) { - logger.debug("Exception deserializing emsg " + eMsg + " (" + MsgUtil.isProtoBuf(rawEMsg) + ").", e); - return null; + + sessionID = null + steamID = null + + connection!!.getNetMsgReceived().removeEventHandler(netMsgReceived) + connection!!.getConnected().removeEventHandler(connected) + connection!!.getDisconnected().removeEventHandler(this) + connection = null // Why do we null here, but SK doesn't? + + heartBeatFunc.stop() + + onClientDisconnected(userInitiated = e.isUserInitiated || expectDisconnection) } } // region ClientMsg Handlers - private void handleMulti(IPacketMsg packetMsg) { + private fun handleMulti(packetMsg: IPacketMsg) { if (!packetMsg.isProto()) { - logger.debug("HandleMulti got non-proto MsgMulti!!"); - return; + logger.debug("HandleMulti got non-proto MsgMulti!!") + return } - var msgMulti = new ClientMsgProtobuf(CMsgMulti.class, packetMsg); - - try (var payloadStream = new MemoryStream(msgMulti.getBody().getMessageBody())) { - InputStream stream = payloadStream; + val msgMulti = ClientMsgProtobuf(CMsgMulti::class.java, packetMsg) - if (msgMulti.getBody().getSizeUnzipped() > 0) { - stream = new GZIPInputStream(payloadStream, 8192); + val payloadBuffer = Buffer().write(msgMulti.body.messageBody.toByteArray()).let { + if (msgMulti.body.sizeUnzipped > 0) { + (it as Source).gzip().buffer() + } else { + it } + } - try (var br = new BinaryReader(stream)) { - while (br.available() > 0) { - int subSize = br.readInt(); - var subData = br.readBytes(subSize); + do { + val packetSize = payloadBuffer.readIntLe() + val packetContent = payloadBuffer.readByteArray(packetSize.toLong()) + val packet = getPacketMsg(packetContent) - if (!onClientMsgReceived(getPacketMsg(subData))) { - break; - } - } + if (!onClientMsgReceived(packet)) { + break } - } catch (Exception e) { - logger.error("error in handleMulti()", e); - } + } while (!payloadBuffer.exhausted()) + + payloadBuffer.close() } - private void handleLogOnResponse(IPacketMsg packetMsg) { + private fun handleLogOnResponse(packetMsg: IPacketMsg) { if (!packetMsg.isProto()) { // a non-proto ClientLogonResponse can come in as a result of connecting but never sending a ClientLogon // in this case, it always fails, so we don't need to do anything special here - logger.debug("Got non-proto logon response, this is indicative of no logon attempt after connecting."); - return; + logger.debug("Got non-proto logon response, this is indicative of no logon attempt after connecting.") + return } - ClientMsgProtobuf logonResp = new ClientMsgProtobuf<>(CMsgClientLogonResponse.class, packetMsg); + val logonResp = ClientMsgProtobuf( + CMsgClientLogonResponse::class.java, + packetMsg + ) + val logonResponse: EResult = EResult.from(logonResp.body.eresult) + val extendedResponse: EResult = EResult.from(logonResp.body.eresultExtended) - EResult logonResponse = EResult.from(logonResp.getBody().getEresult()); - EResult extendedResponse = EResult.from(logonResp.getBody().getEresultExtended()); - logger.debug("handleLogOnResponse got response: " + logonResponse + ", extended: " + extendedResponse); + logger.debug("handleLogOnResponse got response: $logonResponse, extended: $extendedResponse") // Note: Sometimes if you sign in too many times, steam may confuse "InvalidPassword" with "RateLimitExceeded" - if (logonResponse == EResult.OK) { - sessionID = logonResp.getProtoHeader().getClientSessionid(); - steamID = new SteamID(logonResp.getProtoHeader().getSteamid()); + sessionID = logonResp.protoHeader.clientSessionid + steamID = logonResp.protoHeader.steamid.toSteamID() + + cellID = logonResp.body.cellId + publicIP = NetHelpers.getIPAddress(logonResp.body.publicIp) + ipCountryCode = logonResp.body.ipCountryCode - cellID = logonResp.getBody().getCellId(); - publicIP = NetHelpers.getIPAddress(logonResp.getBody().getPublicIp()); - ipCountryCode = logonResp.getBody().getIpCountryCode(); + val hbDelay = logonResp.body.legacyOutOfGameHeartbeatSeconds // restart heartbeat - heartBeatFunc.stop(); - heartBeatFunc.setDelay(logonResp.getBody().getLegacyOutOfGameHeartbeatSeconds() * 1000L); - heartBeatFunc.start(); + heartBeatFunc.stop() + heartBeatFunc.delay = hbDelay * 1000L + heartBeatFunc.start() } else if (logonResponse == EResult.TryAnotherCM || logonResponse == EResult.ServiceUnavailable) { - var connection = this.connection; - if (connection != null) { - getServers().tryMark(connection.getCurrentEndPoint(), connection.getProtocolTypes(), ServerQuality.BAD); - } else { - logger.error("Connection was null trying to mark endpoint bad."); + if (connection?.currentEndPoint != null) { + servers.tryMark( + endPoint = connection?.currentEndPoint, + protocolTypes = connection?.protocolTypes, + quality = ServerQuality.BAD + ) } } } - private void handleLoggedOff(IPacketMsg packetMsg) { - sessionID = null; - steamID = null; + private fun handleLoggedOff(packetMsg: IPacketMsg) { + sessionID = null + steamID = null - cellID = null; - publicIP = null; - ipCountryCode = null; + cellID = null + publicIP = null + ipCountryCode = null - heartBeatFunc.stop(); + heartBeatFunc.stop() if (packetMsg.isProto()) { - ClientMsgProtobuf logoffMsg = new ClientMsgProtobuf<>(CMsgClientLoggedOff.class, packetMsg); - EResult logoffResult = EResult.from(logoffMsg.getBody().getEresult()); + val logoffMsg = ClientMsgProtobuf( + CMsgClientLoggedOff::class.java, + packetMsg + ) + val logoffResult = EResult.from(logoffMsg.body.eresult) - logger.debug("handleLoggedOff got " + logoffResult); + logger.debug("handleLoggedOff got $logoffResult") if (logoffResult == EResult.TryAnotherCM || logoffResult == EResult.ServiceUnavailable) { - var connection = this.connection; - if (connection != null) { - getServers().tryMark(connection.getCurrentEndPoint(), connection.getProtocolTypes(), ServerQuality.BAD); - } else { - logger.error("Connection was null trying to mark endpoint bad."); + if (connection == null) { + logger.error("No connection object during ClientLoggedOff.") + } + + if (connection?.currentEndPoint == null) { + logger.error("No connection endpoint during ClientLoggedOff - cannot update server list status") } + + servers.tryMark( + endPoint = connection?.currentEndPoint, + protocolTypes = connection?.protocolTypes, + quality = ServerQuality.BAD + ) } } else { - logger.debug("handleLoggedOff got unexpected response: " + packetMsg.getMsgType()); + logger.debug("handleLoggedOff got unexpected response: " + packetMsg.getMsgType()) } } - private void handleServerUnavailable(IPacketMsg packetMsg) { - var msgServerUnavailable = new ClientMsg<>(MsgClientServerUnavailable.class, packetMsg); + private fun handleServerUnavailable(packetMsg: IPacketMsg) { + val msgServerUnavailable = ClientMsg(MsgClientServerUnavailable::class.java, packetMsg) - logger.debug("A server of type " + msgServerUnavailable.getBody().getEServerTypeUnavailable() + - "was not available for request: " + EMsg.from(msgServerUnavailable.getBody().getEMsgSent())); + logger.debug( + "A server of type " + msgServerUnavailable.body.eServerTypeUnavailable + + "was not available for request: " + EMsg.from(msgServerUnavailable.body.eMsgSent) + ) - disconnect(false); + disconnect(userInitiated = false) } - private void handleSessionToken(IPacketMsg packetMsg) { - ClientMsgProtobuf sessToken = new ClientMsgProtobuf<>(CMsgClientSessionToken.class, packetMsg); + private fun handleSessionToken(packetMsg: IPacketMsg) { + val sessToken = ClientMsgProtobuf( + CMsgClientSessionToken::class.java, + packetMsg + ) - sessionToken = sessToken.getBody().getToken(); + sessionToken = sessToken.body.token } // endregion - - public SteamConfiguration getConfiguration() { - return configuration; - } - - /** - * @return Bootstrap list of CM servers. - */ - public SmartCMServerList getServers() { - return configuration.getServerList(); - } - - /** - * Returns the local IP of this client. - * - * @return The local IP or null if no connection is available. - */ - public @Nullable InetAddress getLocalIP() { - var connection = this.connection; - if (connection == null) { - return null; - } - return connection.getLocalIP(); - } - - /** - * Returns the current endpoint this client is connected to. - * - * @return The current endpoint or null if no connection is available. - */ - public @Nullable InetSocketAddress getCurrentEndpoint() { - var connection = this.connection; - if (connection == null) { - return null; - } - return connection.getCurrentEndPoint(); - } - - /** - * Gets the public IP address of this client. This value is assigned after a logon attempt has succeeded. - * This value will be null if the client is logged off of Steam. - * - * @return The {@link InetSocketAddress} public ip - */ - public @Nullable InetAddress getPublicIP() { - return publicIP; - } - - /** - * Gets the country code of our public IP address according to Steam. This value is assigned after a logon attempt has succeeded. - * This value will be null if the client is logged off of Steam. - * - * @return the ip country code. - */ - public @Nullable String getIpCountryCode() { - return ipCountryCode; - } - - /** - * Gets the universe of this client. - * - * @return The universe. - */ - public EUniverse getUniverse() { - return configuration.getUniverse(); - } - - /** - * Gets a value indicating whether this instance is isConnected to the remote CM server. - * - * @return true if this instance is isConnected; otherwise, false. - */ - public boolean isConnected() { - return isConnected; - } - - /** - * Gets a value indicating whether isConnected and connection is not connected to the remote CM server. - * Inverse alternative to {@link CMClient#isConnected()} - * - * @return true is this instance is disconnected, otherwise, false. - */ - // > "since the client can technically not be connected but still have a connection" - public boolean isDisconnected() { - return !isConnected && connection == null; - } - - /** - * @return the session token assigned to this client from the AM. - */ - public long getSessionToken() { - return sessionToken; - } - - /** - * @return the Steam recommended Cell ID of this client. This value is assigned after a logon attempt has succeeded. - * This value will be null if the client is logged off of Steam. - */ - public @Nullable Integer getCellID() { - return cellID; - } - - /** - * Gets the session ID of this client. This value is assigned after a logon attempt has succeeded. - * This value will be null if the client is logged off of Steam. - * - * @return The session ID. - */ - public @Nullable Integer getSessionID() { - return sessionID; - } - - /** - * Gets the SteamID of this client. This value is assigned after a logon attempt has succeeded. - * This value will be null if the client is logged off of Steam. - * - * @return The SteamID. - */ - public @Nullable SteamID getSteamID() { - return steamID; - } - - /** - * Gets or sets the connection timeout used when connecting to the Steam server. - * - * @return The connection timeout. - */ - public long getConnectionTimeout() { - return configuration.getConnectionTimeout(); - } - - /** - * @return the network listening interface. Use this for debugging only. - * For your convenience, you can use {@link NetHookNetworkListener} class. - */ - public IDebugNetworkListener getDebugNetworkListener() { - return debugNetworkListener; - } - - /** - * Sets the network listening interface. Use this for debugging only. - * For your convenience, you can use {@link NetHookNetworkListener} class. - * - * @param debugNetworkListener the listener - */ - public void setDebugNetworkListener(IDebugNetworkListener debugNetworkListener) { - this.debugNetworkListener = debugNetworkListener; - } - - public boolean isExpectDisconnection() { - return expectDisconnection; - } - - public void setExpectDisconnection(boolean expectDisconnection) { - this.expectDisconnection = expectDisconnection; - } } diff --git a/src/main/java/in/dragonbra/javasteam/steam/handlers/ClientMsgHandler.kt b/src/main/java/in/dragonbra/javasteam/steam/handlers/ClientMsgHandler.kt index 93f65f69..5ace9369 100644 --- a/src/main/java/in/dragonbra/javasteam/steam/handlers/ClientMsgHandler.kt +++ b/src/main/java/in/dragonbra/javasteam/steam/handlers/ClientMsgHandler.kt @@ -26,9 +26,9 @@ abstract class ClientMsgHandler { * [DisconnectedCallback.isUserInitiated] property will be set to **true**. */ protected var isExpectDisconnection: Boolean - get() = client.isExpectDisconnection + get() = client.expectDisconnection set(expectDisconnection) { - client.isExpectDisconnection = expectDisconnection + client.expectDisconnection = expectDisconnection } /** diff --git a/src/main/java/in/dragonbra/javasteam/steam/steamclient/SteamClient.kt b/src/main/java/in/dragonbra/javasteam/steam/steamclient/SteamClient.kt index a873f743..8d734e6f 100644 --- a/src/main/java/in/dragonbra/javasteam/steam/steamclient/SteamClient.kt +++ b/src/main/java/in/dragonbra/javasteam/steam/steamclient/SteamClient.kt @@ -44,12 +44,15 @@ import java.util.concurrent.atomic.AtomicLong * * @constructor Initializes a new instance of the [SteamClient] class with a specific configuration. * @param configuration The configuration to use for this client. + * @param identifier A specific identifier to be used to uniquely identify this instance. + * @param defaultScope todo */ @Suppress("unused") class SteamClient @JvmOverloads constructor( - configuration: SteamConfiguration? = SteamConfiguration.createDefault(), + configuration: SteamConfiguration = SteamConfiguration.createDefault(), + identifier: String = UUID.randomUUID().toString().replace("-", ""), internal val defaultScope: CoroutineScope = CoroutineScope(Dispatchers.IO + SupervisorJob()), -) : CMClient(configuration) { +) : CMClient(configuration, identifier) { private val handlers = HashMap, ClientMsgHandler>(HANDLERS_COUNT) @@ -221,12 +224,14 @@ class SteamClient @JvmOverloads constructor( * Called when a client message is received from the network. * @param packetMsg The packet message. */ - override fun onClientMsgReceived(packetMsg: IPacketMsg): Boolean { + override fun onClientMsgReceived(packetMsg: IPacketMsg?): Boolean { // let the underlying CMClient handle this message first if (!super.onClientMsgReceived(packetMsg)) { return false } + requireNotNull(packetMsg) + // we want to handle some of the clientMsg's before we pass them along to registered handlers when (packetMsg.getMsgType()) { EMsg.JobHeartbeat -> handleJobHeartbeat(packetMsg) diff --git a/src/test/java/in/dragonbra/javasteam/steam/DummyClient.java b/src/test/java/in/dragonbra/javasteam/steam/DummyClient.java index 6efccce8..79f1e85e 100644 --- a/src/test/java/in/dragonbra/javasteam/steam/DummyClient.java +++ b/src/test/java/in/dragonbra/javasteam/steam/DummyClient.java @@ -9,7 +9,7 @@ */ public class DummyClient extends CMClient { public DummyClient() { - super(SteamConfiguration.createDefault()); + super(SteamConfiguration.createDefault(), "DummyClient"); } public void dummyDisconnect() { From 37d9a3aa1e6c292791c0bcfd19836eca4bf99a5f Mon Sep 17 00:00:00 2001 From: LossyDragon Date: Fri, 6 Jun 2025 00:17:59 -0500 Subject: [PATCH 09/11] Handle potential double disconnects in WebSocketConnection --- .../networking/steam3/WebSocketConnection.kt | 57 +++++++++++++------ .../in/dragonbra/javasteam/steam/CMClient.kt | 2 +- 2 files changed, 40 insertions(+), 19 deletions(-) diff --git a/src/main/java/in/dragonbra/javasteam/networking/steam3/WebSocketConnection.kt b/src/main/java/in/dragonbra/javasteam/networking/steam3/WebSocketConnection.kt index 8f648453..ef50c6a7 100644 --- a/src/main/java/in/dragonbra/javasteam/networking/steam3/WebSocketConnection.kt +++ b/src/main/java/in/dragonbra/javasteam/networking/steam3/WebSocketConnection.kt @@ -24,6 +24,8 @@ import kotlinx.coroutines.isActive import kotlinx.coroutines.launch import java.net.InetAddress import java.net.InetSocketAddress +import java.util.concurrent.CancellationException +import java.util.concurrent.atomic.AtomicBoolean import kotlin.coroutines.CoroutineContext import kotlin.time.DurationUnit import kotlin.time.toDuration @@ -46,6 +48,8 @@ class WebSocketConnection : private var lastFrameTime = System.currentTimeMillis() + private val isDisconnecting = AtomicBoolean(false) + override val coroutineContext: CoroutineContext = Dispatchers.IO + job override fun connect(endPoint: InetSocketAddress, timeout: Int) { @@ -53,6 +57,7 @@ class WebSocketConnection : logger.debug("Trying connection to ${endPoint.hostName}:${endPoint.port}") try { + isDisconnecting.set(false) endpoint = endPoint client = HttpClient(CIO) { @@ -90,6 +95,8 @@ class WebSocketConnection : is Frame.Text -> logger.debug("Received plain text ${frame.readText()}") } } + } catch (e: CancellationException) { + logger.debug("Frame listener was cancelled", e) } catch (e: Exception) { logger.error("An error occurred while receiving data", e) disconnect(false) @@ -106,7 +113,13 @@ class WebSocketConnection : } override fun disconnect(userInitiated: Boolean) { + if (!isDisconnecting.compareAndSet(false, true)) { + logger.error("Already disconnected or disconnecting") + return + } + logger.debug("Disconnect called: $userInitiated") + launch { try { session?.close() @@ -123,6 +136,10 @@ class WebSocketConnection : } override fun send(data: ByteArray) { + if (isDisconnecting.get()) { + return + } + launch { try { val frame = Frame.Binary(true, data) @@ -145,30 +162,34 @@ class WebSocketConnection : */ private fun startConnectionMonitoring() { launch { - while (isActive) { - if (client?.isActive == false || session?.isActive == false) { - logger.error("Client or Session is no longer active") - disconnect(userInitiated = false) - } - - val timeSinceLastFrame = System.currentTimeMillis() - lastFrameTime - - // logger.debug("Watchdog status: $timeSinceLastFrame") - when { - timeSinceLastFrame > 30000 -> { - logger.error("Watchdog: No response for 30 seconds. Disconnecting from steam") + try { + while (isActive && !isDisconnecting.get()) { + if (client?.isActive == false || session?.isActive == false) { + logger.error("Client or Session is no longer active") disconnect(userInitiated = false) - break } - timeSinceLastFrame > 25000 -> logger.debug("Watchdog: No response for 25 seconds") + val timeSinceLastFrame = System.currentTimeMillis() - lastFrameTime - timeSinceLastFrame > 20000 -> logger.debug("Watchdog: No response for 20 seconds") + // logger.debug("Watchdog status: $timeSinceLastFrame") + when { + timeSinceLastFrame > 30000 -> { + logger.error("Watchdog: No response for 30 seconds. Disconnecting from steam") + disconnect(userInitiated = false) + break + } - timeSinceLastFrame > 15000 -> logger.debug("Watchdog: No response for 15 seconds") - } + timeSinceLastFrame > 25000 -> logger.debug("Watchdog: No response for 25 seconds") - delay(5000) + timeSinceLastFrame > 20000 -> logger.debug("Watchdog: No response for 20 seconds") + + timeSinceLastFrame > 15000 -> logger.debug("Watchdog: No response for 15 seconds") + } + + delay(5000) + } + } catch (e: CancellationException) { + logger.debug("Watchdog Cancelled.", e) } } } diff --git a/src/main/java/in/dragonbra/javasteam/steam/CMClient.kt b/src/main/java/in/dragonbra/javasteam/steam/CMClient.kt index 30e94817..5a0599bd 100644 --- a/src/main/java/in/dragonbra/javasteam/steam/CMClient.kt +++ b/src/main/java/in/dragonbra/javasteam/steam/CMClient.kt @@ -306,7 +306,7 @@ constructor( */ fun send(msg: IClientMsg) { if (!isConnected) { - logger.error("Send() was called while not connected to Steam.") + logger.error("Send(${msg.msgType}) was called while not connected to Steam.") } sessionID?.let { msg.setSessionID(it) } From 0c4fb670fb7f7dd55bfd613f87089edd227fafae Mon Sep 17 00:00:00 2001 From: LossyDragon Date: Mon, 9 Jun 2025 15:55:36 -0500 Subject: [PATCH 10/11] log messages cleanup --- .../java/in/dragonbra/javasteam/steam/CMClient.kt | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/src/main/java/in/dragonbra/javasteam/steam/CMClient.kt b/src/main/java/in/dragonbra/javasteam/steam/CMClient.kt index 5a0599bd..38fe1e82 100644 --- a/src/main/java/in/dragonbra/javasteam/steam/CMClient.kt +++ b/src/main/java/in/dragonbra/javasteam/steam/CMClient.kt @@ -77,7 +77,7 @@ constructor( if (data.size < 4) { logger.debug( "PacketMsg too small to contain a message, was only ${data.size} bytes. " + - "Message: " + Strings.toHex(data) + "Message: ${Strings.toHex(data)}" ) return null } @@ -95,7 +95,7 @@ constructor( -> try { return PacketMsg(eMsg, data) } catch (e: IOException) { - logger.debug("Exception deserializing emsg " + eMsg + " (" + MsgUtil.isProtoBuf(rawEMsg) + ").", e) + logger.debug("Exception deserializing emsg $eMsg (${MsgUtil.isProtoBuf(rawEMsg)}).", e) } else -> Unit @@ -439,7 +439,7 @@ constructor( e: DisconnectedEventArgs, ) { logger.debug( - "EventHandler `disconnected` called. User Initiated: ${e.isUserInitiated}," + + "EventHandler `disconnected` called. User Initiated: ${e.isUserInitiated}, " + "Expected Disconnection: $expectDisconnection" ) isConnected = false @@ -575,7 +575,7 @@ constructor( ) } } else { - logger.debug("handleLoggedOff got unexpected response: " + packetMsg.getMsgType()) + logger.debug("handleLoggedOff got unexpected response: ${packetMsg.getMsgType()}") } } @@ -583,8 +583,8 @@ constructor( val msgServerUnavailable = ClientMsg(MsgClientServerUnavailable::class.java, packetMsg) logger.debug( - "A server of type " + msgServerUnavailable.body.eServerTypeUnavailable + - "was not available for request: " + EMsg.from(msgServerUnavailable.body.eMsgSent) + "A server of type ${msgServerUnavailable.body.eServerTypeUnavailable} " + + "was not available for request: ${EMsg.from(msgServerUnavailable.body.eMsgSent)}" ) disconnect(userInitiated = false) From 0c22c3f709991fc669b94f271dff5de8f2a85ba4 Mon Sep 17 00:00:00 2001 From: LossyDragon Date: Mon, 9 Jun 2025 21:07:51 -0500 Subject: [PATCH 11/11] More log messages cleanup. Tidy up kdoc and kotlin safe call operators. --- .../in/dragonbra/javasteam/steam/CMClient.kt | 24 ++++++++----------- 1 file changed, 10 insertions(+), 14 deletions(-) diff --git a/src/main/java/in/dragonbra/javasteam/steam/CMClient.kt b/src/main/java/in/dragonbra/javasteam/steam/CMClient.kt index 38fe1e82..5e84e18f 100644 --- a/src/main/java/in/dragonbra/javasteam/steam/CMClient.kt +++ b/src/main/java/in/dragonbra/javasteam/steam/CMClient.kt @@ -27,6 +27,8 @@ import `in`.dragonbra.javasteam.steam.discovery.ServerQuality import `in`.dragonbra.javasteam.steam.discovery.ServerRecord import `in`.dragonbra.javasteam.steam.discovery.SmartCMServerList import `in`.dragonbra.javasteam.steam.steamclient.SteamClient +import `in`.dragonbra.javasteam.steam.steamclient.callbacks.ConnectedCallback +import `in`.dragonbra.javasteam.steam.steamclient.callbacks.DisconnectedCallback import `in`.dragonbra.javasteam.steam.steamclient.configuration.SteamConfiguration import `in`.dragonbra.javasteam.types.SteamID import `in`.dragonbra.javasteam.util.IDebugNetworkListener @@ -238,8 +240,8 @@ constructor( /** * Connects this client to a Steam3 server. This begins the process of connecting and encrypting the data channel - * between the client and the server. Results are returned asynchronously in a [ConnectedCallback][in.dragonbra.javasteam.steam.steamclient.callbacks.ConnectedCallback]. If the - * server that SteamKit attempts to connect to is down, a [DisconnectedCallback][in.dragonbra.javasteam.steam.steamclient.callbacks.DisconnectedCallback] will be posted instead. + * between the client and the server. Results are returned asynchronously in a [ConnectedCallback]. If the + * server that SteamKit attempts to connect to is down, a [DisconnectedCallback] will be posted instead. * SteamKit will not attempt to reconnect to Steam, you must handle this callback and call Connect again preferably * after a short delay. * @@ -271,12 +273,8 @@ constructor( connection!!.getConnected().addEventHandler(connected) connection!!.getDisconnected().addEventHandler(disconnected) logger.debug( - String.format( - "Connecting to %s with protocol %s, and with connection impl %s", - cmServer.endpoint, - cmServer.protocolTypes, - connection!!.javaClass.getSimpleName() - ) + "Connecting to ${cmServer.endpoint} with protocol ${cmServer.protocolTypes}, " + + "and with connection impl ${connection!!.javaClass.getSimpleName()}", ) connection!!.connect(cmServer.endpoint) } catch (e: Exception) { @@ -293,9 +291,7 @@ constructor( fun disconnect(userInitiated: Boolean = true) { synchronized(connectionLock) { heartBeatFunc.stop() - if (connection != null) { - connection!!.disconnect(userInitiated) - } + connection?.disconnect(userInitiated) } } @@ -382,7 +378,7 @@ constructor( val connectionFactory: IConnectionFactory = configuration.connectionFactory val connection = connectionFactory.createConnection(configuration, protocol) if (connection == null) { - logger.error(String.format("Connection factory returned null connection for protocols %s", protocol)) + logger.error("Connection factory returned null connection for protocols $protocol") throw IllegalArgumentException("Connection factory returned null connection.") } return connection @@ -418,8 +414,8 @@ constructor( } servers.tryMark( - endPoint = connection!!.getCurrentEndPoint(), - protocolTypes = connection!!.getProtocolTypes(), + endPoint = connection?.getCurrentEndPoint(), + protocolTypes = connection?.getProtocolTypes(), quality = ServerQuality.GOOD )