Refactor SimpleDataSource into an interface, add AutoCloseable support, and encapsulate implementation details in a factory method.

Author: rossdanderson

util/src/main/kotlin/com/caplin/integration/datasourcex/util/SimpleDataSource.kt

@@ -5,45 +5,65 @@ import com.caplin.datasource.Peer
 import com.caplin.datasource.PeerStatus
 import com.caplin.datasource.PeerStatus.DOWN
 import com.caplin.datasource.PeerStatus.UP
+import java.lang.AutoCloseable
 import java.util.concurrent.locks.ReentrantLock
 import kotlin.concurrent.withLock
 import kotlinx.collections.immutable.toPersistentMap
 import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.SharedFlow
 import kotlinx.coroutines.flow.asSharedFlow
 import kotlinx.coroutines.flow.first
 
-class SimpleDataSource(private val dataSource: DataSource) : DataSource by dataSource {
+interface SimpleDataSource : DataSource, AutoCloseable {
 
-  private val _state =
-      MutableSharedFlow<Map<Peer, PeerStatus>>(replay = 1, extraBufferCapacity = Int.MAX_VALUE)
+  companion object {
+    operator fun invoke(dataSource: DataSource): SimpleDataSource =
+        object : SimpleDataSource, DataSource by dataSource {
 
-  init {
-    var knownPeers = dataSource.peers.associate { PeerKey(it) as Peer to DOWN }.toPersistentMap()
-    _state.tryEmit(knownPeers)
-    val lock = ReentrantLock()
-    dataSource.addConnectionListener {
-      lock.withLock {
-        knownPeers = knownPeers.putting(PeerKey(it.peer), it.peerStatus)
-        _state.tryEmit(knownPeers)
-      }
-    }
-  }
+          private val _state =
+              MutableSharedFlow<Map<Peer, PeerStatus>>(
+                  replay = 1,
+                  extraBufferCapacity = Int.MAX_VALUE,
+              )
 
-  val peerStates = _state.asSharedFlow()
+          init {
+            var knownPeers =
+                dataSource.peers.associate { PeerKey(it) as Peer to DOWN }.toPersistentMap()
+            _state.tryEmit(knownPeers)
+            val lock = ReentrantLock()
+            dataSource.addConnectionListener {
+              lock.withLock {
+                knownPeers = knownPeers.putting(PeerKey(it.peer), it.peerStatus)
+                _state.tryEmit(knownPeers)
+              }
+            }
+          }
 
-  /** Await all known peers to be connected */
-  suspend fun awaitConnected() {
-    _state.first { it.values.all { it == UP } }
-  }
+          override val peerStates = _state.asSharedFlow()
+
+          /** Await all known peers to be connected */
+          override suspend fun awaitConnected() {
+            _state.first { it.values.all { it == UP } }
+          }
 
-  /** A [Peer] whose identity is its index, so it can be used as a map key. */
-  private class PeerKey(peer: Peer) : Peer by peer {
-    override fun equals(other: Any?) = other is PeerKey && other.index == index
+          override fun close() {
+            dataSource.stop()
+          }
+        }
 
-    override fun hashCode() = index
+    /** A [Peer] whose identity is its index, so it can be used as a map key. */
+    private class PeerKey(peer: Peer) : Peer by peer {
+      override fun equals(other: Any?) = other is PeerKey && other.index == index
 
-    override fun getLoggingId(): String? {
-      return super.getLoggingId()
+      override fun hashCode() = index
+
+      override fun getLoggingId(): String? {
+        return super.getLoggingId()
+      }
     }
   }
+
+  val peerStates: SharedFlow<Map<Peer, PeerStatus>>
+
+  suspend fun awaitConnected()
 }