ExternalVendorCode/corda
1
0
Fork 0
mirror of https://github.com/corda/corda.git synced 2025-06-13 12:48:18 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Merge branch 'os-4.8-merge-point' into shams-4.9-merge-0b8536b9

Browse Source 
# Conflicts:
#	.github/workflows/check-pr-title.yml
#	node/src/main/kotlin/net/corda/node/services/messaging/NodeNettyAcceptorFactory.kt
This commit is contained in:
Shams Asari
2023-07-07 15:20:43 +01:00
parent 2246c94fd5 0b8536b97c
commit 671ce9ef97
36 changed files with 458 additions and 269 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
node-api/src/main/kotlin/net/corda/nodeapi/internal/ArtemisMessagingClient.kt
View File

@ -42,8 +42,8 @@ class ArtemisMessagingClient(private val config: MutualSslConfiguration,
override fun start(): Started = synchronized(this) {
check(started == null) { "start can't be called twice" }
val tcpTransport = p2pConnectorTcpTransport(serverAddress, config, threadPoolName = threadPoolName, trace = trace)
val backupTransports = backupServerAddressPool.map {
p2pConnectorTcpTransport(it, config, threadPoolName = threadPoolName, trace = trace)
val backupTransports = backupServerAddressPool.mapIndexed { index, address ->
p2pConnectorTcpTransport(address, config, threadPoolName = "$threadPoolName-backup${index+1}", trace = trace)
}
log.info("Connecting to message broker: $serverAddress")

11
node-api/src/main/kotlin/net/corda/nodeapi/internal/ArtemisTcpTransport.kt
View File

@ -122,6 +122,7 @@ class ArtemisTcpTransport {
fun rpcAcceptorTcpTransport(hostAndPort: NetworkHostAndPort,
config: BrokerRpcSslOptions?,
enableSSL: Boolean = true,
threadPoolName: String = "RPCServer",
trace: Boolean = false,
remotingThreads: Int? = null): TransportConfiguration {
val options = mutableMapOf<String, Any>()
@ -129,7 +130,7 @@ class ArtemisTcpTransport {
config.keyStorePath.requireOnDefaultFileSystem()
options.putAll(config.toTransportOptions())
}
return createAcceptorTransport(hostAndPort, RPC_PROTOCOLS, options, null, enableSSL, "RPCServer", trace, remotingThreads)
return createAcceptorTransport(hostAndPort, RPC_PROTOCOLS, options, null, enableSSL, threadPoolName, trace, remotingThreads)
}
fun rpcConnectorTcpTransport(hostAndPort: NetworkHostAndPort,
@ -147,14 +148,16 @@ class ArtemisTcpTransport {
fun rpcInternalClientTcpTransport(hostAndPort: NetworkHostAndPort,
config: SslConfiguration,
threadPoolName: String = "Internal-RPCClient",
trace: Boolean = false): TransportConfiguration {
val options = mutableMapOf<String, Any>()
config.addToTransportOptions(options)
return createConnectorTransport(hostAndPort, RPC_PROTOCOLS, options, true, "Internal-RPCClient", trace, null)
return createConnectorTransport(hostAndPort, RPC_PROTOCOLS, options, true, threadPoolName, trace, null)
}
fun rpcInternalAcceptorTcpTransport(hostAndPort: NetworkHostAndPort,
config: SslConfiguration,
threadPoolName: String = "Internal-RPCServer",
trace: Boolean = false,
remotingThreads: Int? = null): TransportConfiguration {
val options = mutableMapOf<String, Any>()
@ -165,7 +168,7 @@ class ArtemisTcpTransport {
options,
trustManagerFactory(requireNotNull(config.trustStore).get()),
true,
"Internal-RPCServer",
threadPoolName,
trace,
remotingThreads
)
@ -210,7 +213,7 @@ class ArtemisTcpTransport {
options[TransportConstants.VERIFY_HOST_PROP_NAME] = false
}
return createTransport(
NodeNettyConnectorFactory::class.java.name,
CordaNettyConnectorFactory::class.java.name,
hostAndPort,
protocols,
options,

32
node-api/src/main/kotlin/net/corda/nodeapi/internal/ArtemisUtils.kt
View File

@ -1,8 +1,14 @@
@file:JvmName("ArtemisUtils")
package net.corda.nodeapi.internal
import net.corda.core.internal.declaredField
import org.apache.activemq.artemis.utils.actors.ProcessorBase
import java.nio.file.FileSystems
import java.nio.file.Path
import java.util.concurrent.Executor
import java.util.concurrent.ThreadFactory
import java.util.concurrent.ThreadPoolExecutor
import java.util.concurrent.atomic.AtomicInteger
/**
* Require that the [Path] is on a default file system, and therefore is one that Artemis is willing to use.
@ -16,3 +22,29 @@ fun requireMessageSize(messageSize: Int, limit: Int) {
require(messageSize <= limit) { "Message exceeds maxMessageSize network parameter, maxMessageSize: [$limit], message size: [$messageSize]" }
}
val Executor.rootExecutor: Executor get() {
var executor: Executor = this
while (executor is ProcessorBase<*>) {
executor = executor.declaredField<Executor>("delegate").value
}
return executor
}
fun Executor.setThreadPoolName(threadPoolName: String) {
(rootExecutor as? ThreadPoolExecutor)?.let { it.threadFactory = NamedThreadFactory(threadPoolName, it.threadFactory) }
}
private class NamedThreadFactory(poolName: String, private val delegate: ThreadFactory) : ThreadFactory {
companion object {
private val poolId = AtomicInteger(0)
}
private val prefix = "$poolName-${poolId.incrementAndGet()}-"
private val nextId = AtomicInteger(0)
override fun newThread(r: Runnable): Thread {
val thread = delegate.newThread(r)
thread.name = "$prefix${nextId.incrementAndGet()}"
return thread
}
}

22
node-api/src/main/kotlin/net/corda/nodeapi/internal/NodeNettyConnectorFactory.kt → node-api/src/main/kotlin/net/corda/nodeapi/internal/CordaNettyConnectorFactory.kt
View File

@ -14,15 +14,16 @@ import org.apache.activemq.artemis.utils.ConfigurationHelper
import java.util.concurrent.Executor
import java.util.concurrent.ScheduledExecutorService
class NodeNettyConnectorFactory : ConnectorFactory {
class CordaNettyConnectorFactory : ConnectorFactory {
override fun createConnector(configuration: MutableMap<String, Any>?,
handler: BufferHandler?,
listener: ClientConnectionLifeCycleListener?,
closeExecutor: Executor?,
threadPool: Executor?,
scheduledThreadPool: ScheduledExecutorService?,
closeExecutor: Executor,
threadPool: Executor,
scheduledThreadPool: ScheduledExecutorService,
protocolManager: ClientProtocolManager?): Connector {
val threadPoolName = ConfigurationHelper.getStringProperty(ArtemisTcpTransport.THREAD_POOL_NAME_NAME, "Connector", configuration)
setThreadPoolName(threadPool, closeExecutor, scheduledThreadPool, threadPoolName)
val trace = ConfigurationHelper.getBooleanProperty(ArtemisTcpTransport.TRACE_NAME, false, configuration)
return NettyConnector(
configuration,
@ -31,7 +32,7 @@ class NodeNettyConnectorFactory : ConnectorFactory {
closeExecutor,
threadPool,
scheduledThreadPool,
MyClientProtocolManager(threadPoolName, trace)
MyClientProtocolManager("$threadPoolName-netty", trace)
)
}
@ -39,6 +40,17 @@ class NodeNettyConnectorFactory : ConnectorFactory {
override fun getDefaults(): Map<String?, Any?> = NettyConnector.DEFAULT_CONFIG
private fun setThreadPoolName(threadPool: Executor, closeExecutor: Executor, scheduledThreadPool: ScheduledExecutorService, name: String) {
threadPool.setThreadPoolName("$name-artemis")
// Artemis will actually wrap the same backing Executor to create multiple "OrderedExecutors". In this scenerio both the threadPool
// and the closeExecutor are the same when it comes to the pool names. If however they are different then given them separate names.
if (threadPool.rootExecutor !== closeExecutor.rootExecutor) {
closeExecutor.setThreadPoolName("$name-artemis-closer")
}
// The scheduler is separate
scheduledThreadPool.setThreadPoolName("$name-artemis-scheduler")
}
private class MyClientProtocolManager(private val threadPoolName: String, private val trace: Boolean) : ActiveMQClientProtocolManager() {
override fun addChannelHandlers(pipeline: ChannelPipeline) {

96
node-api/src/main/kotlin/net/corda/nodeapi/internal/bridging/AMQPBridgeManager.kt
View File

@ -22,6 +22,7 @@ import net.corda.nodeapi.internal.protonwrapper.netty.AMQPClient
import net.corda.nodeapi.internal.protonwrapper.netty.AMQPConfiguration
import net.corda.nodeapi.internal.protonwrapper.netty.ProxyConfig
import net.corda.nodeapi.internal.protonwrapper.netty.RevocationConfig
import net.corda.nodeapi.internal.protonwrapper.netty.sslDelegatedTaskExecutor
import org.apache.activemq.artemis.api.core.ActiveMQObjectClosedException
import org.apache.activemq.artemis.api.core.SimpleString
import org.apache.activemq.artemis.api.core.client.ActiveMQClient.DEFAULT_ACK_BATCH_SIZE
@ -31,6 +32,7 @@ import org.apache.activemq.artemis.api.core.client.ClientSession
import org.slf4j.MDC
import rx.Subscription
import java.time.Duration
import java.util.concurrent.ExecutorService
import java.util.concurrent.Executors
import java.util.concurrent.ScheduledExecutorService
import java.util.concurrent.ScheduledFuture
@ -53,7 +55,7 @@ open class AMQPBridgeManager(keyStore: CertificateStore,
maxMessageSize: Int,
revocationConfig: RevocationConfig,
enableSNI: Boolean,
private val artemisMessageClientFactory: () -> ArtemisSessionProvider,
private val artemisMessageClientFactory: (String) -> ArtemisSessionProvider,
private val bridgeMetricsService: BridgeMetricsService? = null,
trace: Boolean,
sslHandshakeTimeout: Duration?,
@ -78,9 +80,11 @@ open class AMQPBridgeManager(keyStore: CertificateStore,
private val amqpConfig: AMQPConfiguration = AMQPConfigurationImpl(keyStore, trustStore, proxyConfig, maxMessageSize, revocationConfig,useOpenSSL, enableSNI, trace = trace, _sslHandshakeTimeout = sslHandshakeTimeout)
private var sharedEventLoopGroup: EventLoopGroup? = null
private var sslDelegatedTaskExecutor: ExecutorService? = null
private var artemis: ArtemisSessionProvider? = null
companion object {
private val log = contextLogger()
private const val CORDA_NUM_BRIDGE_THREADS_PROP_NAME = "net.corda.nodeapi.amqpbridgemanager.NumBridgeThreads"
@ -97,18 +101,11 @@ open class AMQPBridgeManager(keyStore: CertificateStore,
* however Artemis and the remote Corda instanced will deduplicate these messages.
*/
@Suppress("TooManyFunctions")
private class AMQPBridge(val sourceX500Name: String,
val queueName: String,
val targets: List<NetworkHostAndPort>,
val allowedRemoteLegalNames: Set<CordaX500Name>,
private val amqpConfig: AMQPConfiguration,
sharedEventGroup: EventLoopGroup,
private val artemis: ArtemisSessionProvider,
private val bridgeMetricsService: BridgeMetricsService?,
private val bridgeConnectionTTLSeconds: Int) {
companion object {
private val log = contextLogger()
}
private inner class AMQPBridge(val sourceX500Name: String,
val queueName: String,
val targets: List<NetworkHostAndPort>,
val allowedRemoteLegalNames: Set<CordaX500Name>,
private val amqpConfig: AMQPConfiguration) {
private fun withMDC(block: () -> Unit) {
val oldMDC = MDC.getCopyOfContextMap() ?: emptyMap<String, String>()
@ -134,13 +131,18 @@ open class AMQPBridgeManager(keyStore: CertificateStore,
private fun logWarnWithMDC(msg: String) = withMDC { log.warn(msg) }
val amqpClient = AMQPClient(targets, allowedRemoteLegalNames, amqpConfig, sharedThreadPool = sharedEventGroup)
val amqpClient = AMQPClient(
targets,
allowedRemoteLegalNames,
amqpConfig,
AMQPClient.NettyThreading.Shared(sharedEventLoopGroup!!, sslDelegatedTaskExecutor!!)
)
private var session: ClientSession? = null
private var consumer: ClientConsumer? = null
private var connectedSubscription: Subscription? = null
@Volatile
private var messagesReceived: Boolean = false
private val eventLoop: EventLoop = sharedEventGroup.next()
private val eventLoop: EventLoop = sharedEventLoopGroup!!.next()
private var artemisState: ArtemisState = ArtemisState.STOPPED
set(value) {
logDebugWithMDC { "State change $field to $value" }
@ -152,32 +154,9 @@ open class AMQPBridgeManager(keyStore: CertificateStore,
private var scheduledExecutorService: ScheduledExecutorService
= Executors.newSingleThreadScheduledExecutor(ThreadFactoryBuilder().setNameFormat("bridge-connection-reset-%d").build())
@Suppress("ClassNaming")
private sealed class ArtemisState {
object STARTING : ArtemisState()
data class STARTED(override val pending: ScheduledFuture<Unit>) : ArtemisState()
object CHECKING : ArtemisState()
object RESTARTED : ArtemisState()
object RECEIVING : ArtemisState()
object AMQP_STOPPED : ArtemisState()
object AMQP_STARTING : ArtemisState()
object AMQP_STARTED : ArtemisState()
object AMQP_RESTARTED : ArtemisState()
object STOPPING : ArtemisState()
object STOPPED : ArtemisState()
data class STOPPED_AMQP_START_SCHEDULED(override val pending: ScheduledFuture<Unit>) : ArtemisState()
open val pending: ScheduledFuture<Unit>? = null
override fun toString(): String = javaClass.simpleName
}
private fun artemis(inProgress: ArtemisState, block: (precedingState: ArtemisState) -> ArtemisState) {
val runnable = {
synchronized(artemis) {
synchronized(artemis!!) {
try {
val precedingState = artemisState
artemisState.pending?.cancel(false)
@ -253,7 +232,7 @@ open class AMQPBridgeManager(keyStore: CertificateStore,
}
}
artemis(ArtemisState.STARTING) {
val startedArtemis = artemis.started
val startedArtemis = artemis!!.started
if (startedArtemis == null) {
logInfoWithMDC("Bridge Connected but Artemis is disconnected")
ArtemisState.STOPPED
@ -457,6 +436,29 @@ open class AMQPBridgeManager(keyStore: CertificateStore,
}
}
@Suppress("ClassNaming")
private sealed class ArtemisState {
object STARTING : ArtemisState()
data class STARTED(override val pending: ScheduledFuture<Unit>) : ArtemisState()
object CHECKING : ArtemisState()
object RESTARTED : ArtemisState()
object RECEIVING : ArtemisState()
object AMQP_STOPPED : ArtemisState()
object AMQP_STARTING : ArtemisState()
object AMQP_STARTED : ArtemisState()
object AMQP_RESTARTED : ArtemisState()
object STOPPING : ArtemisState()
object STOPPED : ArtemisState()
data class STOPPED_AMQP_START_SCHEDULED(override val pending: ScheduledFuture<Unit>) : ArtemisState()
open val pending: ScheduledFuture<Unit>? = null
override fun toString(): String = javaClass.simpleName
}
override fun deployBridge(sourceX500Name: String, queueName: String, targets: List<NetworkHostAndPort>, legalNames: Set<CordaX500Name>) {
lock.withLock {
val bridges = queueNamesToBridgesMap.getOrPut(queueName) { mutableListOf() }
@ -467,8 +469,7 @@ open class AMQPBridgeManager(keyStore: CertificateStore,
}
val newAMQPConfig = with(amqpConfig) { AMQPConfigurationImpl(keyStore, trustStore, proxyConfig, maxMessageSize,
revocationConfig, useOpenSsl, enableSNI, sourceX500Name, trace, sslHandshakeTimeout) }
val newBridge = AMQPBridge(sourceX500Name, queueName, targets, legalNames, newAMQPConfig, sharedEventLoopGroup!!, artemis!!,
bridgeMetricsService, bridgeConnectionTTLSeconds)
val newBridge = AMQPBridge(sourceX500Name, queueName, targets, legalNames, newAMQPConfig)
bridges += newBridge
bridgeMetricsService?.bridgeCreated(targets, legalNames)
newBridge
@ -497,15 +498,16 @@ open class AMQPBridgeManager(keyStore: CertificateStore,
// queueNamesToBridgesMap returns a mutable list, .toList converts it to a immutable list so it won't be changed by the [destroyBridge] method.
val bridges = queueNamesToBridgesMap[queueName]?.toList()
destroyBridge(queueName, bridges?.flatMap { it.targets } ?: emptyList())
bridges?.map {
bridges?.associate {
it.sourceX500Name to BridgeEntry(it.queueName, it.targets, it.allowedRemoteLegalNames.toList(), serviceAddress = false)
}?.toMap() ?: emptyMap()
} ?: emptyMap()
}
}
override fun start() {
sharedEventLoopGroup = NioEventLoopGroup(NUM_BRIDGE_THREADS, DefaultThreadFactory("AMQPBridge", Thread.MAX_PRIORITY))
val artemis = artemisMessageClientFactory()
sharedEventLoopGroup = NioEventLoopGroup(NUM_BRIDGE_THREADS, DefaultThreadFactory("NettyBridge", Thread.MAX_PRIORITY))
sslDelegatedTaskExecutor = sslDelegatedTaskExecutor("NettyBridge")
val artemis = artemisMessageClientFactory("ArtemisBridge")
this.artemis = artemis
artemis.start()
}
@ -522,6 +524,8 @@ open class AMQPBridgeManager(keyStore: CertificateStore,
sharedEventLoopGroup = null
queueNamesToBridgesMap.clear()
artemis?.stop()
sslDelegatedTaskExecutor?.shutdown()
sslDelegatedTaskExecutor = null
}
}
}

4
node-api/src/main/kotlin/net/corda/nodeapi/internal/bridging/BridgeControlListener.kt
View File

@ -35,7 +35,7 @@ class BridgeControlListener(private val keyStore: CertificateStore,
maxMessageSize: Int,
revocationConfig: RevocationConfig,
enableSNI: Boolean,
private val artemisMessageClientFactory: () -> ArtemisSessionProvider,
private val artemisMessageClientFactory: (String) -> ArtemisSessionProvider,
bridgeMetricsService: BridgeMetricsService? = null,
trace: Boolean = false,
sslHandshakeTimeout: Duration? = null,
@ -80,7 +80,7 @@ class BridgeControlListener(private val keyStore: CertificateStore,
bridgeNotifyQueue = "$BRIDGE_NOTIFY.$queueDisambiguityId"
bridgeManager.start()
val artemis = artemisMessageClientFactory()
val artemis = artemisMessageClientFactory("BridgeControl")
this.artemis = artemis
artemis.start()
val artemisClient = artemis.started!!

4
node-api/src/main/kotlin/net/corda/nodeapi/internal/bridging/LoopbackBridgeManager.kt
View File

@ -37,7 +37,7 @@ class LoopbackBridgeManager(keyStore: CertificateStore,
maxMessageSize: Int,
revocationConfig: RevocationConfig,
enableSNI: Boolean,
private val artemisMessageClientFactory: () -> ArtemisSessionProvider,
private val artemisMessageClientFactory: (String) -> ArtemisSessionProvider,
private val bridgeMetricsService: BridgeMetricsService? = null,
private val isLocalInbox: (String) -> Boolean,
trace: Boolean,
@ -204,7 +204,7 @@ class LoopbackBridgeManager(keyStore: CertificateStore,
override fun start() {
super.start()
val artemis = artemisMessageClientFactory()
val artemis = artemisMessageClientFactory("LoopbackBridge")
this.artemis = artemis
artemis.start()
}

63
node-api/src/main/kotlin/net/corda/nodeapi/internal/protonwrapper/netty/AMQPClient.kt
View File

@ -32,7 +32,9 @@ import rx.Observable
import rx.subjects.PublishSubject
import java.lang.Long.min
import java.net.InetSocketAddress
import java.util.concurrent.Executor
import java.util.concurrent.ExecutorService
import java.util.concurrent.ThreadPoolExecutor
import java.util.concurrent.TimeUnit
import java.util.concurrent.locks.ReentrantLock
import kotlin.concurrent.withLock
@ -61,8 +63,7 @@ data class ProxyConfig(val version: ProxyVersion, val proxyAddress: NetworkHostA
class AMQPClient(private val targets: List<NetworkHostAndPort>,
val allowedRemoteLegalNames: Set<CordaX500Name>,
private val configuration: AMQPConfiguration,
private val sharedThreadPool: EventLoopGroup? = null,
private val threadPoolName: String = "AMQPClient",
private val nettyThreading: NettyThreading = NettyThreading.NonShared("AMQPClient"),
private val distPointCrlSource: CertDistPointCrlSource = CertDistPointCrlSource.SINGLETON) : AutoCloseable {
companion object {
init {
@ -82,7 +83,6 @@ class AMQPClient(private val targets: List<NetworkHostAndPort>,
private val lock = ReentrantLock()
@Volatile
private var started: Boolean = false
private var workerGroup: EventLoopGroup? = null
@Volatile
private var clientChannel: Channel? = null
// Offset into the list of targets, so that we can implement round-robin reconnect logic.
@ -94,7 +94,6 @@ class AMQPClient(private val targets: List<NetworkHostAndPort>,
private var amqpActive = false
@Volatile
private var amqpChannelHandler: ChannelHandler? = null
private var sslDelegatedTaskExecutor: ExecutorService? = null
val localAddressString: String
get() = clientChannel?.localAddress()?.toString() ?: "<unknownLocalAddress>"
@ -123,7 +122,7 @@ class AMQPClient(private val targets: List<NetworkHostAndPort>,
log.info("Failed to connect to $currentTarget", future.cause())
if (started) {
workerGroup?.schedule({
nettyThreading.eventLoopGroup.schedule({
nextTarget()
restart()
}, retryInterval, TimeUnit.MILLISECONDS)
@ -142,7 +141,7 @@ class AMQPClient(private val targets: List<NetworkHostAndPort>,
clientChannel = null
if (started && !amqpActive) {
log.debug { "Scheduling restart of $currentTarget (AMQP inactive)" }
workerGroup?.schedule({
nettyThreading.eventLoopGroup.schedule({
nextTarget()
restart()
}, retryInterval, TimeUnit.MILLISECONDS)
@ -198,7 +197,6 @@ class AMQPClient(private val targets: List<NetworkHostAndPort>,
val wrappedKeyManagerFactory = CertHoldingKeyManagerFactoryWrapper(keyManagerFactory, parent.configuration)
val target = parent.currentTarget
val delegatedTaskExecutor = checkNotNull(parent.sslDelegatedTaskExecutor)
val handler = if (parent.configuration.useOpenSsl) {
createClientOpenSslHandler(
target,
@ -206,7 +204,7 @@ class AMQPClient(private val targets: List<NetworkHostAndPort>,
wrappedKeyManagerFactory,
trustManagerFactory,
ch.alloc(),
delegatedTaskExecutor
parent.nettyThreading.sslDelegatedTaskExecutor
)
} else {
createClientSslHandler(
@ -214,7 +212,7 @@ class AMQPClient(private val targets: List<NetworkHostAndPort>,
parent.allowedRemoteLegalNames,
wrappedKeyManagerFactory,
trustManagerFactory,
delegatedTaskExecutor
parent.nettyThreading.sslDelegatedTaskExecutor
)
}
handler.handshakeTimeoutMillis = conf.sslHandshakeTimeout.toMillis()
@ -256,7 +254,7 @@ class AMQPClient(private val targets: List<NetworkHostAndPort>,
if (started && amqpActive) {
log.debug { "Scheduling restart of $currentTarget (AMQP active)" }
workerGroup?.schedule({
nettyThreading.eventLoopGroup.schedule({
nextTarget()
restart()
}, retryInterval, TimeUnit.MILLISECONDS)
@ -273,8 +271,7 @@ class AMQPClient(private val targets: List<NetworkHostAndPort>,
return
}
log.info("Connect to: $currentTarget")
sslDelegatedTaskExecutor = sslDelegatedTaskExecutor(threadPoolName)
workerGroup = sharedThreadPool ?: NioEventLoopGroup(NUM_CLIENT_THREADS, DefaultThreadFactory(threadPoolName, Thread.MAX_PRIORITY))
(nettyThreading as? NettyThreading.NonShared)?.start()
started = true
restart()
}
@ -286,7 +283,7 @@ class AMQPClient(private val targets: List<NetworkHostAndPort>,
}
val bootstrap = Bootstrap()
// TODO Needs more configuration control when we profile. e.g. to use EPOLL on Linux
bootstrap.group(workerGroup).channel(NioSocketChannel::class.java).handler(ClientChannelInitializer(this))
bootstrap.group(nettyThreading.eventLoopGroup).channel(NioSocketChannel::class.java).handler(ClientChannelInitializer(this))
// Delegate DNS Resolution to the proxy side, if we are using proxy.
if (configuration.proxyConfig != null) {
bootstrap.resolver(NoopAddressResolverGroup.INSTANCE)
@ -300,16 +297,12 @@ class AMQPClient(private val targets: List<NetworkHostAndPort>,
lock.withLock {
log.info("Stopping connection to: $currentTarget, Local address: $localAddressString")
started = false
if (sharedThreadPool == null) {
workerGroup?.shutdownGracefully()
workerGroup?.terminationFuture()?.sync()
if (nettyThreading is NettyThreading.NonShared) {
nettyThreading.stop()
} else {
clientChannel?.close()?.sync()
}
clientChannel = null
workerGroup = null
sslDelegatedTaskExecutor?.shutdown()
sslDelegatedTaskExecutor = null
log.info("Stopped connection to $currentTarget")
}
}
@ -350,4 +343,36 @@ class AMQPClient(private val targets: List<NetworkHostAndPort>,
private val _onConnection = PublishSubject.create<ConnectionChange>().toSerialized()
val onConnection: Observable<ConnectionChange>
get() = _onConnection
sealed class NettyThreading {
abstract val eventLoopGroup: EventLoopGroup
abstract val sslDelegatedTaskExecutor: Executor
class Shared(override val eventLoopGroup: EventLoopGroup,
override val sslDelegatedTaskExecutor: ExecutorService = sslDelegatedTaskExecutor("AMQPClient")) : NettyThreading()
class NonShared(val threadPoolName: String) : NettyThreading() {
private var _eventLoopGroup: NioEventLoopGroup? = null
override val eventLoopGroup: EventLoopGroup get() = checkNotNull(_eventLoopGroup)
private var _sslDelegatedTaskExecutor: ThreadPoolExecutor? = null
override val sslDelegatedTaskExecutor: ExecutorService get() = checkNotNull(_sslDelegatedTaskExecutor)
fun start() {
check(_eventLoopGroup == null)
check(_sslDelegatedTaskExecutor == null)
_eventLoopGroup = NioEventLoopGroup(NUM_CLIENT_THREADS, DefaultThreadFactory(threadPoolName, Thread.MAX_PRIORITY))
_sslDelegatedTaskExecutor = sslDelegatedTaskExecutor(threadPoolName)
}
fun stop() {
eventLoopGroup.shutdownGracefully()
eventLoopGroup.terminationFuture().sync()
sslDelegatedTaskExecutor.shutdown()
_eventLoopGroup = null
_sslDelegatedTaskExecutor = null
}
}
}
}