StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

solution: support basic auth for websockets

Browse Source
This commit is contained in:
Igor Artamonov
2019-08-15 23:38:12 -04:00
parent 38afb2a1a4
commit 8d83935672
9 changed files with 95 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt
View File

@@ -76,6 +76,7 @@ class UpstreamsConfig {
class WsEndpoint(val url: URI) {
var origin: URI? = null
var basicAuth: BasicAuth? = null
}
open class Auth {

2
src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt
View File

@@ -60,7 +60,7 @@ class UpstreamsConfigReader {
getValueAsString(node, "origin")?.let { origin ->
ws.origin = URI(origin)
}
// ws.auth = readAuth(getMapping(node, "auth"))
ws.basicAuth = readBasicAuth(node)
}
}
} else if (hasAny(connNode, "grpc")) {

23
src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt
View File

@@ -105,7 +105,6 @@ open class ConfiguredUpstreams(
options: UpstreamsConfig.Options,
labels: UpstreamsConfig.Labels) {
var rpcApi: EthereumApi? = null
var wsApi: EthereumWs? = null
val urls = ArrayList<URI>()
up.rpc?.let { endpoint ->
val rpcTransport = DefaultRpcTransport(endpoint.url)
@@ -126,15 +125,21 @@ open class ConfiguredUpstreams(
)
urls.add(endpoint.url)
}
up.ws?.let { endpoint ->
wsApi = EthereumWs(
endpoint.url,
endpoint.origin ?: URI("http://localhost")
)
wsApi!!.connect()
urls.add(endpoint.url)
}
if (rpcApi != null) {
val wsApi: EthereumWs? = up.ws?.let { endpoint ->
val wsApi = EthereumWs(
endpoint.url,
endpoint.origin ?: URI("http://localhost"),
rpcApi!!
)
endpoint.basicAuth?.let { auth ->
wsApi.basicAuth = auth
}
wsApi.connect()
urls.add(endpoint.url)
wsApi
}
log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}")
val ethereumUpstream = EthereumUpstream(chain, rpcApi!!, wsApi, options, NodeDetailsList.NodeDetails(1, labels), targetFor(chain))
ethereumUpstream.start()

23
src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt
View File

@@ -38,13 +38,10 @@ open class EthereumApi(
}
open fun execute(id: Int, method: String, params: List<Any>): Mono<ByteArray> {
val result: Mono<Any> = if (targets.isHardcoded(method)) {
Mono.just(method)
.map { targets.hardcoded(it) }
} else if (targets.isAllowed(method)) {
callUpstream(method, params)
} else {
Mono.error(RpcException(-32601, "Method not allowed or not found"))
val result: Mono<out Any> = when {
targets.isHardcoded(method) -> Mono.just(method).map { targets.hardcoded(it) }
targets.isAllowed(method) -> callUpstream(method, params)
else -> Mono.error(RpcException(-32601, "Method not allowed or not found"))
}
return result
.doOnError { t ->
@@ -72,11 +69,13 @@ open class EthereumApi(
}
}
private fun callUpstream(method: String, params: List<Any>): Mono<Any> {
if (ws != null && method == "eth_blockNumber") {
val head = ws!!.getHead()
if (head != null) {
return Mono.just(HexQuantity.from(head.number).toHex())
private fun callUpstream(method: String, params: List<Any>): Mono<out Any> {
if (method == "eth_blockNumber") {
val current = upstream?.getHead()?.getHead()?.let { head ->
head.map { HexQuantity.from(it.number).toHex() }
}
if (current != null) {
return current
}
}
return Mono.fromCompletionStage(

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt
View File

@@ -57,7 +57,9 @@ open class EthereumUpstream(
open fun createHead(): EthereumHead {
return if (ethereumWs != null) {
EthereumWsHead(ethereumWs)
EthereumWsHead(ethereumWs).apply {
this.start()
}
} else {
EthereumRpcHead(api).apply {
this.start()

36
src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWs.kt
View File

@@ -1,18 +1,22 @@
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.Commands
import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.ws.WebsocketClient
import org.slf4j.LoggerFactory
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.core.publisher.TopicProcessor
import reactor.retry.Repeat
import java.net.URI
import java.time.Duration
import java.util.concurrent.atomic.AtomicReference
class EthereumWs(
private val uri: URI,
private val origin: URI
private val origin: URI,
private val api: EthereumApi
) {
private val log = LoggerFactory.getLogger(EthereumWs::class.java)
@@ -20,20 +24,34 @@ class EthereumWs(
.builder<BlockJson<TransactionId>>()
.name("new-blocks")
.build()
private val head = AtomicReference<BlockJson<TransactionId>>(null)
var basicAuth: UpstreamsConfig.BasicAuth? = null
fun connect() {
log.info("Connecting to WebSocket: $uri")
val client = WebsocketClient()
val client = WebsocketClient(uri, origin)
basicAuth?.let { auth ->
client.setBasicAuth(auth.username, auth.password)
}
try {
client.connect(uri, origin)
client.connect()
} catch (e: Exception) {
log.error("Failed to connect to websocket at $uri. Error: ${e.message}")
return
}
client.onNewBlock {
head.set(it)
topic.onNext(it)
if (it.totalDifficulty == null || it.transactions == null) {
Mono.just(it.hash).flatMap { hash ->
api.executeAndConvert(Commands.eth().getBlock(hash))
}.repeatWhenEmpty { n ->
Repeat.times<Any>(10)
.exponentialBackoff(Duration.ofMillis(50), Duration.ofMillis(250))
.apply(n)
}
.timeout(Duration.ofSeconds(5), Mono.empty())
.subscribe(topic::onNext)
} else {
topic.onNext(it)
}
}
}
@@ -42,8 +60,4 @@ class EthereumWs(
.onBackpressureLatest()
.sample(Duration.ofMillis(100))
}
fun getHead(): BlockJson<TransactionId>? {
return head.get()
}
}

36
src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWsHead.kt
View File

@@ -1,28 +1,56 @@
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.grpc.Chain
import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.json.BlockJson
import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle
import reactor.core.Disposable
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import java.util.concurrent.atomic.AtomicReference
class EthereumWsHead(
private val ws: EthereumWs
): EthereumHead {
): EthereumHead, Lifecycle {
private val log = LoggerFactory.getLogger(EthereumWsHead::class.java)
private var subscription: Disposable? = null
private val head = AtomicReference<BlockJson<TransactionId>>(null)
private val stream: Flux<BlockJson<TransactionId>> = ws.getFlux()
private var stream: Flux<BlockJson<TransactionId>>? = null
override fun getHead(): Mono<BlockJson<TransactionId>> {
val current = head.get()
if (current != null) {
return Mono.just(current)
}
return Mono.from(stream)
return Mono.from(getFlux())
}
override fun getFlux(): Flux<BlockJson<TransactionId>> {
return ws.getFlux()
return stream?.let { Flux.from(it) } ?: Flux.error(Exception("Not started"))
}
override fun isRunning(): Boolean {
return subscription != null
}
override fun start() {
val flux = ws.getFlux()
.distinctUntilChanged { it.hash }
.filter { block ->
val curr = head.get()
curr == null || curr.totalDifficulty < block.totalDifficulty
}.share()
this.subscription = flux.subscribe(head::set)
this.stream = flux
}
override fun stop() {
subscription?.dispose()
subscription = null
}
}