refactor(simulator): Remove failure callback from FlowSource

This change removes the `onFailure` method from FlowSource. Instead, the
FlowConsumer will receive the reason for failure of the source.

4 files changed, 98 insertions, 33 deletions

diff --git a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowConsumerContextTest.kt b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowConsumerContextTest.kt
index 380fd38a..f1a5cbe4 100644
--- a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowConsumerContextTest.kt
+++ b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowConsumerContextTest.kt
@@ -23,7 +23,6 @@
 package org.opendc.simulator.flow
 
 import io.mockk.*
-import kotlinx.coroutines.*
 import org.junit.jupiter.api.*
 import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.flow.internal.FlowConsumerContextImpl
@@ -102,34 +101,4 @@ class FlowConsumerContextTest {
 
         verify(exactly = 0) { consumer.onEvent(any(), any(), FlowEvent.Capacity) }
     }
-
-    @Test
-    fun testFailureNoInfiniteLoop() = runBlockingSimulation {
-        val engine = FlowEngineImpl(coroutineContext, clock)
-
-        val consumer = spyk(object : FlowSource {
-            override fun onPull(conn: FlowConnection, now: Long, delta: Long): Long {
-                conn.close()
-                return Long.MAX_VALUE
-            }
-
-            override fun onEvent(conn: FlowConnection, now: Long, event: FlowEvent) {
-                if (event == FlowEvent.Exit) throw IllegalStateException("onEvent")
-            }
-
-            override fun onFailure(conn: FlowConnection, cause: Throwable) {
-                throw IllegalStateException("onFailure")
-            }
-        })
-
-        val logic = object : FlowConsumerLogic {}
-
-        val context = FlowConsumerContextImpl(engine, consumer, logic)
-
-        context.start()
-
-        delay(1)
-
-        verify(exactly = 1) { consumer.onFailure(any(), any()) }
-    }
 }
diff --git a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowForwarderTest.kt b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowForwarderTest.kt
index cbc48a4e..d125c638 100644
--- a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowForwarderTest.kt
+++ b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowForwarderTest.kt
@@ -219,4 +219,100 @@ internal class FlowForwarderTest {
         assertEquals(source.counters.overcommit, forwarder.counters.overcommit) { "Overcommitted work" }
         assertEquals(2000, clock.millis())
     }
+
+    @Test
+    fun testCoupledExit() = runBlockingSimulation {
+        val engine = FlowEngineImpl(coroutineContext, clock)
+        val forwarder = FlowForwarder(engine, isCoupled = true)
+        val source = FlowSink(engine, 2000.0)
+
+        launch { source.consume(forwarder) }
+
+        forwarder.consume(FixedFlowSource(2000.0, 1.0))
+
+        yield()
+
+        assertFalse(source.isActive)
+    }
+
+    @Test
+    fun testPullFailureCoupled() = runBlockingSimulation {
+        val engine = FlowEngineImpl(coroutineContext, clock)
+        val forwarder = FlowForwarder(engine, isCoupled = true)
+        val source = FlowSink(engine, 2000.0)
+
+        launch { source.consume(forwarder) }
+
+        try {
+            forwarder.consume(object : FlowSource {
+                override fun onPull(conn: FlowConnection, now: Long, delta: Long): Long {
+                    throw IllegalStateException("Test")
+                }
+            })
+        } catch (cause: Throwable) {
+            // Ignore
+        }
+
+        yield()
+
+        assertFalse(source.isActive)
+    }
+
+    @Test
+    fun testEventFailure() = runBlockingSimulation {
+        val engine = FlowEngineImpl(coroutineContext, clock)
+        val forwarder = FlowForwarder(engine)
+        val source = FlowSink(engine, 2000.0)
+
+        launch { source.consume(forwarder) }
+
+        try {
+            forwarder.consume(object : FlowSource {
+                override fun onPull(conn: FlowConnection, now: Long, delta: Long): Long {
+                    return Long.MAX_VALUE
+                }
+
+                override fun onEvent(conn: FlowConnection, now: Long, event: FlowEvent) {
+                    throw IllegalStateException("Test")
+                }
+            })
+        } catch (cause: Throwable) {
+            // Ignore
+        }
+
+        yield()
+
+        assertTrue(source.isActive)
+        source.cancel()
+    }
+
+    @Test
+    fun testEventConvergeFailure() = runBlockingSimulation {
+        val engine = FlowEngineImpl(coroutineContext, clock)
+        val forwarder = FlowForwarder(engine)
+        val source = FlowSink(engine, 2000.0)
+
+        launch { source.consume(forwarder) }
+
+        try {
+            forwarder.consume(object : FlowSource {
+                override fun onPull(conn: FlowConnection, now: Long, delta: Long): Long {
+                    return Long.MAX_VALUE
+                }
+
+                override fun onEvent(conn: FlowConnection, now: Long, event: FlowEvent) {
+                    if (event == FlowEvent.Converge) {
+                        throw IllegalStateException("Test")
+                    }
+                }
+            })
+        } catch (cause: Throwable) {
+            // Ignore
+        }
+
+        yield()
+
+        assertTrue(source.isActive)
+        source.cancel()
+    }
 }
diff --git a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/mux/SimResourceSwitchExclusiveTest.kt b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/mux/ExclusiveFlowMultiplexerTest.kt
index b503087e..c8627446 100644
--- a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/mux/SimResourceSwitchExclusiveTest.kt
+++ b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/mux/ExclusiveFlowMultiplexerTest.kt
@@ -37,7 +37,7 @@ import org.opendc.simulator.flow.source.TraceFlowSource
 /**
  * Test suite for the [ForwardingFlowMultiplexer] class.
  */
-internal class SimResourceSwitchExclusiveTest {
+internal class ExclusiveFlowMultiplexerTest {
     /**
      * Test a trace workload.
      */
diff --git a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/mux/SimResourceSwitchMaxMinTest.kt b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/mux/MaxMinFlowMultiplexerTest.kt
index 089a8d78..9f6b8a2c 100644
--- a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/mux/SimResourceSwitchMaxMinTest.kt
+++ b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/mux/MaxMinFlowMultiplexerTest.kt
@@ -37,7 +37,7 @@ import org.opendc.simulator.flow.source.TraceFlowSource
 /**
  * Test suite for the [FlowMultiplexer] implementations
  */
-internal class SimResourceSwitchMaxMinTest {
+internal class MaxMinFlowMultiplexerTest {
     @Test
     fun testSmoke() = runBlockingSimulation {
         val scheduler = FlowEngineImpl(coroutineContext, clock)