Update to jdk 21

.github/workflows/tag.yml

@@ -12,10 +12,10 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
-      - name: set up JDK 17
+      - name: set up JDK
         uses: actions/setup-java@v4
         with:
-          java-version: '17'
+          java-version: '21'
           distribution: 'temurin'
           cache: gradle
       - name: Grant execute permission for gradlew

.github/workflows/test.yml

@@ -22,7 +22,7 @@ jobs:
       - name: Set up JDK
         uses: actions/setup-java@v4
         with:
-          java-version: 17
+          java-version: 21
           distribution: temurin
       - name: Grant permissions for user space Icmp
         run: sudo sysctl -w net.ipv4.ping_group_range="0   2147483647"

android/build.gradle.kts

@@ -31,11 +31,11 @@ android {
         }
     }
     compileOptions {
-        sourceCompatibility = JavaVersion.VERSION_11
-        targetCompatibility = JavaVersion.VERSION_11
+        sourceCompatibility = JavaVersion.VERSION_21
+        targetCompatibility = JavaVersion.VERSION_21
     }
     kotlinOptions {
-        jvmTarget = "11"
+        jvmTarget = "21"
     }
     packaging {
         resources {

client/build.gradle.kts

@@ -21,12 +21,12 @@ tasks.withType<Test>().configureEach {
 }
 
 java {
-    sourceCompatibility = JavaVersion.VERSION_17
-    targetCompatibility = JavaVersion.VERSION_17
+    sourceCompatibility = JavaVersion.VERSION_21
+    targetCompatibility = JavaVersion.VERSION_21
 }
 
 kotlin {
-    jvmToolchain(17)
+    jvmToolchain(21)
 }
 
 jacoco {

core/build.gradle.kts

@@ -9,12 +9,12 @@ plugins {
 }
 
 java {
-    sourceCompatibility = JavaVersion.VERSION_17
-    targetCompatibility = JavaVersion.VERSION_17
+    sourceCompatibility = JavaVersion.VERSION_21
+    targetCompatibility = JavaVersion.VERSION_21
 }
 
 kotlin {
-    jvmToolchain(17)
+    jvmToolchain(21)
 }
 
 tasks.jacocoTestReport {

core/src/main/kotlin/com/jasonernst/kanonproxy/KAnonProxy.kt

@@ -31,6 +31,7 @@ import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.withContext
+import org.jetbrains.annotations.TestOnly
 import org.slf4j.LoggerFactory
 import java.net.Inet4Address
 import java.net.Inet6Address
@@ -467,4 +468,10 @@ class KAnonProxy(
         incomingQueue.clear()
         logger.debug("KAnonProxy stopped")
     }
+
+    @TestOnly
+    fun disconnectClient(clientAddress: InetSocketAddress) {
+        val outgoingQueue = outgoingQueues[clientAddress]
+        outgoingQueue?.add(SentinelPacket)
+    }
 }

core/src/main/kotlin/com/jasonernst/kanonproxy/tcp/TcpStateMachine.kt

@@ -29,6 +29,7 @@ import java.nio.channels.SelectionKey.OP_READ
 import java.nio.channels.SocketChannel
 import java.util.ArrayList
 import java.util.concurrent.ConcurrentLinkedQueue
+import java.util.concurrent.atomic.AtomicBoolean
 import kotlin.jvm.javaClass
 import kotlin.math.abs
 import kotlin.math.max
@@ -60,6 +61,7 @@ class TcpStateMachine(
     val swapSourceDestination: Boolean = false, // only time we want this is if we're a tcpClient
 ) {
     private val logger = LoggerFactory.getLogger(javaClass)
+    private val isRunning = AtomicBoolean(true)
     private var delayedAckJob: Job? = null
     private var latestAck: Packet? = null
 
@@ -1209,7 +1211,9 @@ class TcpStateMachine(
         //         (for the current value of RTO).
         rtoJob?.cancel()
         rtoJob = null
-        startRtoTimer()
+        if (isRunning.get()) {
+            startRtoTimer()
+        }
 
         while (retransmitQueue.isNotEmpty()) {
             // may be null if the session is shutting down
@@ -2129,7 +2133,7 @@ class TcpStateMachine(
      *        taken per (2.2) rather than using (2.3).
      */
     private fun startRtoTimer() {
-        if (rtoJob == null) {
+        if (rtoJob == null && isRunning.get()) {
             rtoJob =
                 CoroutineScope(Dispatchers.IO).launch {
                     Thread.currentThread().name = "RTO: ${session.getKey()}"
@@ -2179,7 +2183,7 @@ class TcpStateMachine(
 
     private fun setupLatestAckJob(ack: Packet) {
         latestAck = ack
-        if (delayedAckJob == null) {
+        if (delayedAckJob == null && isRunning.get()) {
             delayedAckJob =
                 CoroutineScope(Dispatchers.IO).launch {
                     Thread.currentThread().name = "latestACK: ${session.getKey()}"
@@ -2289,6 +2293,7 @@ class TcpStateMachine(
     }
 
     fun stopJobs() {
+        isRunning.set(false)
         rtoJob?.cancel()
         timeWaitJob?.cancel()
         delayedAckJob?.cancel()

core/src/test/kotlin/com/jasonernst/kanonproxy/tcp/TcpClient.kt

@@ -199,6 +199,7 @@ class TcpClient(
             } catch (e: Exception) {
                 logger.error("Failed to connect: {}, last state: {}", e.message, tcpStateMachine.tcpState.value)
                 isRunning.set(false)
+                kAnonProxy.disconnectClient(session.clientAddress)
                 kAnonProxy.removeSession(session)
                 throw e
             }
@@ -311,7 +312,11 @@ class TcpClient(
                         }
 
                     flow.collect {
-                        logger.debug("Waiting for CLOSED: {} State: {}", clientId, it)
+                        if (waitForTimeWait) {
+                            logger.debug("Waiting for CLOSED: {} State: {}", clientId, it)
+                        } else {
+                            logger.debug("Waiting for CLOSED or TIME_WAIT: {} State: {}", clientId, it)
+                        }
                     }
                 }
             } catch (e: Exception) {
@@ -339,6 +344,7 @@ class TcpClient(
             isRunning.set(false)
             returnQueue.add(SentinelPacket)
             if (!waitForTimeWait) {
+                kAnonProxy.disconnectClient(session.clientAddress)
                 kAnonProxy.removeSession(session)
             }
             logger.debug("Waiting for readjob to finish")
@@ -385,6 +391,7 @@ class TcpClient(
         val session = this
         runBlocking {
             logger.debug("Waiting for readjob to stop")
+            kAnonProxy.disconnectClient(session.clientAddress)
             kAnonProxy.removeSession(session)
             readJob.cancelAndJoin()
             logger.debug("readjob stopped. Waiting for writejob to stop")