我如何处理提交到 Corda 节点保险库中的新状态的事件？

Question

我需要知道新状态何时被提交到节点的保险库中以获取当时的时间戳。所以，我认为如果我能处理提交状态然后触发我的时间戳记录类会很好。

顺便说一句，如果您对捕获状态随时间演变的时间戳有任何建议，请告诉我。

Answer 1

是的，您可以使用CordaRPCOps.vaultTrackBy 方法执行此操作。此方法返回对 Vault 的可观察更新。您可以订阅此 observable，以便在保险库中记录新状态时收到通知。

RPC 示例

fun main(args: Array<String>) {
    require(args.size == 1) { "Usage: ExampleClientRPC <node address>" }
    val nodeAddress = NetworkHostAndPort.parse(args[0])
    val client = CordaRPCClient(nodeAddress)

    val rpcOps = client.start("user1", "test").proxy

    val updates = rpcOps.vaultTrackBy<ContractState>().updates

    // Log the 'placed' IOU states and listen for new ones.
    updates.toBlocking().subscribe { update ->
        update.produced.forEach { stateAndRef ->
            val timeStamp = Instant.now()
            // TODO("Use the timestamp as you wish.")
        }
    }
}

流量测试示例

class FlowTests {
    lateinit var network: MockNetwork
    lateinit var a: StartedMockNode
    lateinit var b: StartedMockNode

    @Before
    fun setup() {
        network = MockNetwork(
                listOf("com.example.contract"),
                // We need each node to operate on a separate thread so that we can
                // subscribe to the vault updates on a separate thread later.
                threadPerNode = true)
        a = network.createPartyNode()
        b = network.createPartyNode()
        listOf(a, b).forEach { it.registerInitiatedFlow(ExampleFlow.Acceptor::class.java) }
    }

    @After
    fun tearDown() {
        network.stopNodes()
    }

    @Test
    fun `flowTest`() {
        // Trying to access the node database in a separate thread would result in a
        // `java.lang.IllegalStateException: Was expecting to find CordaPersistence set on current thread` exception
        val updates = a.services.vaultService.trackBy<ContractState>().updates

        Thread {
            updates.toBlocking().subscribe { update ->
                update.produced.forEach { stateAndRef ->
                    val timeStamp = Instant.now()
                    // TODO("Use the timestamp as you wish.")
                }
            }
        }.start()

        repeat(3) {
            val flow = ExampleFlow.Initiator(1, b.info.singleIdentity())
            a.startFlow(flow).getOrThrow()
        }

        // We give the other thread time to observe the updates.
        Thread.sleep(10000)
    }
}