solution: read upstream config from specified yaml
This commit is contained in:
Igor Artamonov
334
src/main/java/io/emeraldpay/dshackle/config/Upstreams.java
Normal file
334
src/main/java/io/emeraldpay/dshackle/config/Upstreams.java
Normal file
@@ -0,0 +1,334 @@
|
||||
package io.emeraldpay.dshackle.config;
|
||||
|
||||
import org.yaml.snakeyaml.TypeDescription;
|
||||
import org.yaml.snakeyaml.Yaml;
|
||||
import org.yaml.snakeyaml.introspector.MethodProperty;
|
||||
import org.yaml.snakeyaml.introspector.Property;
|
||||
import org.yaml.snakeyaml.introspector.PropertySubstitute;
|
||||
import org.yaml.snakeyaml.nodes.MappingNode;
|
||||
import org.yaml.snakeyaml.nodes.Node;
|
||||
import org.yaml.snakeyaml.nodes.ScalarNode;
|
||||
|
||||
import javax.annotation.Nullable;
|
||||
import java.beans.IntrospectionException;
|
||||
import java.beans.PropertyDescriptor;
|
||||
import java.net.URI;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Optional;
|
||||
|
||||
public class Upstreams {
|
||||
|
||||
private String version;
|
||||
private List<DefaultOptions> defaultOptions;
|
||||
private List<Upstream> upstreams;
|
||||
|
||||
public String getVersion() {
|
||||
return version;
|
||||
}
|
||||
|
||||
public void setVersion(String version) {
|
||||
this.version = version;
|
||||
}
|
||||
|
||||
public List<DefaultOptions> getDefaultOptions() {
|
||||
return defaultOptions;
|
||||
}
|
||||
|
||||
public void setDefaultOptions(List<DefaultOptions> defaultOptions) {
|
||||
this.defaultOptions = defaultOptions;
|
||||
}
|
||||
|
||||
public List<Upstream> getUpstreams() {
|
||||
return upstreams;
|
||||
}
|
||||
|
||||
public void setUpstreams(List<Upstream> upstreams) {
|
||||
this.upstreams = upstreams;
|
||||
}
|
||||
|
||||
public static class Options {
|
||||
private Boolean disableSyncing = true;
|
||||
private Integer minPeers = 1;
|
||||
private Integer quorum = 1;
|
||||
|
||||
public Boolean getDisableSyncing() {
|
||||
return disableSyncing;
|
||||
}
|
||||
|
||||
public void setDisableSyncing(Boolean disableSyncing) {
|
||||
this.disableSyncing = disableSyncing;
|
||||
}
|
||||
|
||||
public Integer getMinPeers() {
|
||||
return minPeers;
|
||||
}
|
||||
|
||||
public void setMinPeers(Integer minPeers) {
|
||||
this.minPeers = minPeers;
|
||||
}
|
||||
|
||||
public Integer getQuorum() {
|
||||
return quorum;
|
||||
}
|
||||
|
||||
public void setQuorum(Integer quorum) {
|
||||
this.quorum = quorum;
|
||||
}
|
||||
}
|
||||
|
||||
public static class OptionsYaml extends TypeDescription {
|
||||
|
||||
public OptionsYaml() {
|
||||
super(Options.class);
|
||||
super.substituteProperty(new PropertySubstitute("disable-syncing", Boolean.class,
|
||||
"getDisableSyncing", "setDisableSyncing"));
|
||||
super.substituteProperty(new PropertySubstitute("min-peers", Integer.class,
|
||||
"getMinPeers", "setMinPeers"));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
public static class DefaultOptions extends Options {
|
||||
private List<String> chains;
|
||||
private Options options;
|
||||
|
||||
public List<String> getChains() {
|
||||
return chains;
|
||||
}
|
||||
|
||||
public void setChains(List<String> chains) {
|
||||
this.chains = chains;
|
||||
}
|
||||
|
||||
public Options getOptions() {
|
||||
return options;
|
||||
}
|
||||
|
||||
public void setOptions(Options options) {
|
||||
this.options = options;
|
||||
}
|
||||
}
|
||||
|
||||
public static class Upstream {
|
||||
private String id;
|
||||
private String chain;
|
||||
|
||||
@Nullable
|
||||
private String provider;
|
||||
private List<Endpoint> endpoints = Collections.emptyList();
|
||||
private Options options;
|
||||
private boolean enabled = true;
|
||||
|
||||
public String getId() {
|
||||
return id;
|
||||
}
|
||||
|
||||
public void setId(String id) {
|
||||
this.id = id;
|
||||
}
|
||||
|
||||
public String getChain() {
|
||||
return chain;
|
||||
}
|
||||
|
||||
public void setChain(String chain) {
|
||||
this.chain = chain;
|
||||
}
|
||||
|
||||
@Nullable
|
||||
public String getProvider() {
|
||||
return provider;
|
||||
}
|
||||
|
||||
public void setProvider(@Nullable String provider) {
|
||||
this.provider = provider;
|
||||
}
|
||||
|
||||
public List<Endpoint> getEndpoints() {
|
||||
return endpoints;
|
||||
}
|
||||
|
||||
public void setEndpoints(List<Endpoint> endpoints) {
|
||||
this.endpoints = endpoints;
|
||||
}
|
||||
|
||||
public Options getOptions() {
|
||||
return options;
|
||||
}
|
||||
|
||||
public void setOptions(Options options) {
|
||||
this.options = options;
|
||||
}
|
||||
|
||||
public boolean isEnabled() {
|
||||
return enabled;
|
||||
}
|
||||
|
||||
public void setEnabled(boolean enabled) {
|
||||
this.enabled = enabled;
|
||||
}
|
||||
}
|
||||
|
||||
public static class Endpoint {
|
||||
private EndpointType type;
|
||||
private URI url;
|
||||
@Nullable
|
||||
private Auth auth;
|
||||
private Boolean enabled = true;
|
||||
@Nullable
|
||||
private URI origin;
|
||||
|
||||
public EndpointType getType() {
|
||||
return type;
|
||||
}
|
||||
|
||||
public void setType(EndpointType type) {
|
||||
this.type = type;
|
||||
}
|
||||
|
||||
public URI getUrl() {
|
||||
return url;
|
||||
}
|
||||
|
||||
public void setUrl(URI url) {
|
||||
this.url = url;
|
||||
}
|
||||
|
||||
@Nullable
|
||||
public Auth getAuth() {
|
||||
return auth;
|
||||
}
|
||||
|
||||
public void setAuth(@Nullable Auth auth) {
|
||||
this.auth = auth;
|
||||
}
|
||||
|
||||
public Boolean getEnabled() {
|
||||
return enabled;
|
||||
}
|
||||
|
||||
public void setEnabled(Boolean enabled) {
|
||||
this.enabled = enabled;
|
||||
}
|
||||
|
||||
@Nullable
|
||||
public URI getOrigin() {
|
||||
return origin;
|
||||
}
|
||||
|
||||
public void setOrigin(@Nullable URI origin) {
|
||||
this.origin = origin;
|
||||
}
|
||||
}
|
||||
|
||||
public static class Auth {
|
||||
private String type;
|
||||
|
||||
public String getType() {
|
||||
return type;
|
||||
}
|
||||
|
||||
public void setType(String type) {
|
||||
this.type = type;
|
||||
}
|
||||
}
|
||||
|
||||
public static class BasicAuth extends Auth {
|
||||
private String key;
|
||||
|
||||
public String getKey() {
|
||||
return key;
|
||||
}
|
||||
|
||||
public void setKey(String key) {
|
||||
this.key = key;
|
||||
}
|
||||
}
|
||||
|
||||
public static class AuthYaml extends TypeDescription {
|
||||
|
||||
public AuthYaml() {
|
||||
super(Auth.class);
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public Property getProperty(String name) {
|
||||
if ("key".equals(name)) {
|
||||
try {
|
||||
return new MethodProperty(new PropertyDescriptor("key", BasicAuth.class, "getKey", "setKey"));
|
||||
} catch (IntrospectionException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
return super.getProperty(name);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object newInstance(Node node) {
|
||||
if (node instanceof MappingNode) {
|
||||
MappingNode mappingNode = (MappingNode)node;
|
||||
Optional<ScalarNode> type = mappingNode.getValue()
|
||||
.stream()
|
||||
.filter((n) -> n.getKeyNode() instanceof ScalarNode && n.getValueNode() instanceof ScalarNode)
|
||||
.filter((n) -> {
|
||||
ScalarNode sn = (ScalarNode)n.getKeyNode();
|
||||
return "type".equals(sn.getValue());
|
||||
})
|
||||
.map((n) -> (ScalarNode)n.getValueNode())
|
||||
.findFirst();
|
||||
if (type.isPresent()) {
|
||||
if ("basic".equals(type.get().getValue())) {
|
||||
return new BasicAuth();
|
||||
} else {
|
||||
throw new IllegalArgumentException("Unsupported auth type: " + type.get().getValue());
|
||||
}
|
||||
} else {
|
||||
throw new IllegalArgumentException("Auth type is not set");
|
||||
}
|
||||
}
|
||||
throw new IllegalArgumentException("Auth is invalid");
|
||||
}
|
||||
}
|
||||
|
||||
public static enum EndpointType {
|
||||
JSON_RPC("json-rpc"),
|
||||
WEBSOCKET("ws", "websocket"),
|
||||
DSHACKLE("dshackle"),
|
||||
UNKNOWN("unknown");
|
||||
|
||||
private final String[] code;
|
||||
|
||||
EndpointType(String ... code) {
|
||||
this.code = code;
|
||||
Arrays.sort(this.code);
|
||||
}
|
||||
|
||||
public static EndpointType byName(String code) {
|
||||
code = code.toLowerCase();
|
||||
for (EndpointType t: EndpointType.values()) {
|
||||
if (Arrays.binarySearch(t.code, code) >= 0) {
|
||||
return t;
|
||||
}
|
||||
}
|
||||
return UNKNOWN;
|
||||
}
|
||||
}
|
||||
|
||||
public static class EndpointTypeYaml extends TypeDescription {
|
||||
|
||||
public EndpointTypeYaml() {
|
||||
super(EndpointType.class);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object newInstance(Node node) {
|
||||
if (node instanceof ScalarNode) {
|
||||
return EndpointType.byName(((ScalarNode)node).getValue());
|
||||
}
|
||||
throw new IllegalArgumentException("Invalid type: " + node.getClass());
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,4 @@
|
||||
package io.emeraldpay.dshackle.config
|
||||
|
||||
class Configuration {
|
||||
}
|
||||
@@ -0,0 +1,17 @@
|
||||
package io.emeraldpay.dshackle.config
|
||||
|
||||
import org.yaml.snakeyaml.Yaml
|
||||
import java.io.InputStream
|
||||
|
||||
class UpstreamsReader {
|
||||
|
||||
fun read(input: InputStream): Upstreams {
|
||||
val yaml = Yaml()
|
||||
yaml.addTypeDescription(Upstreams.EndpointTypeYaml())
|
||||
yaml.addTypeDescription(Upstreams.OptionsYaml())
|
||||
yaml.addTypeDescription(Upstreams.AuthYaml())
|
||||
return yaml.loadAs(input, Upstreams::class.java)
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
@@ -1,12 +1,16 @@
|
||||
package io.emeraldpay.dshackle.upstream
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper
|
||||
import io.emeraldpay.dshackle.config.UpstreamsReader
|
||||
import io.emeraldpay.grpc.Chain
|
||||
import io.infinitape.etherjar.rpc.DefaultRpcClient
|
||||
import io.infinitape.etherjar.rpc.transport.DefaultRpcTransport
|
||||
import org.apache.commons.lang3.StringUtils
|
||||
import org.slf4j.LoggerFactory
|
||||
import org.springframework.beans.factory.annotation.Autowired
|
||||
import org.springframework.core.env.Environment
|
||||
import org.springframework.stereotype.Repository
|
||||
import java.io.File
|
||||
import java.net.URI
|
||||
import javax.annotation.PostConstruct
|
||||
|
||||
@@ -16,48 +20,72 @@ class Upstreams(
|
||||
@Autowired private val objectMapper: ObjectMapper
|
||||
) {
|
||||
|
||||
private val log = LoggerFactory.getLogger(Upstreams::class.java)
|
||||
private var seq = 0
|
||||
private val chainMapping = HashMap<Chain, Upstream>()
|
||||
private val chainMapping = HashMap<Chain, ArrayList<Upstream>>()
|
||||
|
||||
private val chainNames = mapOf(
|
||||
"ethereum" to Chain.ETHEREUM,
|
||||
"ethereum-classic" to Chain.ETHEREUM_CLASSIC,
|
||||
"morden" to Chain.MORDEN
|
||||
)
|
||||
|
||||
@PostConstruct
|
||||
fun start() {
|
||||
env.getProperty("upstream.ethereum")?.let {
|
||||
val api = buildClient(it, Chain.ETHEREUM)
|
||||
chainMapping[Chain.ETHEREUM] = Upstream(Chain.ETHEREUM, api)
|
||||
val path = env.getProperty("upstreams.config")
|
||||
if (StringUtils.isEmpty(path)) {
|
||||
log.error("Path to upstreams is not set (upstreams.config)")
|
||||
System.exit(1)
|
||||
}
|
||||
env.getProperty("upstream.ethereumclassic")?.let {
|
||||
val api = buildClient(it, Chain.ETHEREUM_CLASSIC)
|
||||
val ws = if (env.containsProperty("upstream.ethereumclassic.ws")) {
|
||||
buildWs(env.getProperty("upstream.ethereumclassic.ws")!!, Chain.ETHEREUM_CLASSIC)
|
||||
} else {
|
||||
null
|
||||
val upstreamConfig = File(path)
|
||||
val ok = upstreamConfig.exists() && upstreamConfig.isFile
|
||||
if (!ok) {
|
||||
log.error("Unable to setup upstreams from ${upstreamConfig.path}")
|
||||
System.exit(1)
|
||||
}
|
||||
log.info("Read upstream configuration from ${upstreamConfig.path}")
|
||||
val reader = UpstreamsReader()
|
||||
val config = reader.read(upstreamConfig.inputStream())
|
||||
|
||||
config.upstreams.forEach { up ->
|
||||
val chain = chainNames[up.chain] ?: return@forEach
|
||||
var rpcApi: EthereumApi? = null
|
||||
var wsApi: EthereumWs? = null
|
||||
val urls = ArrayList<URI>()
|
||||
up.endpoints.forEach { endpoint ->
|
||||
if (endpoint.type == io.emeraldpay.dshackle.config.Upstreams.EndpointType.JSON_RPC) {
|
||||
rpcApi = EthereumApi(
|
||||
DefaultRpcClient(DefaultRpcTransport(endpoint.url)),
|
||||
objectMapper,
|
||||
chain
|
||||
)
|
||||
}
|
||||
if (endpoint.type == io.emeraldpay.dshackle.config.Upstreams.EndpointType.WEBSOCKET) {
|
||||
wsApi = EthereumWs(
|
||||
endpoint.url,
|
||||
endpoint.origin ?: URI("http://localhost")
|
||||
)
|
||||
}
|
||||
urls.add(endpoint.url)
|
||||
}
|
||||
if (rpcApi != null) {
|
||||
log.info("Info using ${chain.chainName} upstream, at ${urls.joinToString()}")
|
||||
val current = chainMapping[chain] ?: ArrayList()
|
||||
current.add(Upstream(chain, rpcApi!!, wsApi))
|
||||
chainMapping[chain] = current
|
||||
}
|
||||
chainMapping[Chain.ETHEREUM_CLASSIC] = Upstream(Chain.ETHEREUM_CLASSIC, api, ws)
|
||||
}
|
||||
env.getProperty("upstream.morden")?.let {
|
||||
val api = buildClient(it, Chain.MORDEN)
|
||||
chainMapping[Chain.MORDEN] = Upstream(Chain.MORDEN, api)
|
||||
}
|
||||
}
|
||||
|
||||
private fun buildClient(url: String, chain: Chain): EthereumApi {
|
||||
return EthereumApi(
|
||||
DefaultRpcClient(DefaultRpcTransport(URI(url))),
|
||||
objectMapper,
|
||||
chain
|
||||
)
|
||||
}
|
||||
|
||||
private fun buildWs(url: String, chain: Chain): EthereumWs {
|
||||
val ws = EthereumWs(
|
||||
URI(url),
|
||||
URI("http://localhost")
|
||||
)
|
||||
ws.connect()
|
||||
return ws
|
||||
}
|
||||
|
||||
fun ethereumUpstream(chain: Chain): Upstream? {
|
||||
return chainMapping[chain]
|
||||
val list = chainMapping[chain]
|
||||
if (list == null || list.isEmpty()) {
|
||||
return null
|
||||
}
|
||||
val i = seq++
|
||||
if (seq >= Int.MAX_VALUE / 2) {
|
||||
seq = 0
|
||||
}
|
||||
return list.get(i % list.size)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,57 @@
|
||||
package io.emeraldpay.dshackle.config
|
||||
|
||||
import spock.lang.Specification
|
||||
|
||||
class UpstreamsReaderSpec extends Specification {
|
||||
|
||||
UpstreamsReader reader = new UpstreamsReader()
|
||||
|
||||
def "Parse standard config"() {
|
||||
setup:
|
||||
def config = this.class.getClassLoader().getResourceAsStream("upstreams-basic.yaml")
|
||||
when:
|
||||
def act = reader.read(config)
|
||||
then:
|
||||
act != null
|
||||
act.version == "v1"
|
||||
with(act.defaultOptions) {
|
||||
size() == 1
|
||||
with(get(0)) {
|
||||
chains == ["ethereum"]
|
||||
options.quorum == 1
|
||||
options.minPeers == 3
|
||||
options.disableSyncing
|
||||
}
|
||||
}
|
||||
act.upstreams.size() == 2
|
||||
with(act.upstreams.get(0)) {
|
||||
id == "local"
|
||||
chain == "ethereum"
|
||||
provider == "geth"
|
||||
endpoints.size() == 2
|
||||
with(endpoints.get(0)) {
|
||||
type == Upstreams.EndpointType.JSON_RPC
|
||||
url == new URI("http://localhost:8545")
|
||||
}
|
||||
with(endpoints.get(1)) {
|
||||
type == Upstreams.EndpointType.WEBSOCKET
|
||||
url == new URI("ws://localhost:8546")
|
||||
}
|
||||
}
|
||||
with(act.upstreams.get(1)) {
|
||||
id == "infura"
|
||||
chain == "ethereum"
|
||||
provider == "infura"
|
||||
endpoints.size() == 1
|
||||
with(endpoints.get(0)) {
|
||||
type == Upstreams.EndpointType.JSON_RPC
|
||||
url == new URI("https://mainnet.infura.io/v3/fa28c968191849c1aff541ad1d8511f2")
|
||||
auth instanceof Upstreams.BasicAuth
|
||||
with((Upstreams.BasicAuth)auth) {
|
||||
key == "4fc258fe41a68149c199ad8f281f2015"
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
}
|
||||
29
src/test/resources/upstreams-basic.yaml
Normal file
29
src/test/resources/upstreams-basic.yaml
Normal file
@@ -0,0 +1,29 @@
|
||||
version: v1
|
||||
|
||||
defaultOptions:
|
||||
- chains:
|
||||
- ethereum
|
||||
options:
|
||||
disable-syncing: true
|
||||
min-peers: 3
|
||||
quorum: 1
|
||||
|
||||
upstreams:
|
||||
- id: local
|
||||
chain: ethereum
|
||||
provider: geth
|
||||
endpoints:
|
||||
- type: json-rpc
|
||||
url: "http://localhost:8545"
|
||||
- type: ws
|
||||
url: "ws://localhost:8546"
|
||||
origin: "http://localhost"
|
||||
- id: infura
|
||||
chain: ethereum
|
||||
provider: infura
|
||||
endpoints:
|
||||
- type: json-rpc
|
||||
url: "https://mainnet.infura.io/v3/fa28c968191849c1aff541ad1d8511f2"
|
||||
auth:
|
||||
type: basic
|
||||
key: 4fc258fe41a68149c199ad8f281f2015
|
||||
Reference in New Issue
Block a user