ExternalVendorCode/corda
1
0
Fork 0
mirror of https://github.com/corda/corda.git synced 2025-06-20 16:10:26 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Multi threaded state machine

Browse Source
This commit is contained in:
Andras Slemmer
2017-11-22 16:03:44 +00:00
parent 2edf632f7f
commit b71f0c49fb
50 changed files with 1515 additions and 375 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -7,18 +7,22 @@ import net.corda.nodeapi.ArtemisTcpTransport
import net.corda.nodeapi.ConnectionDirection
import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.NODE_USER
import net.corda.nodeapi.internal.config.SSLConfiguration
import org.apache.activemq.artemis.api.core.client.ActiveMQClient
import org.apache.activemq.artemis.api.core.client.*
import org.apache.activemq.artemis.api.core.client.ActiveMQClient.DEFAULT_ACK_BATCH_SIZE
import org.apache.activemq.artemis.api.core.client.ClientProducer
import org.apache.activemq.artemis.api.core.client.ClientSession
import org.apache.activemq.artemis.api.core.client.ClientSessionFactory
class ArtemisMessagingClient(private val config: SSLConfiguration, private val serverAddress: NetworkHostAndPort, private val maxMessageSize: Int) {
class ArtemisMessagingClient(
private val config: SSLConfiguration,
private val serverAddress: NetworkHostAndPort,
private val maxMessageSize: Int,
private val autoCommitSends: Boolean = true,
private val autoCommitAcks: Boolean = true,
private val confirmationWindowSize: Int = -1
) {
companion object {
private val log = loggerFor<ArtemisMessagingClient>()
}
class Started(val sessionFactory: ClientSessionFactory, val session: ClientSession, val producer: ClientProducer)
class Started(val serverLocator: ServerLocator, val sessionFactory: ClientSessionFactory, val session: ClientSession, val producer: ClientProducer)
var started: Started? = null
private set
@ -35,17 +39,18 @@ class ArtemisMessagingClient(private val config: SSLConfiguration, private val s
clientFailureCheckPeriod = -1
minLargeMessageSize = maxMessageSize
isUseGlobalPools = nodeSerializationEnv != null
confirmationWindowSize = this@ArtemisMessagingClient.confirmationWindowSize
}
val sessionFactory = locator.createSessionFactory()
// Login using the node username. The broker will authenticate us as its node (as opposed to another peer)
// using our TLS certificate.
// Note that the acknowledgement of messages is not flushed to the Artermis journal until the default buffer
// size of 1MB is acknowledged.
val session = sessionFactory!!.createSession(NODE_USER, NODE_USER, false, true, true, locator.isPreAcknowledge, DEFAULT_ACK_BATCH_SIZE)
val session = sessionFactory!!.createSession(NODE_USER, NODE_USER, false, autoCommitSends, autoCommitAcks, locator.isPreAcknowledge, DEFAULT_ACK_BATCH_SIZE)
session.start()
// Create a general purpose producer.
val producer = session.createProducer()
return Started(sessionFactory, session, producer).also { started = it }
return Started(locator, sessionFactory, session, producer).also { started = it }
}
fun stop() = synchronized(this) {
@ -55,6 +60,7 @@ class ArtemisMessagingClient(private val config: SSLConfiguration, private val s
session.commit()
// Closing the factory closes all the sessions it produced as well.
sessionFactory.close()
serverLocator.close()
}
started = null
}