Skip to content

Commit 335e1b2

Browse files
rossdandersonrossdanderson
committed
Add utility methods to wait for DataSource connection
1 parent 39d7091 commit 335e1b2

3 files changed

Lines changed: 91 additions & 5 deletions

File tree

‎util/api/datasourcex-util.api‎

Lines changed: 35 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -84,6 +84,39 @@ public final class com/caplin/integration/datasourcex/util/ReadWriteLock {
8484
public final fun writeUnlock ()V
8585
}
8686

87+
public final class com/caplin/integration/datasourcex/util/SimpleDataSource : com/caplin/datasource/DataSource {
88+
public fun <init> (Lcom/caplin/datasource/DataSource;)V
89+
public fun addChannelListener (Lcom/caplin/datasource/namespace/Namespace;Lcom/caplin/datasource/channel/ChannelListener;)V
90+
public fun addConnectionListener (Lcom/caplin/datasource/ConnectionListener;)V
91+
public fun addGenericChannelListener (Lcom/caplin/datasource/namespace/Namespace;Lcom/caplin/datasource/channel/ChannelListener;)V
92+
public fun addJsonChannelListener (Lcom/caplin/datasource/namespace/Namespace;Lcom/caplin/datasource/channel/JsonChannelListener;)V
93+
public final fun awaitConnected (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
94+
public fun createActiveJsonSubscription (Ljava/lang/String;Lcom/caplin/datasource/subscription/SubjectConsumer;Ljava/lang/Class;)Lcom/caplin/datasource/subscription/ActiveSubscription;
95+
public fun createActivePublisher (Lcom/caplin/datasource/namespace/Namespace;Lcom/caplin/datasource/publisher/DataProvider;)Lcom/caplin/datasource/publisher/ActivePublisher;
96+
public fun createActiveSubscription (Ljava/lang/String;Lcom/caplin/datasource/subscription/SubscriptionListener;)Lcom/caplin/datasource/subscription/ActiveSubscription;
97+
public fun createBroadcastJsonSubscription (Lcom/caplin/datasource/namespace/Namespace;Lcom/caplin/datasource/subscription/SubjectConsumer;Ljava/lang/Class;)Lcom/caplin/datasource/subscription/BroadcastSubscription;
98+
public fun createBroadcastPublisher (Lcom/caplin/datasource/namespace/Namespace;)Lcom/caplin/datasource/publisher/BroadcastPublisher;
99+
public fun createBroadcastSubscription (Lcom/caplin/datasource/namespace/Namespace;Lcom/caplin/datasource/subscription/SubscriptionListener;)Lcom/caplin/datasource/subscription/BroadcastSubscription;
100+
public fun createCachingPublisher (Lcom/caplin/datasource/namespace/Namespace;Lcom/caplin/datasource/publisher/CachingDataProvider;)Lcom/caplin/datasource/publisher/CachingPublisher;
101+
public fun createCachingPublisher (Ljava/lang/String;Lcom/caplin/datasource/namespace/Namespace;Lcom/caplin/datasource/publisher/CachingDataProvider;)Lcom/caplin/datasource/publisher/CachingPublisher;
102+
public fun createCompatibilityPublisher (Lcom/caplin/datasource/namespace/Namespace;Lcom/caplin/datasource/publisher/DataProvider;)Lcom/caplin/datasource/publisher/CompatibilityPublisher;
103+
public fun createService (Lcom/caplin/datasource/Service;)V
104+
public fun getCommandOption (Ljava/lang/String;[Ljava/lang/String;)Ljava/lang/String;
105+
public fun getConfiguration ()Lcom/caplin/datasource/DataSourceConfiguration;
106+
public fun getExtraConfiguration ()Lcom/caplin/datasource/Configuration;
107+
public fun getFieldManager ()Lcom/caplin/datasource/fields/FieldManager;
108+
public fun getLogger ()Ljava/util/logging/Logger;
109+
public fun getMBeanServer ()Ljavax/management/MBeanServer;
110+
public final fun getPeerStates ()Lkotlinx/coroutines/flow/SharedFlow;
111+
public fun getPeers ()Ljava/util/Set;
112+
public fun getPeersInOrder ()Ljava/util/List;
113+
public fun removeConnectionListener (Lcom/caplin/datasource/ConnectionListener;)V
114+
public fun setStatusDown ()V
115+
public fun setStatusUp ()V
116+
public fun start ()V
117+
public fun stop ()V
118+
}
119+
87120
public abstract interface class com/caplin/integration/datasourcex/util/SimpleDataSourceConfig {
88121
public abstract fun getExtraConfig ()Ljava/lang/String;
89122
public abstract fun getLocalLabel ()Ljava/lang/String;
@@ -134,8 +167,8 @@ public final class com/caplin/integration/datasourcex/util/SimpleDataSourceConfi
134167

135168
public final class com/caplin/integration/datasourcex/util/SimpleDataSourceFactory {
136169
public static final field INSTANCE Lcom/caplin/integration/datasourcex/util/SimpleDataSourceFactory;
137-
public static final fun createDataSource (Lcom/caplin/integration/datasourcex/util/SimpleDataSourceConfig;Lcom/caplin/datasource/messaging/json/JsonHandler;)Lcom/caplin/datasource/DataSource;
138-
public static synthetic fun createDataSource$default (Lcom/caplin/integration/datasourcex/util/SimpleDataSourceConfig;Lcom/caplin/datasource/messaging/json/JsonHandler;ILjava/lang/Object;)Lcom/caplin/datasource/DataSource;
170+
public static final fun createDataSource (Lcom/caplin/integration/datasourcex/util/SimpleDataSourceConfig;Lcom/caplin/datasource/messaging/json/JsonHandler;)Lcom/caplin/integration/datasourcex/util/SimpleDataSource;
171+
public static synthetic fun createDataSource$default (Lcom/caplin/integration/datasourcex/util/SimpleDataSourceConfig;Lcom/caplin/datasource/messaging/json/JsonHandler;ILjava/lang/Object;)Lcom/caplin/integration/datasourcex/util/SimpleDataSource;
139172
public static final fun createJackson2JsonHandler (Lcom/fasterxml/jackson/databind/ObjectMapper;)Lcom/caplin/integration/datasourcex/util/serialization/jackson2/Jackson2JsonHandler;
140173
public static final fun createJackson3JsonHandler (Ltools/jackson/databind/ObjectMapper;)Lcom/caplin/integration/datasourcex/util/serialization/jackson3/Jackson3JsonHandler;
141174
public final fun getDefaultJackson2JsonHandler ()Lcom/caplin/integration/datasourcex/util/serialization/jackson2/Jackson2JsonHandler;

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

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,49 @@
1+
package com.caplin.integration.datasourcex.util
2+
3+
import com.caplin.datasource.DataSource
4+
import com.caplin.datasource.Peer
5+
import com.caplin.datasource.PeerStatus
6+
import com.caplin.datasource.PeerStatus.DOWN
7+
import com.caplin.datasource.PeerStatus.UP
8+
import java.util.concurrent.locks.ReentrantLock
9+
import kotlin.concurrent.withLock
10+
import kotlinx.collections.immutable.toPersistentMap
11+
import kotlinx.coroutines.flow.MutableSharedFlow
12+
import kotlinx.coroutines.flow.asSharedFlow
13+
import kotlinx.coroutines.flow.first
14+
15+
class SimpleDataSource(private val dataSource: DataSource) : DataSource by dataSource {
16+
17+
private val _state =
18+
MutableSharedFlow<Map<Peer, PeerStatus>>(replay = 1, extraBufferCapacity = Int.MAX_VALUE)
19+
20+
init {
21+
var knownPeers = dataSource.peers.associate { PeerKey(it) as Peer to DOWN }.toPersistentMap()
22+
_state.tryEmit(knownPeers)
23+
val lock = ReentrantLock()
24+
dataSource.addConnectionListener {
25+
lock.withLock {
26+
knownPeers = knownPeers.putting(PeerKey(it.peer), it.peerStatus)
27+
_state.tryEmit(knownPeers)
28+
}
29+
}
30+
}
31+
32+
val peerStates = _state.asSharedFlow()
33+
34+
/** Await all known peers to be connected */
35+
suspend fun awaitConnected() {
36+
_state.first { it.values.all { it == UP } }
37+
}
38+
39+
/** A [Peer] whose identity is its index, so it can be used as a map key. */
40+
private class PeerKey(peer: Peer) : Peer by peer {
41+
override fun equals(other: Any?) = other is PeerKey && other.index == index
42+
43+
override fun hashCode() = index
44+
45+
override fun getLoggingId(): String? {
46+
return super.getLoggingId()
47+
}
48+
}
49+
}

‎util/src/main/kotlin/com/caplin/integration/datasourcex/util/SimpleDataSourceFactory.kt‎

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,8 @@ package com.caplin.integration.datasourcex.util
22

33
import com.caplin.datasource.DataSource
44
import com.caplin.datasource.messaging.json.JsonHandler
5+
import com.caplin.integration.datasourcex.util.SimpleDataSourceFactory.defaultJackson3JsonHandler
6+
import com.caplin.integration.datasourcex.util.SimpleDataSourceFactory.defaultJackson3ObjectMapper
57
import com.caplin.integration.datasourcex.util.serialization.jackson2.Jackson2JsonHandler
68
import com.caplin.integration.datasourcex.util.serialization.jackson2.registerDataSourceModule
79
import com.caplin.integration.datasourcex.util.serialization.jackson3.Jackson3JsonHandler
@@ -83,7 +85,7 @@ object SimpleDataSourceFactory {
8385
fun createDataSource(
8486
simpleConfig: SimpleDataSourceConfig,
8587
jsonHandler: JsonHandler<*> = defaultJackson3JsonHandler,
86-
): DataSource {
88+
): SimpleDataSource {
8789
val logPath =
8890
simpleConfig.logDirectory
8991
?: run {
@@ -165,7 +167,9 @@ object SimpleDataSourceFactory {
165167
"""
166168
.trimMargin()
167169

168-
return DataSource.fromConfigString(config, Logger.getLogger(DataSource::class.qualifiedName))
169-
.apply { extraConfiguration.jsonHandler = jsonHandler }
170+
return SimpleDataSource(
171+
DataSource.fromConfigString(config, Logger.getLogger(DataSource::class.qualifiedName))
172+
.apply { extraConfiguration.jsonHandler = jsonHandler },
173+
)
170174
}
171175
}

0 commit comments

Comments
 (0)