fixed review comments

This commit is contained in:
Maxksim Fomenkov
2022-09-20 13:07:33 +03:00
parent 26d149c228
commit 21528d45b4
4 changed files with 20 additions and 15 deletions

View File

@@ -63,6 +63,12 @@ open class NativeSubscribe(
val nonce = request.nonce.takeIf { it != 0L }
val matcher = Selector.convertToMatcher(request.selector)
/**
* Try to proxy request subscription directly to the upstream dshackle instance.
* If not possible - performs subscription logic on the current instance
* @see EthereumLikeMultistream.tryProxy
*/
val publisher = getUpstream(chain)?.tryProxy(matcher, request) ?: run {
val method = request.method
val params: Any? = request.payload?.takeIf { !it.isEmpty }?.let {

View File

@@ -12,5 +12,12 @@ interface EthereumLikeMultistream : Upstream {
fun getHead(mather: Selector.Matcher): Head
fun tryProxy(mather: Selector.Matcher, request: BlockchainOuterClass.NativeSubscribeRequest): Flux<out Any>?
/**
* Tries to proxy the native subscribe request to the managed upstreams if
* - any of them matches the matcher criteria
* - all of matching above are gRPC ones
* in this case the upstream dshackle instances can sign the results and they will just proxied as is with original signs
* Otherwise return null
*/
fun tryProxy(matcher: Selector.Matcher, request: BlockchainOuterClass.NativeSubscribeRequest): Flux<out Any>?
}

View File

@@ -97,21 +97,17 @@ open class EthereumMultistream(
}
override fun tryProxy(
mather: Selector.Matcher,
matcher: Selector.Matcher,
request: BlockchainOuterClass.NativeSubscribeRequest
): Flux<out Any>? =
upstreams.filter {
mather.matches(it)
matcher.matches(it)
}.takeIf { ups ->
ups.all { it.isGrpc() }
}?.map {
it as GrpcUpstream
}?.map {
it.getBlockchainApi().nativeSubscribe(
request.toBuilder()
.setSelector(BlockchainOuterClass.Selector.getDefaultInstance())
.build()
)
it.getBlockchainApi().nativeSubscribe(request)
}?.let {
Flux.merge(it)
}

View File

@@ -91,19 +91,15 @@ open class EthereumPosMultiStream(
return head!!
}
override fun tryProxy(mather: Selector.Matcher, request: BlockchainOuterClass.NativeSubscribeRequest): Flux<out Any>? =
override fun tryProxy(matcher: Selector.Matcher, request: BlockchainOuterClass.NativeSubscribeRequest): Flux<out Any>? =
upstreams.filter {
mather.matches(it)
matcher.matches(it)
}.takeIf { ups ->
ups.all { it.isGrpc() }
}?.map {
it as GrpcUpstream
}?.map {
it.getBlockchainApi().nativeSubscribe(
request.toBuilder()
.setSelector(BlockchainOuterClass.Selector.getDefaultInstance())
.build()
)
it.getBlockchainApi().nativeSubscribe(request)
}?.let {
Flux.merge(it)
}