Merge pull request #98 from p2p-org/fix-upstream-id

now return first resolver as upstream-id in case of quorum rpc read.

src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt

@@ -119,7 +119,7 @@ class QuorumRpcReader(
                 .filter { it.isResolved() } // return nothing if not resolved
                 .map { quorum ->
                     // TODO find actual quorum number
-                    Result(quorum.getResult()!!, quorum.getSignature(), 1, quorum.getResolvedBy().map { it.nodeId() })
+                    Result(quorum.getResult()!!, quorum.getSignature(), 1, quorum.getResolvedBy())
                 }
                 .switchIfEmpty(defaultResult)
         }
@@ -198,6 +198,6 @@ class QuorumRpcReader(
         val value: ByteArray,
         val signature: ResponseSigner.Signature?,
         val quorum: Int,
-        val resolvers: Collection<Byte>
+        val resolvers: Collection<Upstream>
     )
 }

src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt

@@ -311,7 +311,7 @@ open class NativeCall(
             .map {
                 val bytes = ctx.resultDecorator.processResult(it)
                 validateResult(bytes, "remote", ctx)
-                CallResult.ok(ctx.id, ctx.nonce, bytes, it.signature, ctx.upstream.getId(), ctx)
+                CallResult.ok(ctx.id, ctx.nonce, bytes, it.signature, it.resolvers.first().getId(), ctx)
             }
             .onErrorResume { t ->
                 Mono.just(CallResult.fail(ctx.id, ctx.nonce, t, ctx))
@@ -393,7 +393,11 @@ open class NativeCall(
         override fun processResult(result: QuorumRpcReader.Result): ByteArray {
             val bytes = result.value
             if (bytes.last() == quoteCode) {
-                val suffix = result.resolvers.first().toUByte().toString(16).padStart(2, padChar = '0').toByteArray()
+                val suffix = result.resolvers
+                    .map { it.nodeId() }
+                    .first()
+                    .toUByte()
+                    .toString(16).padStart(2, padChar = '0').toByteArray()
                 bytes[bytes.lastIndex] = suffix.first()
                 return bytes + suffix.last() + quoteCode
             }

src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy

@@ -32,6 +32,7 @@ import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Multistream
 import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.MultistreamHolder
+import io.emeraldpay.dshackle.upstream.Upstream
 import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
 import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
@@ -119,10 +120,14 @@ class NativeCallSpec extends Specification {
         setup:
         def quorum = new AlwaysQuorum()
 
+        def ups = Mock(Upstream) {
+            _ * nodeId() >> (byte) 1
+        }
+
         def nativeCall = nativeCall()
         nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) {
             1 * create(_, _, _) >> Mock(Reader) {
-                1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"foo\"".bytes, null, 1, Collections.singletonList((byte) 1)))
+                1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"foo\"".bytes, null, 1, Collections.singletonList(ups)))
             }
         }
         def call = new NativeCall.ValidCallContext(1, 10, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum,
@@ -586,6 +591,9 @@ class NativeCallSpec extends Specification {
 
     def "Decorate eth_newFilter result"() {
         setup:
+        def ups = Mock(Upstream) {
+            _ * nodeId() >> (byte)255
+        }
         def quorum = new AlwaysQuorum()
         def methods = new ManagedCallMethods(
                 new DefaultEthereumMethods(Chain.ETHEREUM),
@@ -601,7 +609,7 @@ class NativeCallSpec extends Specification {
         def nativeCall = nativeCall(multistreamHolder)
         nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) {
             1 * create(_, _, _) >> Mock(Reader) {
-                1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"0xab\"".bytes, null, 1, Collections.singletonList((byte)255)))
+                1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"0xab\"".bytes, null, 1, Collections.singletonList(ups)))
             }
         }
         def call = new NativeCall.ValidCallContext(1, 10, multistream, Selector.empty, quorum,
@@ -618,6 +626,9 @@ class NativeCallSpec extends Specification {
 
     def "Decorate eth_newFilter result with short nodeId"() {
         setup:
+        def ups = Mock(Upstream) {
+            _ * nodeId() >> (byte)1
+        }
         def quorum = new AlwaysQuorum()
         def methods = new ManagedCallMethods(
                 new DefaultEthereumMethods(Chain.ETHEREUM),
@@ -633,7 +644,7 @@ class NativeCallSpec extends Specification {
         def nativeCall = nativeCall(multistreamHolder)
         nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) {
             1 * create(_, _, _) >> Mock(Reader) {
-                1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"0xab\"".bytes, null, 1, Collections.singletonList((byte)1)))
+                1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"0xab\"".bytes, null, 1, Collections.singletonList(ups)))
             }
         }
         def call = new NativeCall.ValidCallContext(1, 10, multistream, Selector.empty, quorum,