ExternalVendorCode/corda
1
0
Fork 0
mirror of https://github.com/corda/corda.git synced 2024-12-27 00:21:12 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Merge pull request #428 from corda/mnesbit-merge-20180131

Browse Source 
Merge up bridging code move from OS
This commit is contained in:
Matthew Nesbit 2018-01-31 14:50:34 +00:00 committed by GitHub
commit 2edf632f7f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
26 changed files with 155 additions and 90 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

102
docs/source/clientrpc.rst
View File

@ -1,34 +1,37 @@
Client RPC
==========
There are multiple ways to interact with a node from a *client program*, but if your client is written in a JVM
compatible language the easiest way to do so is using the client library. The library connects to your running
node using a message queue protocol and then provides a simple RPC interface to interact with it. You make calls
on a Java object as normal, and the marshalling back and forth is handled for you.
.. contents::
The starting point for the client library is the `CordaRPCClient`_ class. This provides a ``start`` method that
returns a `CordaRPCConnection`_, holding an implementation of the `CordaRPCOps`_ that may be accessed with ``proxy``
in Kotlin and ``getProxy()`` in Java. Observables that are returned by RPCs can be subscribed to in order to receive
an ongoing stream of updates from the node. More detail on how to use this is provided in the docs for the proxy method.
Overview
--------
Corda provides a client library that allows you to easily write clients in a JVM-compatible language to interact
with a running node. The library connects to the node using a message queue protocol and then provides a simple RPC
interface to interact with the node. You make calls on a Java object as normal, and the marshalling back and forth is
handled for you.
The starting point for the client library is the `CordaRPCClient`_ class. `CordaRPCClient`_ provides a ``start`` method
that returns a `CordaRPCConnection`_. A `CordaRPCConnection`_ allows you to access an implementation of the
`CordaRPCOps`_ interface with ``proxy`` in Kotlin or ``getProxy()`` in Java. The observables that are returned by RPC
operations can be subscribed to in order to receive an ongoing stream of updates from the node. More detail on this
functionality is provided in the docs for the ``proxy`` method.
.. warning:: The returned `CordaRPCConnection`_ is somewhat expensive to create and consumes a small amount of
server side resources. When you're done with it, call ``close`` on it. Alternatively you may use the ``use``
method on `CordaRPCClient`_ which cleans up automatically after the passed in lambda finishes. Don't create
a new proxy for every call you make - reuse an existing one.
For a brief tutorial on how one can use the RPC API see :doc:`tutorial-clientrpc-api`.
For a brief tutorial on using the RPC API, see :doc:`tutorial-clientrpc-api`.
.. _rpc_security_mgmt_ref:
RPC permissions
---------------
If a node's owner needs to interact with their node via RPC (e.g. to read the contents of the node's storage), they
must define one or more RPC users. Each user is authenticated with a username and password, and is assigned a set of
permissions that RPC can use for fine-grain access control.
For a node's owner to interact with their node via RPC, they must define one or more RPC users. Each user is
authenticated with a username and password, and is assigned a set of permissions that control which RPC operations they
can perform.
These users are added to the node's ``node.conf`` file.
The simplest way of adding an RPC user is to include it in the ``rpcUsers`` list:
RPC users are created by adding them to the ``rpcUsers`` list in the node's ``node.conf`` file:
.. container:: codeset
@ -43,7 +46,12 @@ The simplest way of adding an RPC user is to include it in the ``rpcUsers`` list
...
]
Users need permissions to invoke any RPC call. By default, nothing is allowed. These permissions are specified as follows:
By default, RPC users are not permissioned to perform any RPC operations.
Granting flow permissions
~~~~~~~~~~~~~~~~~~~~~~~~~
You provide an RPC user with the permission to start a specific flow using the syntax
``StartFlow.<fully qualified flow name>``:
.. container:: codeset
@ -61,14 +69,64 @@ Users need permissions to invoke any RPC call. By default, nothing is allowed. T
...
]
Permissions Syntax
^^^^^^^^^^^^^^^^^^
You can also provide an RPC user with the permission to start any flow using the syntax
``InvokeRpc.startFlow``:
Fine grained permissions allow a user to invoke a specific RPC operation, or to start a specific flow. The syntax is:
.. container:: codeset
- to start a specific flow: ``StartFlow.<fully qualified flow name>`` e.g., ``StartFlow.net.corda.flows.ExampleFlow1``.
- to invoke a RPC operation: ``InvokeRpc.<rpc method name>`` e.g., ``InvokeRpc.nodeInfo``.
.. note:: Permission ``InvokeRpc.startFlow`` allows a user to initiate all flows.
.. sourcecode:: groovy
rpcUsers=[
{
username=exampleUser
password=examplePass
permissions=[
"InvokeRpc.startFlow"
]
}
...
]
Granting other RPC permissions
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
You provide an RPC user with the permission to perform a specific RPC operation using the syntax
``InvokeRpc.<rpc method name>``:
.. container:: codeset
.. sourcecode:: groovy
rpcUsers=[
{
username=exampleUser
password=examplePass
permissions=[
"InvokeRpc.nodeInfo",
"InvokeRpc.networkMapSnapshot"
]
}
...
]
Granting all permissions
~~~~~~~~~~~~~~~~~~~~~~~~
You can provide an RPC user with the permission to perform any RPC operation (including starting any flow) using the
``ALL`` permission:
.. container:: codeset
.. sourcecode:: groovy
rpcUsers=[
{
username=exampleUser
password=examplePass
permissions=[
"ALL"
]
}
...
]
.. _authentication_ref:

2
node/src/main/kotlin/net/corda/node/services/messaging/ArtemisMessagingClient.kt → node-api/src/main/kotlin/net/corda/nodeapi/internal/ArtemisMessagingClient.kt
View File

@ -1,4 +1,4 @@
package net.corda.node.services.messaging
package net.corda.nodeapi.internal
import net.corda.core.serialization.internal.nodeSerializationEnv
import net.corda.core.utilities.NetworkHostAndPort

13
node/src/main/kotlin/net/corda/node/services/messaging/AMQPBridgeManager.kt → node-api/src/main/kotlin/net/corda/nodeapi/internal/bridging/AMQPBridgeManager.kt
View File

@ -1,4 +1,4 @@
package net.corda.node.services.messaging
package net.corda.nodeapi.internal.bridging
import io.netty.channel.EventLoopGroup
import io.netty.channel.nio.NioEventLoopGroup
@ -7,14 +7,15 @@ import net.corda.core.internal.VisibleForTesting
import net.corda.core.node.NodeInfo
import net.corda.core.utilities.NetworkHostAndPort
import net.corda.core.utilities.debug
import net.corda.node.internal.protonwrapper.messages.MessageStatus
import net.corda.node.internal.protonwrapper.netty.AMQPClient
import net.corda.node.services.config.NodeConfiguration
import net.corda.node.services.messaging.AMQPBridgeManager.AMQPBridge.Companion.getBridgeName
import net.corda.nodeapi.internal.ArtemisMessagingClient
import net.corda.nodeapi.internal.ArtemisMessagingComponent
import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.NODE_USER
import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.PEER_USER
import net.corda.nodeapi.internal.ArtemisMessagingComponent.RemoteInboxAddress.Companion.translateLocalQueueToInboxAddress
import net.corda.nodeapi.internal.bridging.AMQPBridgeManager.AMQPBridge.Companion.getBridgeName
import net.corda.nodeapi.internal.config.NodeSSLConfiguration
import net.corda.nodeapi.internal.protonwrapper.messages.MessageStatus
import net.corda.nodeapi.internal.protonwrapper.netty.AMQPClient
import org.apache.activemq.artemis.api.core.SimpleString
import org.apache.activemq.artemis.api.core.client.ActiveMQClient.DEFAULT_ACK_BATCH_SIZE
import org.apache.activemq.artemis.api.core.client.ClientConsumer
@ -34,7 +35,7 @@ import kotlin.concurrent.withLock
* The Netty thread pool used by the AMQPBridges is also shared and managed by the AMQPBridgeManager.
*/
@VisibleForTesting
class AMQPBridgeManager(val config: NodeConfiguration, val p2pAddress: NetworkHostAndPort, val maxMessageSize: Int) : BridgeManager {
class AMQPBridgeManager(val config: NodeSSLConfiguration, val p2pAddress: NetworkHostAndPort, val maxMessageSize: Int) : BridgeManager {
private val lock = ReentrantLock()
private val bridgeNameToBridgeMap = mutableMapOf<String, AMQPBridge>()

12
node/src/main/kotlin/net/corda/node/services/messaging/BridgeControlListener.kt → node-api/src/main/kotlin/net/corda/nodeapi/internal/bridging/BridgeControlListener.kt
View File

@ -1,25 +1,25 @@
package net.corda.node.services.messaging
package net.corda.nodeapi.internal.bridging
import net.corda.core.serialization.SerializationDefaults
import net.corda.core.serialization.deserialize
import net.corda.core.serialization.serialize
import net.corda.core.utilities.NetworkHostAndPort
import net.corda.core.utilities.contextLogger
import net.corda.node.services.config.NodeConfiguration
import net.corda.nodeapi.internal.ArtemisMessagingClient
import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.BRIDGE_CONTROL
import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.BRIDGE_NOTIFY
import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.P2P_PREFIX
import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.PEERS_PREFIX
import net.corda.nodeapi.internal.BridgeControl
import net.corda.nodeapi.internal.config.NodeSSLConfiguration
import org.apache.activemq.artemis.api.core.RoutingType
import org.apache.activemq.artemis.api.core.SimpleString
import org.apache.activemq.artemis.api.core.client.ClientConsumer
import org.apache.activemq.artemis.api.core.client.ClientMessage
import java.util.*
internal class BridgeControlListener(val config: NodeConfiguration,
val p2pAddress: NetworkHostAndPort,
val maxMessageSize: Int) : AutoCloseable {
class BridgeControlListener(val config: NodeSSLConfiguration,
val p2pAddress: NetworkHostAndPort,
val maxMessageSize: Int) : AutoCloseable {
private val bridgeId: String = UUID.randomUUID().toString()
private val bridgeManager: BridgeManager = AMQPBridgeManager(config, p2pAddress, maxMessageSize)
private val validInboundQueues = mutableSetOf<String>()

2
node-api/src/main/kotlin/net/corda/nodeapi/internal/BridgeControlMessages.kt → node-api/src/main/kotlin/net/corda/nodeapi/internal/bridging/BridgeControlMessages.kt
View File

@ -1,4 +1,4 @@
package net.corda.nodeapi.internal
package net.corda.nodeapi.internal.bridging
import net.corda.core.identity.CordaX500Name
import net.corda.core.serialization.CordaSerializable

4
node/src/main/kotlin/net/corda/node/services/messaging/BridgeManager.kt → node-api/src/main/kotlin/net/corda/nodeapi/internal/bridging/BridgeManager.kt
View File

@ -1,4 +1,4 @@
package net.corda.node.services.messaging
package net.corda.nodeapi.internal.bridging
import net.corda.core.identity.CordaX500Name
import net.corda.core.internal.VisibleForTesting
@ -6,7 +6,7 @@ import net.corda.core.node.NodeInfo
import net.corda.core.utilities.NetworkHostAndPort
/**
* Provides an internal interface that the [ArtemisMessagingServer] delegates to for Bridge activities.
* Provides an internal interface that the [BridgeControlListener] delegates to for Bridge activities.
*/
@VisibleForTesting
interface BridgeManager : AutoCloseable {

8
node/src/main/kotlin/net/corda/node/internal/protonwrapper/engine/ConnectionStateMachine.kt → node-api/src/main/kotlin/net/corda/nodeapi/internal/protonwrapper/engine/ConnectionStateMachine.kt
View File

@ -1,4 +1,4 @@
package net.corda.node.internal.protonwrapper.engine
package net.corda.nodeapi.internal.protonwrapper.engine
import io.netty.buffer.ByteBuf
import io.netty.buffer.PooledByteBufAllocator
@ -7,9 +7,9 @@ import io.netty.channel.Channel
import io.netty.channel.ChannelHandlerContext
import net.corda.core.utilities.NetworkHostAndPort
import net.corda.core.utilities.debug
import net.corda.node.internal.protonwrapper.messages.MessageStatus
import net.corda.node.internal.protonwrapper.messages.impl.ReceivedMessageImpl
import net.corda.node.internal.protonwrapper.messages.impl.SendableMessageImpl
import net.corda.nodeapi.internal.protonwrapper.messages.MessageStatus
import net.corda.nodeapi.internal.protonwrapper.messages.impl.ReceivedMessageImpl
import net.corda.nodeapi.internal.protonwrapper.messages.impl.SendableMessageImpl
import org.apache.qpid.proton.Proton
import org.apache.qpid.proton.amqp.Binary
import org.apache.qpid.proton.amqp.Symbol

8
node/src/main/kotlin/net/corda/node/internal/protonwrapper/engine/EventProcessor.kt → node-api/src/main/kotlin/net/corda/nodeapi/internal/protonwrapper/engine/EventProcessor.kt
View File

@ -1,12 +1,12 @@
package net.corda.node.internal.protonwrapper.engine
package net.corda.nodeapi.internal.protonwrapper.engine
import io.netty.buffer.ByteBuf
import io.netty.channel.Channel
import io.netty.channel.ChannelHandlerContext
import net.corda.core.utilities.debug
import net.corda.node.internal.protonwrapper.messages.MessageStatus
import net.corda.node.internal.protonwrapper.messages.impl.ReceivedMessageImpl
import net.corda.node.internal.protonwrapper.messages.impl.SendableMessageImpl
import net.corda.nodeapi.internal.protonwrapper.messages.MessageStatus
import net.corda.nodeapi.internal.protonwrapper.messages.impl.ReceivedMessageImpl
import net.corda.nodeapi.internal.protonwrapper.messages.impl.SendableMessageImpl
import org.apache.qpid.proton.Proton
import org.apache.qpid.proton.amqp.messaging.Accepted
import org.apache.qpid.proton.amqp.messaging.Rejected

2
node/src/main/kotlin/net/corda/node/internal/protonwrapper/engine/NettyWritable.kt → node-api/src/main/kotlin/net/corda/nodeapi/internal/protonwrapper/engine/NettyWritable.kt
View File

@ -1,4 +1,4 @@
package net.corda.node.internal.protonwrapper.engine
package net.corda.nodeapi.internal.protonwrapper.engine
import io.netty.buffer.ByteBuf
import org.apache.qpid.proton.codec.WritableBuffer

2
node/src/main/kotlin/net/corda/node/internal/protonwrapper/messages/ApplicationMessage.kt → node-api/src/main/kotlin/net/corda/nodeapi/internal/protonwrapper/messages/ApplicationMessage.kt
View File

@ -1,4 +1,4 @@
package net.corda.node.internal.protonwrapper.messages
package net.corda.nodeapi.internal.protonwrapper.messages
import net.corda.core.utilities.NetworkHostAndPort

2
node/src/main/kotlin/net/corda/node/internal/protonwrapper/messages/MessageStatus.kt → node-api/src/main/kotlin/net/corda/nodeapi/internal/protonwrapper/messages/MessageStatus.kt
View File

@ -1,4 +1,4 @@
package net.corda.node.internal.protonwrapper.messages
package net.corda.nodeapi.internal.protonwrapper.messages
/**
* The processing state of a message.

2
node/src/main/kotlin/net/corda/node/internal/protonwrapper/messages/ReceivedMessage.kt → node-api/src/main/kotlin/net/corda/nodeapi/internal/protonwrapper/messages/ReceivedMessage.kt
View File

@ -1,4 +1,4 @@
package net.corda.node.internal.protonwrapper.messages
package net.corda.nodeapi.internal.protonwrapper.messages
import net.corda.core.utilities.NetworkHostAndPort

2
node/src/main/kotlin/net/corda/node/internal/protonwrapper/messages/SendableMessage.kt → node-api/src/main/kotlin/net/corda/nodeapi/internal/protonwrapper/messages/SendableMessage.kt
View File

@ -1,4 +1,4 @@
package net.corda.node.internal.protonwrapper.messages
package net.corda.nodeapi.internal.protonwrapper.messages
import net.corda.core.concurrent.CordaFuture

6
node/src/main/kotlin/net/corda/node/internal/protonwrapper/messages/impl/ReceivedMessageImpl.kt → node-api/src/main/kotlin/net/corda/nodeapi/internal/protonwrapper/messages/impl/ReceivedMessageImpl.kt
View File

@ -1,9 +1,9 @@
package net.corda.node.internal.protonwrapper.messages.impl
package net.corda.nodeapi.internal.protonwrapper.messages.impl
import io.netty.channel.Channel
import net.corda.core.utilities.NetworkHostAndPort
import net.corda.node.internal.protonwrapper.messages.MessageStatus
import net.corda.node.internal.protonwrapper.messages.ReceivedMessage
import net.corda.nodeapi.internal.protonwrapper.messages.MessageStatus
import net.corda.nodeapi.internal.protonwrapper.messages.ReceivedMessage
import org.apache.qpid.proton.engine.Delivery
/**

6
node/src/main/kotlin/net/corda/node/internal/protonwrapper/messages/impl/SendableMessageImpl.kt → node-api/src/main/kotlin/net/corda/nodeapi/internal/protonwrapper/messages/impl/SendableMessageImpl.kt
View File

@ -1,11 +1,11 @@
package net.corda.node.internal.protonwrapper.messages.impl
package net.corda.nodeapi.internal.protonwrapper.messages.impl
import io.netty.buffer.ByteBuf
import net.corda.core.concurrent.CordaFuture
import net.corda.core.internal.concurrent.openFuture
import net.corda.core.utilities.NetworkHostAndPort
import net.corda.node.internal.protonwrapper.messages.MessageStatus
import net.corda.node.internal.protonwrapper.messages.SendableMessage
import net.corda.nodeapi.internal.protonwrapper.messages.MessageStatus
import net.corda.nodeapi.internal.protonwrapper.messages.SendableMessage
/**
* An internal packet management class that allows handling of the encoded buffers and

10
node/src/main/kotlin/net/corda/node/internal/protonwrapper/netty/AMQPChannelHandler.kt → node-api/src/main/kotlin/net/corda/nodeapi/internal/protonwrapper/netty/AMQPChannelHandler.kt
View File

@ -1,4 +1,4 @@
package net.corda.node.internal.protonwrapper.netty
package net.corda.nodeapi.internal.protonwrapper.netty
import io.netty.buffer.ByteBuf
import io.netty.channel.ChannelDuplexHandler
@ -10,11 +10,11 @@ import io.netty.handler.ssl.SslHandshakeCompletionEvent
import io.netty.util.ReferenceCountUtil
import net.corda.core.identity.CordaX500Name
import net.corda.core.utilities.debug
import net.corda.node.internal.protonwrapper.engine.EventProcessor
import net.corda.node.internal.protonwrapper.messages.ReceivedMessage
import net.corda.node.internal.protonwrapper.messages.impl.ReceivedMessageImpl
import net.corda.node.internal.protonwrapper.messages.impl.SendableMessageImpl
import net.corda.nodeapi.internal.crypto.x509
import net.corda.nodeapi.internal.protonwrapper.engine.EventProcessor
import net.corda.nodeapi.internal.protonwrapper.messages.ReceivedMessage
import net.corda.nodeapi.internal.protonwrapper.messages.impl.ReceivedMessageImpl
import net.corda.nodeapi.internal.protonwrapper.messages.impl.SendableMessageImpl
import org.apache.qpid.proton.engine.ProtonJTransport
import org.apache.qpid.proton.engine.Transport
import org.apache.qpid.proton.engine.impl.ProtocolTracer

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

@ -1,4 +1,4 @@
package net.corda.node.internal.protonwrapper.netty
package net.corda.nodeapi.internal.protonwrapper.netty
import io.netty.bootstrap.Bootstrap
import io.netty.channel.*
@ -12,9 +12,9 @@ import io.netty.util.internal.logging.Slf4JLoggerFactory
import net.corda.core.identity.CordaX500Name
import net.corda.core.utilities.NetworkHostAndPort
import net.corda.core.utilities.contextLogger
import net.corda.node.internal.protonwrapper.messages.ReceivedMessage
import net.corda.node.internal.protonwrapper.messages.SendableMessage
import net.corda.node.internal.protonwrapper.messages.impl.SendableMessageImpl
import net.corda.nodeapi.internal.protonwrapper.messages.ReceivedMessage
import net.corda.nodeapi.internal.protonwrapper.messages.SendableMessage
import net.corda.nodeapi.internal.protonwrapper.messages.impl.SendableMessageImpl
import rx.Observable
import rx.subjects.PublishSubject
import java.security.KeyStore

8
node/src/main/kotlin/net/corda/node/internal/protonwrapper/netty/AMQPServer.kt → node-api/src/main/kotlin/net/corda/nodeapi/internal/protonwrapper/netty/AMQPServer.kt
View File

@ -1,4 +1,4 @@
package net.corda.node.internal.protonwrapper.netty
package net.corda.nodeapi.internal.protonwrapper.netty
import io.netty.bootstrap.ServerBootstrap
import io.netty.channel.Channel
@ -14,9 +14,9 @@ import io.netty.util.internal.logging.InternalLoggerFactory
import io.netty.util.internal.logging.Slf4JLoggerFactory
import net.corda.core.utilities.NetworkHostAndPort
import net.corda.core.utilities.contextLogger
import net.corda.node.internal.protonwrapper.messages.ReceivedMessage
import net.corda.node.internal.protonwrapper.messages.SendableMessage
import net.corda.node.internal.protonwrapper.messages.impl.SendableMessageImpl
import net.corda.nodeapi.internal.protonwrapper.messages.ReceivedMessage
import net.corda.nodeapi.internal.protonwrapper.messages.SendableMessage
import net.corda.nodeapi.internal.protonwrapper.messages.impl.SendableMessageImpl
import org.apache.qpid.proton.engine.Delivery
import rx.Observable
import rx.subjects.PublishSubject

2
node/src/main/kotlin/net/corda/node/internal/protonwrapper/netty/ConnectionChange.kt → node-api/src/main/kotlin/net/corda/nodeapi/internal/protonwrapper/netty/ConnectionChange.kt
View File

@ -1,4 +1,4 @@
package net.corda.node.internal.protonwrapper.netty
package net.corda.nodeapi.internal.protonwrapper.netty
import java.net.InetSocketAddress
import java.security.cert.X509Certificate

2
node/src/main/kotlin/net/corda/node/internal/protonwrapper/netty/SSLHelper.kt → node-api/src/main/kotlin/net/corda/nodeapi/internal/protonwrapper/netty/SSLHelper.kt
View File

@ -1,4 +1,4 @@
package net.corda.node.internal.protonwrapper.netty
package net.corda.nodeapi.internal.protonwrapper.netty
import io.netty.handler.ssl.SslHandler
import net.corda.core.utilities.NetworkHostAndPort

8
node/src/integration-test/kotlin/net/corda/node/amqp/AMQPBridgeTest.kt
View File

@ -5,15 +5,15 @@ import com.nhaarman.mockito_kotlin.whenever
import net.corda.core.crypto.toStringShort
import net.corda.core.internal.div
import net.corda.core.utilities.NetworkHostAndPort
import net.corda.node.internal.protonwrapper.netty.AMQPServer
import net.corda.node.services.config.CertChainPolicyConfig
import net.corda.node.services.config.NodeConfiguration
import net.corda.node.services.config.configureWithDevSSLCertificate
import net.corda.node.services.messaging.AMQPBridgeManager
import net.corda.node.services.messaging.ArtemisMessagingClient
import net.corda.node.services.messaging.ArtemisMessagingServer
import net.corda.node.services.messaging.BridgeManager
import net.corda.nodeapi.internal.ArtemisMessagingClient
import net.corda.nodeapi.internal.ArtemisMessagingComponent
import net.corda.nodeapi.internal.bridging.AMQPBridgeManager
import net.corda.nodeapi.internal.bridging.BridgeManager
import net.corda.nodeapi.internal.protonwrapper.netty.AMQPServer
import net.corda.testing.core.*
import net.corda.testing.internal.rigorousMock
import org.apache.activemq.artemis.api.core.Message.HDR_DUPLICATE_DETECTION_ID

8
node/src/integration-test/kotlin/net/corda/node/amqp/ProtonWrapperTests.kt
View File

@ -8,16 +8,16 @@ import net.corda.core.identity.CordaX500Name
import net.corda.core.internal.div
import net.corda.core.toFuture
import net.corda.core.utilities.NetworkHostAndPort
import net.corda.node.internal.protonwrapper.messages.MessageStatus
import net.corda.node.internal.protonwrapper.netty.AMQPClient
import net.corda.node.internal.protonwrapper.netty.AMQPServer
import net.corda.node.services.config.CertChainPolicyConfig
import net.corda.node.services.config.NodeConfiguration
import net.corda.node.services.config.configureWithDevSSLCertificate
import net.corda.node.services.messaging.ArtemisMessagingClient
import net.corda.node.services.messaging.ArtemisMessagingServer
import net.corda.nodeapi.internal.ArtemisMessagingClient
import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.P2P_PREFIX
import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.PEER_USER
import net.corda.nodeapi.internal.protonwrapper.messages.MessageStatus
import net.corda.nodeapi.internal.protonwrapper.netty.AMQPClient
import net.corda.nodeapi.internal.protonwrapper.netty.AMQPServer
import net.corda.testing.core.*
import net.corda.testing.internal.rigorousMock
import org.apache.activemq.artemis.api.core.RoutingType

1
node/src/main/kotlin/net/corda/node/internal/Node.kt
View File

@ -33,6 +33,7 @@ import net.corda.node.utilities.AffinityExecutor
import net.corda.node.utilities.DemoClock
import net.corda.nodeapi.internal.ShutdownHook
import net.corda.nodeapi.internal.addShutdownHook
import net.corda.nodeapi.internal.bridging.BridgeControlListener
import net.corda.nodeapi.internal.persistence.CordaPersistence
import net.corda.nodeapi.internal.serialization.*
import net.corda.nodeapi.internal.serialization.amqp.AMQPServerSerializationScheme

23
node/src/main/kotlin/net/corda/node/services/messaging/P2PMessagingClient.kt
View File

@ -22,12 +22,13 @@ import net.corda.node.services.statemachine.FlowMessagingImpl
import net.corda.node.utilities.AffinityExecutor
import net.corda.node.utilities.AppendOnlyPersistentMap
import net.corda.node.utilities.PersistentMap
import net.corda.nodeapi.internal.ArtemisMessagingClient
import net.corda.nodeapi.internal.ArtemisMessagingComponent.*
import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.BRIDGE_CONTROL
import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.BRIDGE_NOTIFY
import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.PEERS_PREFIX
import net.corda.nodeapi.internal.BridgeControl
import net.corda.nodeapi.internal.BridgeEntry
import net.corda.nodeapi.internal.bridging.BridgeControl
import net.corda.nodeapi.internal.bridging.BridgeEntry
import net.corda.nodeapi.internal.persistence.CordaPersistence
import net.corda.nodeapi.internal.persistence.NODE_DATABASE_PREFIX
import org.apache.activemq.artemis.api.core.ActiveMQObjectClosedException
@ -232,14 +233,16 @@ class P2PMessagingClient(config: NodeConfiguration,
val bridgeConsumer = session.createConsumer(bridgeNotifyQueue)
bridgeNotifyConsumer = bridgeConsumer
bridgeConsumer.setMessageHandler { msg ->
val data: ByteArray = ByteArray(msg.bodySize).apply { msg.bodyBuffer.readBytes(this) }
val notifyMessage = data.deserialize<BridgeControl>(context = SerializationDefaults.P2P_CONTEXT)
log.info(notifyMessage.toString())
when (notifyMessage) {
is BridgeControl.BridgeToNodeSnapshotRequest -> enumerateBridges(session, inboxes)
else -> log.error("Unexpected Bridge Control message type on notify topc $notifyMessage")
state.locked {
val data: ByteArray = ByteArray(msg.bodySize).apply { msg.bodyBuffer.readBytes(this) }
val notifyMessage = data.deserialize<BridgeControl>(context = SerializationDefaults.P2P_CONTEXT)
log.info(notifyMessage.toString())
when (notifyMessage) {
is BridgeControl.BridgeToNodeSnapshotRequest -> enumerateBridges(session, inboxes)
else -> log.error("Unexpected Bridge Control message type on notify topc $notifyMessage")
}
msg.acknowledge()
}
msg.acknowledge()
}
networkChangeSubscription = networkMap.changed.subscribe { updateBridgesOnNetworkChange(it) }
}
@ -257,7 +260,7 @@ class P2PMessagingClient(config: NodeConfiguration,
fun gatherAddresses(node: NodeInfo): Sequence<BridgeEntry> {
return node.legalIdentitiesAndCerts.map {
val messagingAddress = NodeAddress(it.party.owningKey, node.addresses.first())
BridgeEntry(messagingAddress.queueName, node.addresses, listOf(it.party.name))
BridgeEntry(messagingAddress.queueName, node.addresses, node.legalIdentities.map { it.name })
}.filter { artemis.started!!.session.queueQuery(SimpleString(it.queueName)).isExists }.asSequence()
}

1
node/src/main/kotlin/net/corda/node/services/messaging/RPCMessagingClient.kt
View File

@ -5,6 +5,7 @@ import net.corda.core.messaging.RPCOps
import net.corda.core.serialization.SingletonSerializeAsToken
import net.corda.core.utilities.NetworkHostAndPort
import net.corda.node.internal.security.RPCSecurityManager
import net.corda.nodeapi.internal.ArtemisMessagingClient
import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.NODE_USER
import net.corda.nodeapi.internal.config.SSLConfiguration
import net.corda.nodeapi.internal.crypto.X509Utilities

1
node/src/main/kotlin/net/corda/node/services/messaging/VerifierMessagingClient.kt
View File

@ -11,6 +11,7 @@ import net.corda.node.utilities.AffinityExecutor
import net.corda.nodeapi.VerifierApi
import net.corda.nodeapi.VerifierApi.VERIFICATION_REQUESTS_QUEUE_NAME
import net.corda.nodeapi.VerifierApi.VERIFICATION_RESPONSES_QUEUE_NAME_PREFIX
import net.corda.nodeapi.internal.ArtemisMessagingClient
import net.corda.nodeapi.internal.config.SSLConfiguration
import org.apache.activemq.artemis.api.core.RoutingType
import org.apache.activemq.artemis.api.core.SimpleString