forked from hugegraph/hugegraph-sync
feat(pd): integrate `pd-client` submodule
This commit is contained in:
parent
bd1d9db77b
commit
3a1618faa2
|
|
@ -0,0 +1,73 @@
|
|||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<!--
|
||||
Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
contributor license agreements. See the NOTICE file distributed with
|
||||
this work for additional information regarding copyright ownership.
|
||||
The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
(the "License"); you may not use this file except in compliance with
|
||||
the License. You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
-->
|
||||
|
||||
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xmlns="http://maven.apache.org/POM/4.0.0"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<parent>
|
||||
<groupId>org.apache.hugegraph</groupId>
|
||||
<artifactId>hugegraph-pd</artifactId>
|
||||
<version>${revision}</version>
|
||||
<relativePath>../pom.xml</relativePath>
|
||||
</parent>
|
||||
<artifactId>hg-pd-client</artifactId>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.projectlombok</groupId>
|
||||
<artifactId>lombok</artifactId>
|
||||
<version>1.18.20</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.logging.log4j</groupId>
|
||||
<artifactId>log4j-slf4j-impl</artifactId>
|
||||
<version>2.17.0</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.hugegraph</groupId>
|
||||
<artifactId>hg-pd-grpc</artifactId>
|
||||
<version>${revision}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.hugegraph</groupId>
|
||||
<artifactId>hg-pd-common</artifactId>
|
||||
<version>${revision}</version>
|
||||
<scope>compile</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>junit</groupId>
|
||||
<artifactId>junit</artifactId>
|
||||
<version>4.13.2</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>commons-io</groupId>
|
||||
<artifactId>commons-io</artifactId>
|
||||
<version>2.8.0</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.yaml</groupId>
|
||||
<artifactId>snakeyaml</artifactId>
|
||||
<version>1.28</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</project>
|
||||
|
|
@ -0,0 +1,265 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.hugegraph.pd.client;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.util.LinkedList;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ConcurrentMap;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
import java.util.function.Predicate;
|
||||
import java.util.stream.Stream;
|
||||
|
||||
import org.apache.hugegraph.pd.common.KVPair;
|
||||
import org.apache.hugegraph.pd.common.PDException;
|
||||
import org.apache.hugegraph.pd.grpc.Metapb;
|
||||
import org.apache.hugegraph.pd.grpc.PDGrpc;
|
||||
import org.apache.hugegraph.pd.grpc.PDGrpc.PDBlockingStub;
|
||||
import org.apache.hugegraph.pd.grpc.Pdpb;
|
||||
import org.apache.hugegraph.pd.grpc.Pdpb.GetMembersRequest;
|
||||
import org.apache.hugegraph.pd.grpc.Pdpb.GetMembersResponse;
|
||||
|
||||
import io.grpc.Channel;
|
||||
import io.grpc.ClientCall;
|
||||
import io.grpc.ManagedChannel;
|
||||
import io.grpc.ManagedChannelBuilder;
|
||||
import io.grpc.MethodDescriptor;
|
||||
import io.grpc.StatusRuntimeException;
|
||||
import io.grpc.stub.AbstractBlockingStub;
|
||||
import io.grpc.stub.AbstractStub;
|
||||
import io.grpc.stub.ClientCalls;
|
||||
import io.grpc.stub.StreamObserver;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
@Slf4j
|
||||
public abstract class AbstractClient implements Closeable {
|
||||
|
||||
private static final ConcurrentHashMap<String, ManagedChannel> chs = new ConcurrentHashMap<>();
|
||||
public static Pdpb.ResponseHeader okHeader = Pdpb.ResponseHeader.newBuilder().setError(
|
||||
Pdpb.Error.newBuilder().setType(Pdpb.ErrorType.OK)).build();
|
||||
protected final Pdpb.RequestHeader header;
|
||||
protected final AbstractClientStubProxy stubProxy;
|
||||
protected final PDConfig config;
|
||||
protected ManagedChannel channel = null;
|
||||
protected volatile ConcurrentMap<String, AbstractBlockingStub> stubs = null;
|
||||
|
||||
protected AbstractClient(PDConfig config) {
|
||||
String[] hosts = config.getServerHost().split(",");
|
||||
this.stubProxy = new AbstractClientStubProxy(hosts);
|
||||
this.header = Pdpb.RequestHeader.getDefaultInstance();
|
||||
this.config = config;
|
||||
}
|
||||
|
||||
public static Pdpb.ResponseHeader newErrorHeader(int errorCode, String errorMsg) {
|
||||
Pdpb.ResponseHeader header = Pdpb.ResponseHeader.newBuilder().setError(
|
||||
Pdpb.Error.newBuilder().setTypeValue(errorCode).setMessage(errorMsg)).build();
|
||||
return header;
|
||||
}
|
||||
|
||||
protected static void handleErrors(Pdpb.ResponseHeader header) throws PDException {
|
||||
if (header.hasError() && header.getError().getType() != Pdpb.ErrorType.OK) {
|
||||
throw new PDException(header.getError().getTypeValue(),
|
||||
String.format("PD request error, error code = %d, msg = %s",
|
||||
header.getError().getTypeValue(),
|
||||
header.getError().getMessage()));
|
||||
}
|
||||
}
|
||||
|
||||
protected AbstractBlockingStub getBlockingStub() throws PDException {
|
||||
if (stubProxy.getBlockingStub() == null) {
|
||||
synchronized (this) {
|
||||
if (stubProxy.getBlockingStub() == null) {
|
||||
String host = resetStub();
|
||||
if (host.isEmpty()) {
|
||||
throw new PDException(Pdpb.ErrorType.PD_UNREACHABLE_VALUE,
|
||||
"PD unreachable, pd.peers=" +
|
||||
config.getServerHost());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return (AbstractBlockingStub) stubProxy.getBlockingStub()
|
||||
.withDeadlineAfter(config.getGrpcTimeOut(),
|
||||
TimeUnit.MILLISECONDS);
|
||||
}
|
||||
|
||||
protected AbstractStub getStub() throws PDException {
|
||||
if (stubProxy.getStub() == null) {
|
||||
synchronized (this) {
|
||||
if (stubProxy.getStub() == null) {
|
||||
String host = resetStub();
|
||||
if (host.isEmpty()) {
|
||||
throw new PDException(Pdpb.ErrorType.PD_UNREACHABLE_VALUE,
|
||||
"PD unreachable, pd.peers=" +
|
||||
config.getServerHost());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return stubProxy.getStub();
|
||||
}
|
||||
|
||||
protected abstract AbstractStub createStub();
|
||||
|
||||
protected abstract AbstractBlockingStub createBlockingStub();
|
||||
|
||||
private String resetStub() {
|
||||
String leaderHost = "";
|
||||
for (int i = 0; i < stubProxy.getHostCount(); i++) {
|
||||
String host = stubProxy.nextHost();
|
||||
channel = ManagedChannelBuilder.forTarget(host).usePlaintext().build();
|
||||
PDBlockingStub blockingStub = PDGrpc.newBlockingStub(channel)
|
||||
.withDeadlineAfter(config.getGrpcTimeOut(),
|
||||
TimeUnit.MILLISECONDS);
|
||||
try {
|
||||
GetMembersRequest request = Pdpb.GetMembersRequest.newBuilder()
|
||||
.setHeader(header).build();
|
||||
GetMembersResponse members = blockingStub.getMembers(request);
|
||||
Metapb.Member leader = members.getLeader();
|
||||
leaderHost = leader.getGrpcUrl();
|
||||
close();
|
||||
channel = ManagedChannelBuilder.forTarget(leaderHost).usePlaintext().build();
|
||||
stubProxy.setBlockingStub(createBlockingStub());
|
||||
stubProxy.setStub(createStub());
|
||||
log.info("PDClient connect to host = {} success", leaderHost);
|
||||
break;
|
||||
} catch (Exception e) {
|
||||
log.error("PDClient connect to {} exception {}, {}", host, e.getMessage(),
|
||||
e.getCause() != null ? e.getCause().getMessage() : "");
|
||||
}
|
||||
}
|
||||
return leaderHost;
|
||||
}
|
||||
|
||||
protected <ReqT, RespT, StubT extends AbstractBlockingStub<StubT>> RespT blockingUnaryCall(
|
||||
MethodDescriptor<ReqT, RespT> method, ReqT req) throws PDException {
|
||||
return blockingUnaryCall(method, req, 5);
|
||||
}
|
||||
|
||||
protected <ReqT, RespT, StubT extends AbstractBlockingStub<StubT>> RespT blockingUnaryCall(
|
||||
MethodDescriptor<ReqT, RespT> method, ReqT req, int retry) throws PDException {
|
||||
AbstractBlockingStub stub = getBlockingStub();
|
||||
try {
|
||||
RespT resp =
|
||||
ClientCalls.blockingUnaryCall(stub.getChannel(), method, stub.getCallOptions(),
|
||||
req);
|
||||
return resp;
|
||||
} catch (Exception e) {
|
||||
log.error(method.getFullMethodName() + " exception, {}", e.getMessage());
|
||||
if (e instanceof StatusRuntimeException) {
|
||||
if (retry < stubProxy.getHostCount()) {
|
||||
// 网络不通，关掉之前连接，换host重新连接
|
||||
synchronized (this) {
|
||||
stubProxy.setBlockingStub(null);
|
||||
}
|
||||
return blockingUnaryCall(method, req, ++retry);
|
||||
}
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
// this.stubs = new ConcurrentHashMap<String,AbstractBlockingStub>(hosts.length);
|
||||
private AbstractBlockingStub getConcurrentBlockingStub(String address) {
|
||||
AbstractBlockingStub stub = stubs.get(address);
|
||||
if (stub != null) {
|
||||
return stub;
|
||||
}
|
||||
Channel ch = ManagedChannelBuilder.forTarget(address).usePlaintext().build();
|
||||
PDBlockingStub blockingStub =
|
||||
PDGrpc.newBlockingStub(ch).withDeadlineAfter(config.getGrpcTimeOut(),
|
||||
TimeUnit.MILLISECONDS);
|
||||
stubs.put(address, blockingStub);
|
||||
return blockingStub;
|
||||
|
||||
}
|
||||
|
||||
protected <ReqT, RespT> KVPair<Boolean, RespT> concurrentBlockingUnaryCall(
|
||||
MethodDescriptor<ReqT, RespT> method, ReqT req, Predicate<RespT> predicate) {
|
||||
LinkedList<String> hostList = this.stubProxy.getHostList();
|
||||
if (this.stubs == null) {
|
||||
synchronized (this) {
|
||||
if (this.stubs == null) {
|
||||
this.stubs = new ConcurrentHashMap<>(hostList.size());
|
||||
}
|
||||
}
|
||||
}
|
||||
Stream<RespT> respTStream = hostList.parallelStream().map((address) -> {
|
||||
AbstractBlockingStub stub = getConcurrentBlockingStub(address);
|
||||
RespT resp = ClientCalls.blockingUnaryCall(stub.getChannel(),
|
||||
method, stub.getCallOptions(), req);
|
||||
return resp;
|
||||
});
|
||||
KVPair<Boolean, RespT> pair;
|
||||
AtomicReference<RespT> response = new AtomicReference<>();
|
||||
boolean result = respTStream.anyMatch((r) -> {
|
||||
response.set(r);
|
||||
return predicate.test(r);
|
||||
});
|
||||
if (result) {
|
||||
pair = new KVPair<>(true, null);
|
||||
} else {
|
||||
pair = new KVPair<>(false, response.get());
|
||||
}
|
||||
return pair;
|
||||
}
|
||||
|
||||
protected <ReqT, RespT> void streamingCall(MethodDescriptor<ReqT, RespT> method, ReqT request,
|
||||
StreamObserver<RespT> responseObserver,
|
||||
int retry) throws PDException {
|
||||
AbstractStub stub = getStub();
|
||||
try {
|
||||
ClientCall<ReqT, RespT> call = stub.getChannel().newCall(method, stub.getCallOptions());
|
||||
ClientCalls.asyncServerStreamingCall(call, request, responseObserver);
|
||||
} catch (Exception e) {
|
||||
if (e instanceof StatusRuntimeException) {
|
||||
if (retry < stubProxy.getHostCount()) {
|
||||
synchronized (this) {
|
||||
stubProxy.setStub(null);
|
||||
}
|
||||
streamingCall(method, request, responseObserver, ++retry);
|
||||
return;
|
||||
}
|
||||
}
|
||||
log.error("rpc call with exception, {}", e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
closeChannel(channel);
|
||||
if (stubs != null) {
|
||||
for (AbstractBlockingStub stub : stubs.values()) {
|
||||
closeChannel((ManagedChannel) stub.getChannel());
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private void closeChannel(ManagedChannel channel) {
|
||||
try {
|
||||
while (channel != null &&
|
||||
!channel.shutdownNow().awaitTermination(100, TimeUnit.MILLISECONDS)) {
|
||||
continue;
|
||||
}
|
||||
} catch (Exception e) {
|
||||
log.info("Close channel with error : ", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,72 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.hugegraph.pd.client;
|
||||
|
||||
import java.util.LinkedList;
|
||||
|
||||
import io.grpc.stub.AbstractBlockingStub;
|
||||
import io.grpc.stub.AbstractStub;
|
||||
|
||||
public class AbstractClientStubProxy {
|
||||
|
||||
private final LinkedList<String> hostList = new LinkedList<>();
|
||||
private AbstractBlockingStub blockingStub;
|
||||
private AbstractStub stub;
|
||||
|
||||
public AbstractClientStubProxy(String[] hosts) {
|
||||
for (String host : hosts) {
|
||||
if (!host.isEmpty()) {
|
||||
hostList.offer(host);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public LinkedList<String> getHostList() {
|
||||
return hostList;
|
||||
}
|
||||
|
||||
public String nextHost() {
|
||||
String host = hostList.poll();
|
||||
hostList.offer(host); //移到尾部
|
||||
return host;
|
||||
}
|
||||
|
||||
public AbstractBlockingStub getBlockingStub() {
|
||||
return this.blockingStub;
|
||||
}
|
||||
|
||||
public void setBlockingStub(AbstractBlockingStub stub) {
|
||||
this.blockingStub = stub;
|
||||
}
|
||||
|
||||
public String getHost() {
|
||||
return hostList.peek();
|
||||
}
|
||||
|
||||
public int getHostCount() {
|
||||
return hostList.size();
|
||||
}
|
||||
|
||||
public AbstractStub getStub() {
|
||||
return stub;
|
||||
}
|
||||
|
||||
public void setStub(AbstractStub stub) {
|
||||
this.stub = stub;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,44 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.hugegraph.pd.client;
|
||||
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
import io.grpc.ManagedChannel;
|
||||
import io.grpc.ManagedChannelBuilder;
|
||||
|
||||
public class Channels {
|
||||
|
||||
private static final ConcurrentHashMap<String, ManagedChannel> chs = new ConcurrentHashMap<>();
|
||||
|
||||
public static ManagedChannel getChannel(String target) {
|
||||
|
||||
ManagedChannel channel;
|
||||
if ((channel = chs.get(target)) == null || channel.isShutdown() || channel.isTerminated()) {
|
||||
synchronized (chs) {
|
||||
if ((channel = chs.get(target)) == null || channel.isShutdown() ||
|
||||
channel.isTerminated()) {
|
||||
channel = ManagedChannelBuilder.forTarget(target).usePlaintext().build();
|
||||
chs.put(target, channel);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return channel;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,338 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.hugegraph.pd.client;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Map.Entry;
|
||||
import java.util.Objects;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
import org.apache.hugegraph.pd.common.GraphCache;
|
||||
import org.apache.hugegraph.pd.common.KVPair;
|
||||
import org.apache.hugegraph.pd.common.PDException;
|
||||
import org.apache.hugegraph.pd.common.PartitionUtils;
|
||||
import org.apache.hugegraph.pd.grpc.Metapb;
|
||||
import org.apache.hugegraph.pd.grpc.Metapb.Partition;
|
||||
import org.apache.hugegraph.pd.grpc.Metapb.Shard;
|
||||
import org.apache.hugegraph.pd.grpc.Metapb.ShardGroup;
|
||||
import org.apache.hugegraph.pd.grpc.Pdpb.CachePartitionResponse;
|
||||
import org.apache.hugegraph.pd.grpc.Pdpb.CacheResponse;
|
||||
|
||||
import com.google.common.collect.Range;
|
||||
import com.google.common.collect.RangeMap;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
@Slf4j
|
||||
public class ClientCache {
|
||||
|
||||
private final AtomicBoolean initialized = new AtomicBoolean(false);
|
||||
private final org.apache.hugegraph.pd.client.PDClient client;
|
||||
private volatile Map<Integer, KVPair<ShardGroup, Shard>> groups;
|
||||
private volatile Map<Long, Metapb.Store> stores;
|
||||
private volatile Map<String, GraphCache> caches = new ConcurrentHashMap<>();
|
||||
|
||||
public ClientCache(org.apache.hugegraph.pd.client.PDClient pdClient) {
|
||||
groups = new ConcurrentHashMap<>();
|
||||
stores = new ConcurrentHashMap<>();
|
||||
client = pdClient;
|
||||
}
|
||||
|
||||
private GraphCache getGraphCache(String graphName) {
|
||||
GraphCache graph;
|
||||
if ((graph = caches.get(graphName)) == null) {
|
||||
synchronized (caches) {
|
||||
if ((graph = caches.get(graphName)) == null) {
|
||||
graph = new GraphCache();
|
||||
caches.put(graphName, graph);
|
||||
}
|
||||
}
|
||||
}
|
||||
return graph;
|
||||
}
|
||||
|
||||
public KVPair<Partition, Shard> getPartitionById(String graphName, int partId) {
|
||||
try {
|
||||
GraphCache graph = initGraph(graphName);
|
||||
Partition partition = graph.getPartition(partId);
|
||||
Shard shard = groups.get(partId).getValue();
|
||||
if (partition == null || shard == null) {
|
||||
return null;
|
||||
}
|
||||
return new KVPair<>(partition, shard);
|
||||
} catch (Exception e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
|
||||
private KVPair<Partition, Shard> getPair(int partId, GraphCache graph) {
|
||||
Partition p = graph.getPartition(partId);
|
||||
KVPair<ShardGroup, Shard> pair = groups.get(partId);
|
||||
if (p != null && pair != null) {
|
||||
Shard s = pair.getValue();
|
||||
if (s == null) {
|
||||
pair.setValue(getLeader(partId));
|
||||
return new KVPair<>(p, pair.getValue());
|
||||
} else {
|
||||
return new KVPair<>(p, s);
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* 根据key的hashcode返回分区信息
|
||||
*
|
||||
* @param graphName
|
||||
* @param code
|
||||
* @return
|
||||
*/
|
||||
public KVPair<Partition, Shard> getPartitionByCode(String graphName, long code) {
|
||||
try {
|
||||
GraphCache graph = initGraph(graphName);
|
||||
RangeMap<Long, Integer> range = graph.getRange();
|
||||
Integer pId = range.get(code);
|
||||
if (pId != null) {
|
||||
return getPair(pId, graph);
|
||||
}
|
||||
return null;
|
||||
} catch (PDException e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
|
||||
private GraphCache initGraph(String graphName) throws PDException {
|
||||
initCache();
|
||||
GraphCache graph = getGraphCache(graphName);
|
||||
if (!graph.getInitialized().get()) {
|
||||
synchronized (graph) {
|
||||
if (!graph.getInitialized().get()) {
|
||||
CachePartitionResponse pc = client.getPartitionCache(graphName);
|
||||
RangeMap<Long, Integer> range = graph.getRange();
|
||||
List<Partition> ps = pc.getPartitionsList();
|
||||
HashMap<Integer, Partition> gps = new HashMap<>(ps.size(), 1);
|
||||
for (Partition p : ps) {
|
||||
gps.put(p.getId(), p);
|
||||
range.put(Range.closedOpen(p.getStartKey(), p.getEndKey()), p.getId());
|
||||
}
|
||||
graph.setPartitions(gps);
|
||||
graph.getInitialized().set(true);
|
||||
}
|
||||
}
|
||||
}
|
||||
return graph;
|
||||
}
|
||||
|
||||
private void initCache() throws PDException {
|
||||
if (!initialized.get()) {
|
||||
synchronized (this) {
|
||||
if (!initialized.get()) {
|
||||
CacheResponse cache = client.getClientCache();
|
||||
List<ShardGroup> shardGroups = cache.getShardsList();
|
||||
for (ShardGroup s : shardGroups) {
|
||||
this.groups.put(s.getId(), new KVPair<>(s, getLeader(s.getId())));
|
||||
}
|
||||
List<Metapb.Store> stores = cache.getStoresList();
|
||||
for (Metapb.Store store : stores) {
|
||||
this.stores.put(store.getId(), store);
|
||||
}
|
||||
List<Metapb.Graph> graphs = cache.getGraphsList();
|
||||
for (Metapb.Graph g : graphs) {
|
||||
GraphCache c = new GraphCache(g);
|
||||
caches.put(g.getGraphName(), c);
|
||||
}
|
||||
initialized.set(true);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 返回key所在的分区信息
|
||||
*
|
||||
* @param key
|
||||
* @return
|
||||
*/
|
||||
public KVPair<Partition, Shard> getPartitionByKey(String graphName, byte[] key) {
|
||||
int code = PartitionUtils.calcHashcode(key);
|
||||
return getPartitionByCode(graphName, code);
|
||||
}
|
||||
|
||||
public boolean update(String graphName, int partId, Partition partition) {
|
||||
GraphCache graph = getGraphCache(graphName);
|
||||
try {
|
||||
Partition p = graph.getPartition(partId);
|
||||
if (p != null && p.equals(partition)) {
|
||||
return false;
|
||||
}
|
||||
RangeMap<Long, Integer> range = graph.getRange();
|
||||
graph.addPartition(partId, partition);
|
||||
if (p != null) {
|
||||
// old [1-3) 被 [2-3)覆盖了。当 [1-3) 变成[1-2) 不应该删除原先的[1-3)
|
||||
// 当确认老的 start, end 都是自己的时候，才可以删除老的. (即还没覆盖）
|
||||
if (Objects.equals(partition.getId(), range.get(partition.getStartKey())) &&
|
||||
Objects.equals(partition.getId(), range.get(partition.getEndKey() - 1))) {
|
||||
range.remove(range.getEntry(partition.getStartKey()).getKey());
|
||||
}
|
||||
}
|
||||
range.put(Range.closedOpen(partition.getStartKey(), partition.getEndKey()), partId);
|
||||
} catch (Exception e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
public void removePartition(String graphName, int partId) {
|
||||
GraphCache graph = getGraphCache(graphName);
|
||||
Partition p = graph.removePartition(partId);
|
||||
if (p != null) {
|
||||
RangeMap<Long, Integer> range = graph.getRange();
|
||||
if (Objects.equals(p.getId(), range.get(p.getStartKey())) &&
|
||||
Objects.equals(p.getId(), range.get(p.getEndKey() - 1))) {
|
||||
range.remove(range.getEntry(p.getStartKey()).getKey());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* remove all partitions
|
||||
*/
|
||||
public void removePartitions() {
|
||||
for (Entry<String, GraphCache> entry : caches.entrySet()) {
|
||||
removePartitions(entry.getValue());
|
||||
}
|
||||
}
|
||||
|
||||
private void removePartitions(GraphCache graph) {
|
||||
graph.getState().clear();
|
||||
graph.getRange().clear();
|
||||
}
|
||||
|
||||
/**
|
||||
* remove partition cache of graphName
|
||||
*
|
||||
* @param graphName
|
||||
*/
|
||||
public void removeAll(String graphName) {
|
||||
GraphCache graph = caches.get(graphName);
|
||||
if (graph != null) {
|
||||
removePartitions(graph);
|
||||
}
|
||||
}
|
||||
|
||||
public boolean updateShardGroup(ShardGroup shardGroup) {
|
||||
KVPair<ShardGroup, Shard> old = groups.get(shardGroup.getId());
|
||||
Shard leader = getLeader(shardGroup);
|
||||
if (old != null) {
|
||||
old.setKey(shardGroup);
|
||||
old.setValue(leader);
|
||||
return false;
|
||||
}
|
||||
groups.put(shardGroup.getId(), new KVPair<>(shardGroup, leader));
|
||||
return true;
|
||||
}
|
||||
|
||||
public void deleteShardGroup(int shardGroupId) {
|
||||
groups.remove(shardGroupId);
|
||||
}
|
||||
|
||||
public ShardGroup getShardGroup(int groupId) {
|
||||
KVPair<ShardGroup, Shard> pair = groups.get(groupId);
|
||||
if (pair != null) {
|
||||
return pair.getKey();
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
public boolean addStore(Long storeId, Metapb.Store store) {
|
||||
Metapb.Store oldStore = stores.get(storeId);
|
||||
if (oldStore != null && oldStore.equals(store)) {
|
||||
return false;
|
||||
}
|
||||
stores.put(storeId, store);
|
||||
return true;
|
||||
}
|
||||
|
||||
public Metapb.Store getStoreById(Long storeId) {
|
||||
return stores.get(storeId);
|
||||
}
|
||||
|
||||
public void removeStore(Long storeId) {
|
||||
stores.remove(storeId);
|
||||
}
|
||||
|
||||
public void reset() {
|
||||
groups = new ConcurrentHashMap<>();
|
||||
stores = new ConcurrentHashMap<>();
|
||||
caches = new ConcurrentHashMap<>();
|
||||
}
|
||||
|
||||
public Shard getLeader(int partitionId) {
|
||||
KVPair<ShardGroup, Shard> pair = groups.get(partitionId);
|
||||
if (pair != null) {
|
||||
if (pair.getValue() != null) {
|
||||
return pair.getValue();
|
||||
}
|
||||
for (Shard shard : pair.getKey().getShardsList()) {
|
||||
if (shard.getRole() == Metapb.ShardRole.Leader) {
|
||||
pair.setValue(shard);
|
||||
return shard;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
public Shard getLeader(ShardGroup shardGroup) {
|
||||
if (shardGroup != null) {
|
||||
for (Shard shard : shardGroup.getShardsList()) {
|
||||
if (shard.getRole() == Metapb.ShardRole.Leader) {
|
||||
return shard;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
public void updateLeader(int partitionId, Shard leader) {
|
||||
KVPair<ShardGroup, Shard> pair = groups.get(partitionId);
|
||||
if (pair != null && leader != null) {
|
||||
Shard l = getLeader(partitionId);
|
||||
if (l == null || leader.getStoreId() != l.getStoreId()) {
|
||||
ShardGroup shardGroup = pair.getKey();
|
||||
ShardGroup.Builder builder = ShardGroup.newBuilder(shardGroup).clearShards();
|
||||
for (var shard : shardGroup.getShardsList()) {
|
||||
builder.addShards(
|
||||
Shard.newBuilder()
|
||||
.setStoreId(shard.getStoreId())
|
||||
.setRole(shard.getStoreId() == leader.getStoreId() ?
|
||||
Metapb.ShardRole.Leader : Metapb.ShardRole.Follower)
|
||||
.build()
|
||||
);
|
||||
}
|
||||
pair.setKey(builder.build());
|
||||
pair.setValue(leader);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,30 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.hugegraph.pd.client;
|
||||
|
||||
import org.apache.hugegraph.pd.grpc.discovery.NodeInfos;
|
||||
import org.apache.hugegraph.pd.grpc.discovery.Query;
|
||||
|
||||
public interface Discoverable {
|
||||
|
||||
NodeInfos getNodeInfos(Query query);
|
||||
|
||||
void scheduleTask();
|
||||
|
||||
void cancelTask();
|
||||
}
|
||||
|
|
@ -0,0 +1,221 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.hugegraph.pd.client;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.util.LinkedList;
|
||||
import java.util.Timer;
|
||||
import java.util.TimerTask;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.concurrent.locks.ReentrantReadWriteLock;
|
||||
import java.util.function.Consumer;
|
||||
import java.util.function.Function;
|
||||
|
||||
import org.apache.hugegraph.pd.common.PDException;
|
||||
import org.apache.hugegraph.pd.grpc.discovery.DiscoveryServiceGrpc;
|
||||
import org.apache.hugegraph.pd.grpc.discovery.NodeInfo;
|
||||
import org.apache.hugegraph.pd.grpc.discovery.NodeInfos;
|
||||
import org.apache.hugegraph.pd.grpc.discovery.Query;
|
||||
import org.apache.hugegraph.pd.grpc.discovery.RegisterInfo;
|
||||
|
||||
import io.grpc.ManagedChannel;
|
||||
import io.grpc.ManagedChannelBuilder;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
@Slf4j
|
||||
public abstract class DiscoveryClient implements Closeable, Discoverable {
|
||||
|
||||
private final Timer timer = new Timer("serverHeartbeat", true);
|
||||
private final AtomicBoolean requireResetStub = new AtomicBoolean(false);
|
||||
protected int period; //心跳周期
|
||||
LinkedList<String> pdAddresses = new LinkedList<>();
|
||||
ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();
|
||||
private volatile int currentIndex; // 当前在用pd地址位置
|
||||
private int maxTime = 6;
|
||||
private ManagedChannel channel = null;
|
||||
private DiscoveryServiceGrpc.DiscoveryServiceBlockingStub registerStub;
|
||||
private DiscoveryServiceGrpc.DiscoveryServiceBlockingStub blockingStub;
|
||||
|
||||
public DiscoveryClient(String centerAddress, int delay) {
|
||||
String[] addresses = centerAddress.split(",");
|
||||
for (int i = 0; i < addresses.length; i++) {
|
||||
String singleAddress = addresses[i];
|
||||
if (singleAddress == null || singleAddress.length() <= 0) {
|
||||
continue;
|
||||
}
|
||||
pdAddresses.add(addresses[i]);
|
||||
}
|
||||
this.period = delay;
|
||||
if (maxTime < addresses.length) {
|
||||
maxTime = addresses.length;
|
||||
}
|
||||
}
|
||||
|
||||
private <V, R> R tryWithTimes(Function<V, R> function, V v) {
|
||||
R r;
|
||||
Exception ex = null;
|
||||
for (int i = 0; i < maxTime; i++) {
|
||||
try {
|
||||
r = function.apply(v);
|
||||
return r;
|
||||
} catch (Exception e) {
|
||||
requireResetStub.set(true);
|
||||
resetStub();
|
||||
ex = e;
|
||||
}
|
||||
}
|
||||
if (ex != null) {
|
||||
log.error("Try discovery method with error: {}", ex.getMessage());
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
/***
|
||||
* 按照pd列表重置stub
|
||||
*/
|
||||
private void resetStub() {
|
||||
String errLog = null;
|
||||
for (int i = currentIndex + 1; i <= pdAddresses.size() + currentIndex; i++) {
|
||||
currentIndex = i % pdAddresses.size();
|
||||
String singleAddress = pdAddresses.get(currentIndex);
|
||||
try {
|
||||
if (requireResetStub.get()) {
|
||||
resetChannel(singleAddress);
|
||||
}
|
||||
errLog = null;
|
||||
break;
|
||||
} catch (Exception e) {
|
||||
requireResetStub.set(true);
|
||||
if (errLog == null) {
|
||||
errLog = e.getMessage();
|
||||
}
|
||||
continue;
|
||||
}
|
||||
}
|
||||
if (errLog != null) {
|
||||
log.error(errLog);
|
||||
}
|
||||
}
|
||||
|
||||
/***
|
||||
* 按照某个pd的地址重置channel和stub
|
||||
* @param singleAddress
|
||||
* @throws PDException
|
||||
*/
|
||||
private void resetChannel(String singleAddress) throws PDException {
|
||||
|
||||
readWriteLock.writeLock().lock();
|
||||
try {
|
||||
if (requireResetStub.get()) {
|
||||
while (channel != null && !channel.shutdownNow().awaitTermination(
|
||||
100, TimeUnit.MILLISECONDS)) {
|
||||
continue;
|
||||
}
|
||||
channel = ManagedChannelBuilder.forTarget(
|
||||
singleAddress).usePlaintext().build();
|
||||
this.registerStub = DiscoveryServiceGrpc.newBlockingStub(
|
||||
channel);
|
||||
this.blockingStub = DiscoveryServiceGrpc.newBlockingStub(
|
||||
channel);
|
||||
requireResetStub.set(false);
|
||||
}
|
||||
} catch (Exception e) {
|
||||
throw new PDException(-1, String.format(
|
||||
"Reset channel with error : %s.", e.getMessage()));
|
||||
} finally {
|
||||
readWriteLock.writeLock().unlock();
|
||||
}
|
||||
}
|
||||
|
||||
/***
|
||||
* 获取注册节点信息
|
||||
* @param query
|
||||
* @return
|
||||
*/
|
||||
@Override
|
||||
public NodeInfos getNodeInfos(Query query) {
|
||||
return tryWithTimes((q) -> {
|
||||
this.readWriteLock.readLock().lock();
|
||||
NodeInfos nodes;
|
||||
try {
|
||||
nodes = this.blockingStub.getNodes(q);
|
||||
} catch (Exception e) {
|
||||
throw e;
|
||||
} finally {
|
||||
this.readWriteLock.readLock().unlock();
|
||||
}
|
||||
return nodes;
|
||||
}, query);
|
||||
}
|
||||
|
||||
/***
|
||||
* 启动心跳任务
|
||||
*/
|
||||
@Override
|
||||
public void scheduleTask() {
|
||||
timer.schedule(new TimerTask() {
|
||||
@Override
|
||||
public void run() {
|
||||
NodeInfo nodeInfo = getRegisterNode();
|
||||
tryWithTimes((t) -> {
|
||||
RegisterInfo register;
|
||||
readWriteLock.readLock().lock();
|
||||
try {
|
||||
register = registerStub.register(t);
|
||||
log.debug("Discovery Client work done.");
|
||||
Consumer<RegisterInfo> consumer = getRegisterConsumer();
|
||||
if (consumer != null) {
|
||||
consumer.accept(register);
|
||||
}
|
||||
} catch (Exception e) {
|
||||
throw e;
|
||||
} finally {
|
||||
readWriteLock.readLock().unlock();
|
||||
}
|
||||
return register;
|
||||
}, nodeInfo);
|
||||
}
|
||||
}, 0, period);
|
||||
}
|
||||
|
||||
abstract NodeInfo getRegisterNode();
|
||||
|
||||
abstract Consumer<RegisterInfo> getRegisterConsumer();
|
||||
|
||||
@Override
|
||||
public void cancelTask() {
|
||||
this.timer.cancel();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
this.timer.cancel();
|
||||
readWriteLock.writeLock().lock();
|
||||
try {
|
||||
while (channel != null && !channel.shutdownNow().awaitTermination(
|
||||
100, TimeUnit.MILLISECONDS)) {
|
||||
continue;
|
||||
}
|
||||
} catch (Exception e) {
|
||||
log.info("Close channel with error : {}.", e);
|
||||
} finally {
|
||||
readWriteLock.writeLock().unlock();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,137 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.hugegraph.pd.client;
|
||||
|
||||
import java.util.Map;
|
||||
import java.util.function.Consumer;
|
||||
|
||||
import org.apache.hugegraph.pd.grpc.discovery.NodeInfo;
|
||||
import org.apache.hugegraph.pd.grpc.discovery.RegisterType;
|
||||
|
||||
public class DiscoveryClientImpl extends DiscoveryClient {
|
||||
|
||||
private final String id;
|
||||
private final RegisterType type; // 心跳类型，备用
|
||||
private final String version;
|
||||
private final String appName;
|
||||
private final int times; // 心跳过期次数，备用
|
||||
private final String address;
|
||||
private final Map labels;
|
||||
private final Consumer registerConsumer;
|
||||
|
||||
private DiscoveryClientImpl(Builder builder) {
|
||||
super(builder.centerAddress, builder.delay);
|
||||
period = builder.delay;
|
||||
id = builder.id;
|
||||
type = builder.type;
|
||||
version = builder.version;
|
||||
appName = builder.appName;
|
||||
times = builder.times;
|
||||
address = builder.address;
|
||||
labels = builder.labels;
|
||||
registerConsumer = builder.registerConsumer;
|
||||
}
|
||||
|
||||
public static Builder newBuilder() {
|
||||
return new Builder();
|
||||
}
|
||||
|
||||
@Override
|
||||
NodeInfo getRegisterNode() {
|
||||
return NodeInfo.newBuilder().setAddress(this.address)
|
||||
.setVersion(this.version)
|
||||
.setAppName(this.appName).setInterval(this.period)
|
||||
.setId(this.id).putAllLabels(labels).build();
|
||||
}
|
||||
|
||||
@Override
|
||||
Consumer getRegisterConsumer() {
|
||||
return registerConsumer;
|
||||
}
|
||||
|
||||
public static final class Builder {
|
||||
|
||||
private int delay;
|
||||
private String centerAddress;
|
||||
private String id;
|
||||
private RegisterType type;
|
||||
private String address;
|
||||
private Map labels;
|
||||
private String version;
|
||||
private String appName;
|
||||
private int times;
|
||||
private Consumer registerConsumer;
|
||||
|
||||
private Builder() {
|
||||
}
|
||||
|
||||
public Builder setDelay(int val) {
|
||||
delay = val;
|
||||
return this;
|
||||
}
|
||||
|
||||
public Builder setCenterAddress(String val) {
|
||||
centerAddress = val;
|
||||
return this;
|
||||
}
|
||||
|
||||
public Builder setId(String val) {
|
||||
id = val;
|
||||
return this;
|
||||
}
|
||||
|
||||
public Builder setType(RegisterType val) {
|
||||
type = val;
|
||||
return this;
|
||||
}
|
||||
|
||||
public Builder setAddress(String val) {
|
||||
address = val;
|
||||
return this;
|
||||
}
|
||||
|
||||
public Builder setLabels(Map val) {
|
||||
labels = val;
|
||||
return this;
|
||||
}
|
||||
|
||||
public Builder setVersion(String val) {
|
||||
version = val;
|
||||
return this;
|
||||
}
|
||||
|
||||
public Builder setAppName(String val) {
|
||||
appName = val;
|
||||
return this;
|
||||
}
|
||||
|
||||
public Builder setTimes(int val) {
|
||||
times = val;
|
||||
return this;
|
||||
}
|
||||
|
||||
public Builder setRegisterConsumer(Consumer registerConsumer) {
|
||||
this.registerConsumer = registerConsumer;
|
||||
return this;
|
||||
}
|
||||
|
||||
public DiscoveryClientImpl build() {
|
||||
return new DiscoveryClientImpl(this);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,343 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.hugegraph.pd.client;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.util.HashMap;
|
||||
import java.util.LinkedList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.Semaphore;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
import java.util.function.BiConsumer;
|
||||
import java.util.function.Consumer;
|
||||
|
||||
import org.apache.hugegraph.pd.common.PDException;
|
||||
import org.apache.hugegraph.pd.grpc.kv.K;
|
||||
import org.apache.hugegraph.pd.grpc.kv.KResponse;
|
||||
import org.apache.hugegraph.pd.grpc.kv.Kv;
|
||||
import org.apache.hugegraph.pd.grpc.kv.KvResponse;
|
||||
import org.apache.hugegraph.pd.grpc.kv.KvServiceGrpc;
|
||||
import org.apache.hugegraph.pd.grpc.kv.LockRequest;
|
||||
import org.apache.hugegraph.pd.grpc.kv.LockResponse;
|
||||
import org.apache.hugegraph.pd.grpc.kv.ScanPrefixResponse;
|
||||
import org.apache.hugegraph.pd.grpc.kv.TTLRequest;
|
||||
import org.apache.hugegraph.pd.grpc.kv.TTLResponse;
|
||||
import org.apache.hugegraph.pd.grpc.kv.WatchEvent;
|
||||
import org.apache.hugegraph.pd.grpc.kv.WatchKv;
|
||||
import org.apache.hugegraph.pd.grpc.kv.WatchRequest;
|
||||
import org.apache.hugegraph.pd.grpc.kv.WatchResponse;
|
||||
import org.apache.hugegraph.pd.grpc.kv.WatchType;
|
||||
|
||||
import io.grpc.stub.AbstractBlockingStub;
|
||||
import io.grpc.stub.AbstractStub;
|
||||
import io.grpc.stub.StreamObserver;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
@Slf4j
|
||||
public class KvClient<T extends WatchResponse> extends AbstractClient implements Closeable {
|
||||
|
||||
private final AtomicLong clientId = new AtomicLong(0);
|
||||
private final Semaphore semaphore = new Semaphore(1);
|
||||
private final ConcurrentHashMap<Long, StreamObserver> observers = new ConcurrentHashMap<>();
|
||||
|
||||
public KvClient(PDConfig pdConfig) {
|
||||
super(pdConfig);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected AbstractStub createStub() {
|
||||
return KvServiceGrpc.newStub(channel);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected AbstractBlockingStub createBlockingStub() {
|
||||
return KvServiceGrpc.newBlockingStub(channel);
|
||||
}
|
||||
|
||||
public KvResponse put(String key, String value) throws PDException {
|
||||
Kv kv = Kv.newBuilder().setKey(key).setValue(value).build();
|
||||
KvResponse response = blockingUnaryCall(KvServiceGrpc.getPutMethod(), kv);
|
||||
handleErrors(response.getHeader());
|
||||
return response;
|
||||
}
|
||||
|
||||
public KResponse get(String key) throws PDException {
|
||||
K k = K.newBuilder().setKey(key).build();
|
||||
KResponse response = blockingUnaryCall(KvServiceGrpc.getGetMethod(), k);
|
||||
handleErrors(response.getHeader());
|
||||
return response;
|
||||
}
|
||||
|
||||
public KvResponse delete(String key) throws PDException {
|
||||
K k = K.newBuilder().setKey(key).build();
|
||||
KvResponse response = blockingUnaryCall(KvServiceGrpc.getDeleteMethod(), k);
|
||||
handleErrors(response.getHeader());
|
||||
return response;
|
||||
}
|
||||
|
||||
public KvResponse deletePrefix(String prefix) throws PDException {
|
||||
K k = K.newBuilder().setKey(prefix).build();
|
||||
KvResponse response = blockingUnaryCall(KvServiceGrpc.getDeletePrefixMethod(), k);
|
||||
handleErrors(response.getHeader());
|
||||
return response;
|
||||
}
|
||||
|
||||
public ScanPrefixResponse scanPrefix(String prefix) throws PDException {
|
||||
K k = K.newBuilder().setKey(prefix).build();
|
||||
ScanPrefixResponse response = blockingUnaryCall(KvServiceGrpc.getScanPrefixMethod(), k);
|
||||
handleErrors(response.getHeader());
|
||||
return response;
|
||||
}
|
||||
|
||||
public TTLResponse keepTTLAlive(String key) throws PDException {
|
||||
TTLRequest request = TTLRequest.newBuilder().setKey(key).build();
|
||||
TTLResponse response = blockingUnaryCall(KvServiceGrpc.getKeepTTLAliveMethod(), request);
|
||||
handleErrors(response.getHeader());
|
||||
return response;
|
||||
}
|
||||
|
||||
public TTLResponse putTTL(String key, String value, long ttl) throws PDException {
|
||||
TTLRequest request =
|
||||
TTLRequest.newBuilder().setKey(key).setValue(value).setTtl(ttl).build();
|
||||
TTLResponse response = blockingUnaryCall(KvServiceGrpc.getPutTTLMethod(), request);
|
||||
handleErrors(response.getHeader());
|
||||
return response;
|
||||
}
|
||||
|
||||
private void onEvent(WatchResponse value, Consumer<T> consumer) {
|
||||
log.info("receive message for {},event Count:{}", value, value.getEventsCount());
|
||||
clientId.compareAndSet(0L, value.getClientId());
|
||||
if (value.getEventsCount() != 0) {
|
||||
consumer.accept((T) value);
|
||||
}
|
||||
}
|
||||
|
||||
private StreamObserver<WatchResponse> getObserver(String key, Consumer<T> consumer,
|
||||
BiConsumer<String, Consumer> listenWrapper,
|
||||
long client) {
|
||||
StreamObserver<WatchResponse> observer;
|
||||
if ((observer = observers.get(client)) == null) {
|
||||
synchronized (this) {
|
||||
if ((observer = observers.get(client)) == null) {
|
||||
observer = getObserver(key, consumer, listenWrapper);
|
||||
observers.put(client, observer);
|
||||
}
|
||||
}
|
||||
}
|
||||
return observer;
|
||||
}
|
||||
|
||||
private StreamObserver<WatchResponse> getObserver(String key, Consumer<T> consumer,
|
||||
BiConsumer<String, Consumer> listenWrapper) {
|
||||
return new StreamObserver<WatchResponse>() {
|
||||
@Override
|
||||
public void onNext(WatchResponse value) {
|
||||
switch (value.getState()) {
|
||||
case Starting:
|
||||
boolean b = clientId.compareAndSet(0, value.getClientId());
|
||||
if (b) {
|
||||
observers.put(value.getClientId(), this);
|
||||
log.info("set watch client id to :{}", value.getClientId());
|
||||
}
|
||||
semaphore.release();
|
||||
break;
|
||||
case Started:
|
||||
onEvent(value, consumer);
|
||||
break;
|
||||
case Leader_Changed:
|
||||
listenWrapper.accept(key, consumer);
|
||||
break;
|
||||
case Alive:
|
||||
// only for check client is alive, do nothing
|
||||
break;
|
||||
default:
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onError(Throwable t) {
|
||||
listenWrapper.accept(key, consumer);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onCompleted() {
|
||||
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
public void listen(String key, Consumer<T> consumer) throws PDException {
|
||||
long value = clientId.get();
|
||||
StreamObserver<WatchResponse> observer = getObserver(key, consumer, listenWrapper, value);
|
||||
acquire();
|
||||
WatchRequest k = WatchRequest.newBuilder().setClientId(value).setKey(key).build();
|
||||
streamingCall(KvServiceGrpc.getWatchMethod(), k, observer, 1);
|
||||
}
|
||||
|
||||
public void listenPrefix(String prefix, Consumer<T> consumer) throws PDException {
|
||||
long value = clientId.get();
|
||||
StreamObserver<WatchResponse> observer =
|
||||
getObserver(prefix, consumer, prefixListenWrapper, value);
|
||||
acquire();
|
||||
WatchRequest k =
|
||||
WatchRequest.newBuilder().setClientId(clientId.get()).setKey(prefix).build();
|
||||
streamingCall(KvServiceGrpc.getWatchPrefixMethod(), k, observer, 1);
|
||||
}
|
||||
|
||||
private void acquire() {
|
||||
if (clientId.get() == 0L) {
|
||||
try {
|
||||
semaphore.acquire();
|
||||
if (clientId.get() != 0L) {
|
||||
semaphore.release();
|
||||
}
|
||||
} catch (InterruptedException e) {
|
||||
log.error("get semaphore with error:", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public List<String> getWatchList(T response) {
|
||||
List<String> values = new LinkedList<>();
|
||||
List<WatchEvent> eventsList = response.getEventsList();
|
||||
for (WatchEvent event : eventsList) {
|
||||
if (event.getType() != WatchType.Put) {
|
||||
return null;
|
||||
}
|
||||
String value = event.getCurrent().getValue();
|
||||
values.add(value);
|
||||
}
|
||||
return values;
|
||||
}
|
||||
|
||||
public Map<String, String> getWatchMap(T response) {
|
||||
Map<String, String> values = new HashMap<>();
|
||||
List<WatchEvent> eventsList = response.getEventsList();
|
||||
for (WatchEvent event : eventsList) {
|
||||
if (event.getType() != WatchType.Put) {
|
||||
return null;
|
||||
}
|
||||
WatchKv current = event.getCurrent();
|
||||
String key = current.getKey();
|
||||
String value = current.getValue();
|
||||
values.put(key, value);
|
||||
}
|
||||
return values;
|
||||
}
|
||||
|
||||
public LockResponse lock(String key, long ttl) throws PDException {
|
||||
acquire();
|
||||
LockResponse response;
|
||||
try {
|
||||
LockRequest k =
|
||||
LockRequest.newBuilder().setKey(key).setClientId(clientId.get()).setTtl(ttl)
|
||||
.build();
|
||||
response = blockingUnaryCall(KvServiceGrpc.getLockMethod(), k);
|
||||
handleErrors(response.getHeader());
|
||||
if (clientId.compareAndSet(0L, response.getClientId())) {
|
||||
semaphore.release();
|
||||
}
|
||||
} catch (Exception e) {
|
||||
if (clientId.get() == 0L) {
|
||||
semaphore.release();
|
||||
}
|
||||
throw e;
|
||||
}
|
||||
return response;
|
||||
}
|
||||
|
||||
public LockResponse lockWithoutReentrant(String key, long ttl) throws PDException {
|
||||
acquire();
|
||||
LockResponse response;
|
||||
try {
|
||||
LockRequest k =
|
||||
LockRequest.newBuilder().setKey(key).setClientId(clientId.get()).setTtl(ttl)
|
||||
.build();
|
||||
response = blockingUnaryCall(KvServiceGrpc.getLockWithoutReentrantMethod(), k);
|
||||
handleErrors(response.getHeader());
|
||||
if (clientId.compareAndSet(0L, response.getClientId())) {
|
||||
semaphore.release();
|
||||
}
|
||||
} catch (Exception e) {
|
||||
if (clientId.get() == 0L) {
|
||||
semaphore.release();
|
||||
}
|
||||
throw e;
|
||||
}
|
||||
return response;
|
||||
}
|
||||
|
||||
public LockResponse isLocked(String key) throws PDException {
|
||||
LockRequest k = LockRequest.newBuilder().setKey(key).setClientId(clientId.get()).build();
|
||||
LockResponse response = blockingUnaryCall(KvServiceGrpc.getIsLockedMethod(), k);
|
||||
handleErrors(response.getHeader());
|
||||
return response;
|
||||
}
|
||||
|
||||
public LockResponse unlock(String key) throws PDException {
|
||||
assert clientId.get() != 0;
|
||||
LockRequest k = LockRequest.newBuilder().setKey(key).setClientId(clientId.get()).build();
|
||||
LockResponse response = blockingUnaryCall(KvServiceGrpc.getUnlockMethod(), k);
|
||||
handleErrors(response.getHeader());
|
||||
clientId.compareAndSet(0L, response.getClientId());
|
||||
assert clientId.get() == response.getClientId();
|
||||
return response;
|
||||
}
|
||||
|
||||
public LockResponse keepAlive(String key) throws PDException {
|
||||
assert clientId.get() != 0;
|
||||
LockRequest k = LockRequest.newBuilder().setKey(key).setClientId(clientId.get()).build();
|
||||
LockResponse response = blockingUnaryCall(KvServiceGrpc.getKeepAliveMethod(), k);
|
||||
handleErrors(response.getHeader());
|
||||
clientId.compareAndSet(0L, response.getClientId());
|
||||
assert clientId.get() == response.getClientId();
|
||||
return response;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
super.close();
|
||||
}
|
||||
|
||||
BiConsumer<String, Consumer> listenWrapper = (key, consumer) -> {
|
||||
try {
|
||||
listen(key, consumer);
|
||||
} catch (PDException e) {
|
||||
try {
|
||||
log.warn("start listen with warning:", e);
|
||||
Thread.sleep(1000);
|
||||
} catch (InterruptedException ex) {
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
BiConsumer<String, Consumer> prefixListenWrapper = (key, consumer) -> {
|
||||
try {
|
||||
listenPrefix(key, consumer);
|
||||
} catch (PDException e) {
|
||||
try {
|
||||
log.warn("start listenPrefix with warning:", e);
|
||||
Thread.sleep(1000);
|
||||
} catch (InterruptedException ex) {
|
||||
}
|
||||
}
|
||||
};
|
||||
}
|
||||
|
|
@ -0,0 +1,71 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.hugegraph.pd.client;
|
||||
|
||||
import org.apache.hugegraph.pd.common.KVPair;
|
||||
import org.apache.hugegraph.pd.grpc.PDGrpc;
|
||||
import org.apache.hugegraph.pd.grpc.Pdpb;
|
||||
|
||||
import com.google.protobuf.ByteString;
|
||||
|
||||
import io.grpc.stub.AbstractBlockingStub;
|
||||
import io.grpc.stub.AbstractStub;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
@Slf4j
|
||||
public class LicenseClient extends AbstractClient {
|
||||
|
||||
public LicenseClient(PDConfig config) {
|
||||
super(config);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected AbstractStub createStub() {
|
||||
return PDGrpc.newStub(channel);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected AbstractBlockingStub createBlockingStub() {
|
||||
return PDGrpc.newBlockingStub(channel);
|
||||
}
|
||||
|
||||
public Pdpb.PutLicenseResponse putLicense(byte[] content) {
|
||||
Pdpb.PutLicenseRequest request = Pdpb.PutLicenseRequest.newBuilder()
|
||||
.setContent(
|
||||
ByteString.copyFrom(content))
|
||||
.build();
|
||||
try {
|
||||
KVPair<Boolean, Pdpb.PutLicenseResponse> pair = concurrentBlockingUnaryCall(
|
||||
PDGrpc.getPutLicenseMethod(), request,
|
||||
(rs) -> rs.getHeader().getError().getType().equals(Pdpb.ErrorType.OK));
|
||||
if (pair.getKey()) {
|
||||
Pdpb.PutLicenseResponse.Builder builder = Pdpb.PutLicenseResponse.newBuilder();
|
||||
builder.setHeader(okHeader);
|
||||
return builder.build();
|
||||
} else {
|
||||
return pair.getValue();
|
||||
}
|
||||
} catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
log.debug("put license with error:{} ", e);
|
||||
Pdpb.ResponseHeader rh =
|
||||
newErrorHeader(Pdpb.ErrorType.LICENSE_ERROR_VALUE, e.getMessage());
|
||||
return Pdpb.PutLicenseResponse.newBuilder().setHeader(rh).build();
|
||||
}
|
||||
}
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
|
|
@ -0,0 +1,83 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.hugegraph.pd.client;
|
||||
|
||||
public final class PDConfig {
|
||||
|
||||
//TODO multi-server
|
||||
private String serverHost = "localhost:9000";
|
||||
private long grpcTimeOut = 60000; // grpc调用超时时间 10秒
|
||||
|
||||
// 是否接收PD异步通知
|
||||
private boolean enablePDNotify = false;
|
||||
|
||||
private boolean enableCache = false;
|
||||
|
||||
private PDConfig() {
|
||||
}
|
||||
|
||||
public static PDConfig of() {
|
||||
return new PDConfig();
|
||||
}
|
||||
|
||||
public static PDConfig of(String serverHost) {
|
||||
PDConfig config = new PDConfig();
|
||||
config.serverHost = serverHost;
|
||||
return config;
|
||||
}
|
||||
|
||||
public static PDConfig of(String serverHost, long timeOut) {
|
||||
PDConfig config = new PDConfig();
|
||||
config.serverHost = serverHost;
|
||||
config.grpcTimeOut = timeOut;
|
||||
return config;
|
||||
}
|
||||
|
||||
public String getServerHost() {
|
||||
return serverHost;
|
||||
}
|
||||
|
||||
public long getGrpcTimeOut() {
|
||||
return grpcTimeOut;
|
||||
}
|
||||
|
||||
@Deprecated
|
||||
public PDConfig setEnablePDNotify(boolean enablePDNotify) {
|
||||
this.enablePDNotify = enablePDNotify;
|
||||
|
||||
// TODO 临时代码，hugegraph修改完后删除
|
||||
this.enableCache = enablePDNotify;
|
||||
return this;
|
||||
}
|
||||
|
||||
public boolean isEnableCache() {
|
||||
return enableCache;
|
||||
}
|
||||
|
||||
public PDConfig setEnableCache(boolean enableCache) {
|
||||
this.enableCache = enableCache;
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "PDConfig{" +
|
||||
"serverHost='" + serverHost + '\'' +
|
||||
'}';
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,154 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.hugegraph.pd.client;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.util.function.Consumer;
|
||||
|
||||
import org.apache.hugegraph.pd.grpc.pulse.PartitionHeartbeatRequest;
|
||||
import org.apache.hugegraph.pd.grpc.pulse.PulseResponse;
|
||||
import org.apache.hugegraph.pd.pulse.PulseServerNotice;
|
||||
|
||||
/**
|
||||
* Bidirectional communication interface of pd-client and pd-server
|
||||
*/
|
||||
public interface PDPulse {
|
||||
|
||||
/*** inner static methods ***/
|
||||
static <T> Listener<T> listener(Consumer<T> onNext) {
|
||||
return listener(onNext, t -> {
|
||||
}, () -> {
|
||||
});
|
||||
}
|
||||
|
||||
static <T> Listener<T> listener(Consumer<T> onNext, Consumer<Throwable> onError) {
|
||||
return listener(onNext, onError, () -> {
|
||||
});
|
||||
}
|
||||
|
||||
static <T> Listener<T> listener(Consumer<T> onNext, Runnable onCompleted) {
|
||||
return listener(onNext, t -> {
|
||||
}, onCompleted);
|
||||
}
|
||||
|
||||
static <T> Listener<T> listener(Consumer<T> onNext, Consumer<Throwable> onError,
|
||||
Runnable onCompleted) {
|
||||
return new Listener<>() {
|
||||
@Override
|
||||
public void onNext(T response) {
|
||||
onNext.accept(response);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onNotice(PulseServerNotice<T> notice) {
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onError(Throwable throwable) {
|
||||
onError.accept(throwable);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onCompleted() {
|
||||
onCompleted.run();
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* @param listener
|
||||
* @return
|
||||
*/
|
||||
Notifier<PartitionHeartbeatRequest.Builder> connectPartition(Listener<PulseResponse> listener);
|
||||
|
||||
/**
|
||||
* 切换成新的host。做 channel/host的检查，如果需要关闭，notifier调用close方法。
|
||||
*
|
||||
* @param host new host
|
||||
* @param notifier notifier
|
||||
* @return true if create new stub, otherwise false
|
||||
*/
|
||||
boolean resetStub(String host, Notifier notifier);
|
||||
|
||||
/**
|
||||
* Interface of pulse.
|
||||
*/
|
||||
interface Listener<T> {
|
||||
|
||||
/**
|
||||
* Invoked on new events.
|
||||
*
|
||||
* @param response the response.
|
||||
*/
|
||||
@Deprecated
|
||||
default void onNext(T response) {
|
||||
}
|
||||
|
||||
/**
|
||||
* Invoked on new events.
|
||||
*
|
||||
* @param notice a wrapper of response
|
||||
*/
|
||||
default void onNotice(PulseServerNotice<T> notice) {
|
||||
notice.ack();
|
||||
}
|
||||
|
||||
/**
|
||||
* Invoked on errors.
|
||||
*
|
||||
* @param throwable the error.
|
||||
*/
|
||||
void onError(Throwable throwable);
|
||||
|
||||
/**
|
||||
* Invoked on completion.
|
||||
*/
|
||||
void onCompleted();
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* Interface of notifier that can send notice to server.
|
||||
*
|
||||
* @param <T>
|
||||
*/
|
||||
interface Notifier<T> extends Closeable {
|
||||
|
||||
/**
|
||||
* closes this watcher and all its resources.
|
||||
*/
|
||||
@Override
|
||||
void close();
|
||||
|
||||
/**
|
||||
* Send notice to pd-server.
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
void notifyServer(T t);
|
||||
|
||||
/**
|
||||
* Send an error report to pd-server.
|
||||
*
|
||||
* @param error
|
||||
*/
|
||||
void crash(String error);
|
||||
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,197 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.hugegraph.pd.client;
|
||||
|
||||
import java.util.Objects;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
|
||||
import org.apache.hugegraph.pd.grpc.pulse.HgPdPulseGrpc;
|
||||
import org.apache.hugegraph.pd.grpc.pulse.PartitionHeartbeatRequest;
|
||||
import org.apache.hugegraph.pd.grpc.pulse.PulseAckRequest;
|
||||
import org.apache.hugegraph.pd.grpc.pulse.PulseCreateRequest;
|
||||
import org.apache.hugegraph.pd.grpc.pulse.PulseNoticeRequest;
|
||||
import org.apache.hugegraph.pd.grpc.pulse.PulseRequest;
|
||||
import org.apache.hugegraph.pd.grpc.pulse.PulseResponse;
|
||||
import org.apache.hugegraph.pd.grpc.pulse.PulseType;
|
||||
import org.apache.hugegraph.pd.pulse.PartitionNotice;
|
||||
|
||||
import com.google.common.util.concurrent.ThreadFactoryBuilder;
|
||||
|
||||
import io.grpc.ManagedChannel;
|
||||
import io.grpc.stub.StreamObserver;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
@Slf4j
|
||||
public final class PDPulseImpl implements PDPulse {
|
||||
|
||||
private static final ConcurrentHashMap<String, ManagedChannel> chs = new ConcurrentHashMap<>();
|
||||
private final ExecutorService threadPool;
|
||||
private HgPdPulseGrpc.HgPdPulseStub stub;
|
||||
private String pdServerAddress;
|
||||
|
||||
// TODO: support several servers.
|
||||
public PDPulseImpl(String pdServerAddress) {
|
||||
this.pdServerAddress = pdServerAddress;
|
||||
this.stub = HgPdPulseGrpc.newStub(Channels.getChannel(pdServerAddress));
|
||||
var namedThreadFactory =
|
||||
new ThreadFactoryBuilder().setNameFormat("ack-notice-pool-%d").build();
|
||||
threadPool = Executors.newSingleThreadExecutor(namedThreadFactory);
|
||||
}
|
||||
|
||||
private String getCurrentHost() {
|
||||
return this.pdServerAddress;
|
||||
}
|
||||
|
||||
private boolean checkChannel() {
|
||||
return stub != null && !((ManagedChannel) stub.getChannel()).isShutdown();
|
||||
}
|
||||
|
||||
/* TODO: handle this override problem */
|
||||
@Override
|
||||
public Notifier<PartitionHeartbeatRequest.Builder> connectPartition(Listener<PulseResponse>
|
||||
listener) {
|
||||
return new PartitionHeartbeat(listener);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean resetStub(String host, Notifier notifier) {
|
||||
log.info("reset stub: current, {}, new: {}, channel state:{}", getCurrentHost(), host,
|
||||
checkChannel());
|
||||
if (Objects.equals(host, getCurrentHost()) && checkChannel()) {
|
||||
return false;
|
||||
}
|
||||
|
||||
if (notifier != null) {
|
||||
notifier.close();
|
||||
}
|
||||
|
||||
this.stub = HgPdPulseGrpc.newStub(Channels.getChannel(host));
|
||||
log.info("pd pulse connect to {}", host);
|
||||
this.pdServerAddress = host;
|
||||
return true;
|
||||
}
|
||||
|
||||
/*** PartitionHeartbeat's implement ***/
|
||||
private class PartitionHeartbeat extends
|
||||
AbstractConnector<PartitionHeartbeatRequest.Builder,
|
||||
PulseResponse> {
|
||||
|
||||
private long observerId = -1;
|
||||
|
||||
PartitionHeartbeat(Listener<PulseResponse> listener) {
|
||||
super(listener, PulseType.PULSE_TYPE_PARTITION_HEARTBEAT);
|
||||
}
|
||||
|
||||
private void setObserverId(long observerId) {
|
||||
if (this.observerId == -1) {
|
||||
this.observerId = observerId;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void notifyServer(PartitionHeartbeatRequest.Builder requestBuilder) {
|
||||
this.reqStream.onNext(PulseRequest.newBuilder()
|
||||
.setNoticeRequest(
|
||||
PulseNoticeRequest.newBuilder()
|
||||
.setPartitionHeartbeatRequest(
|
||||
requestBuilder.build()
|
||||
).build()
|
||||
).build()
|
||||
);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onNext(PulseResponse pulseResponse) {
|
||||
this.setObserverId(pulseResponse.getObserverId());
|
||||
long noticeId = pulseResponse.getNoticeId();
|
||||
this.listener.onNext(pulseResponse);
|
||||
this.listener.onNotice(new PartitionNotice(noticeId,
|
||||
e -> super.ackNotice(e, observerId),
|
||||
pulseResponse));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private abstract class AbstractConnector<N, L> implements Notifier<N>,
|
||||
StreamObserver<PulseResponse> {
|
||||
|
||||
Listener<L> listener;
|
||||
StreamObserver<PulseRequest> reqStream;
|
||||
PulseType pulseType;
|
||||
PulseRequest.Builder reqBuilder = PulseRequest.newBuilder();
|
||||
PulseAckRequest.Builder ackBuilder = PulseAckRequest.newBuilder();
|
||||
|
||||
private AbstractConnector(Listener<L> listener, PulseType pulseType) {
|
||||
this.listener = listener;
|
||||
this.pulseType = pulseType;
|
||||
this.init();
|
||||
}
|
||||
|
||||
void init() {
|
||||
PulseCreateRequest.Builder builder = PulseCreateRequest.newBuilder()
|
||||
.setPulseType(this.pulseType);
|
||||
|
||||
this.reqStream = PDPulseImpl.this.stub.pulse(this);
|
||||
this.reqStream.onNext(reqBuilder.clear().setCreateRequest(builder).build());
|
||||
}
|
||||
|
||||
/*** notifier ***/
|
||||
@Override
|
||||
public void close() {
|
||||
this.reqStream.onCompleted();
|
||||
}
|
||||
|
||||
@Override
|
||||
public abstract void notifyServer(N t);
|
||||
|
||||
@Override
|
||||
public void crash(String error) {
|
||||
this.reqStream.onError(new Throwable(error));
|
||||
}
|
||||
|
||||
/*** listener ***/
|
||||
@Override
|
||||
public abstract void onNext(PulseResponse pulseResponse);
|
||||
|
||||
@Override
|
||||
public void onError(Throwable throwable) {
|
||||
this.listener.onError(throwable);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onCompleted() {
|
||||
this.listener.onCompleted();
|
||||
}
|
||||
|
||||
protected void ackNotice(long noticeId, long observerId) {
|
||||
threadPool.execute(() -> {
|
||||
// log.info("send ack: {}, ts: {}", noticeId, System.currentTimeMillis());
|
||||
this.reqStream.onNext(reqBuilder.clear()
|
||||
.setAckRequest(
|
||||
this.ackBuilder.clear()
|
||||
.setNoticeId(noticeId)
|
||||
.setObserverId(observerId)
|
||||
.build()
|
||||
).build()
|
||||
);
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,140 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.hugegraph.pd.client;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.util.function.Consumer;
|
||||
|
||||
import org.apache.hugegraph.pd.grpc.watch.WatchResponse;
|
||||
import org.apache.hugegraph.pd.watch.NodeEvent;
|
||||
import org.apache.hugegraph.pd.watch.PartitionEvent;
|
||||
|
||||
public interface PDWatch {
|
||||
|
||||
/**
|
||||
* Watch the events of all store-nodes registered in the remote PD-Server.
|
||||
*
|
||||
* @param listener
|
||||
* @return
|
||||
*/
|
||||
//PDWatcher watchNode(Listener<NodeEvent> listener);
|
||||
|
||||
/*** inner static methods ***/
|
||||
static <T> Listener<T> listener(Consumer<T> onNext) {
|
||||
return listener(onNext, t -> {
|
||||
}, () -> {
|
||||
});
|
||||
}
|
||||
|
||||
static <T> Listener<T> listener(Consumer<T> onNext, Consumer<Throwable> onError) {
|
||||
return listener(onNext, onError, () -> {
|
||||
});
|
||||
}
|
||||
|
||||
static <T> Listener<T> listener(Consumer<T> onNext, Runnable onCompleted) {
|
||||
return listener(onNext, t -> {
|
||||
}, onCompleted);
|
||||
}
|
||||
|
||||
static <T> Listener<T> listener(Consumer<T> onNext, Consumer<Throwable> onError,
|
||||
Runnable onCompleted) {
|
||||
return new Listener<T>() {
|
||||
@Override
|
||||
public void onNext(T response) {
|
||||
onNext.accept(response);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onError(Throwable throwable) {
|
||||
onError.accept(throwable);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onCompleted() {
|
||||
onCompleted.run();
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Watch the events of the store-nodes assigned to a specified graph.
|
||||
*
|
||||
* @param graph the graph name which you want to watch
|
||||
* @param listener
|
||||
* @return
|
||||
*/
|
||||
//PDWatcher watchNode(String graph, Listener<NodeEvent> listener);
|
||||
|
||||
String getCurrentHost();
|
||||
|
||||
boolean checkChannel();
|
||||
|
||||
/**
|
||||
* @param listener
|
||||
* @return
|
||||
*/
|
||||
Watcher watchPartition(Listener<PartitionEvent> listener);
|
||||
|
||||
Watcher watchNode(Listener<NodeEvent> listener);
|
||||
|
||||
Watcher watchGraph(Listener<WatchResponse> listener);
|
||||
|
||||
Watcher watchShardGroup(Listener<WatchResponse> listener);
|
||||
|
||||
/**
|
||||
* Interface of Watcher.
|
||||
*/
|
||||
interface Listener<T> {
|
||||
|
||||
/**
|
||||
* Invoked on new events.
|
||||
*
|
||||
* @param response the response.
|
||||
*/
|
||||
void onNext(T response);
|
||||
|
||||
/**
|
||||
* Invoked on errors.
|
||||
*
|
||||
* @param throwable the error.
|
||||
*/
|
||||
void onError(Throwable throwable);
|
||||
|
||||
/**
|
||||
* Invoked on completion.
|
||||
*/
|
||||
default void onCompleted() {
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
interface Watcher extends Closeable {
|
||||
|
||||
/**
|
||||
* closes this watcher and all its resources.
|
||||
*/
|
||||
@Override
|
||||
void close();
|
||||
|
||||
/**
|
||||
* Requests the latest revision processed and propagates it to listeners
|
||||
*/
|
||||
// TODO: what's it for?
|
||||
//void requestProgress();
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,204 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.hugegraph.pd.client;
|
||||
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import org.apache.hugegraph.pd.grpc.watch.HgPdWatchGrpc;
|
||||
import org.apache.hugegraph.pd.grpc.watch.WatchCreateRequest;
|
||||
import org.apache.hugegraph.pd.grpc.watch.WatchNodeResponse;
|
||||
import org.apache.hugegraph.pd.grpc.watch.WatchPartitionResponse;
|
||||
import org.apache.hugegraph.pd.grpc.watch.WatchRequest;
|
||||
import org.apache.hugegraph.pd.grpc.watch.WatchResponse;
|
||||
import org.apache.hugegraph.pd.grpc.watch.WatchType;
|
||||
import org.apache.hugegraph.pd.watch.NodeEvent;
|
||||
import org.apache.hugegraph.pd.watch.PartitionEvent;
|
||||
|
||||
import io.grpc.ManagedChannel;
|
||||
import io.grpc.stub.StreamObserver;
|
||||
|
||||
final class PDWatchImpl implements PDWatch {
|
||||
|
||||
private final HgPdWatchGrpc.HgPdWatchStub stub;
|
||||
|
||||
private final String pdServerAddress;
|
||||
|
||||
// TODO: support several servers.
|
||||
PDWatchImpl(String pdServerAddress) {
|
||||
this.pdServerAddress = pdServerAddress;
|
||||
this.stub = HgPdWatchGrpc.newStub(Channels.getChannel(pdServerAddress));
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getCurrentHost() {
|
||||
return this.pdServerAddress;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean checkChannel() {
|
||||
return stub != null && !((ManagedChannel) stub.getChannel()).isShutdown();
|
||||
}
|
||||
|
||||
/**
|
||||
* Get Partition change watcher.
|
||||
*
|
||||
* @param listener
|
||||
* @return
|
||||
*/
|
||||
@Override
|
||||
public Watcher watchPartition(Listener<PartitionEvent> listener) {
|
||||
return new PartitionWatcher(listener);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get Store-Node change watcher.
|
||||
*
|
||||
* @param listener
|
||||
* @return
|
||||
*/
|
||||
@Override
|
||||
public Watcher watchNode(Listener<NodeEvent> listener) {
|
||||
return new NodeWatcher(listener);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Watcher watchGraph(Listener<WatchResponse> listener) {
|
||||
return new GraphWatcher(listener);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Watcher watchShardGroup(Listener<WatchResponse> listener) {
|
||||
return new ShardGroupWatcher(listener);
|
||||
}
|
||||
|
||||
private class GraphWatcher extends AbstractWatcher<WatchResponse> {
|
||||
|
||||
private GraphWatcher(Listener listener) {
|
||||
super(listener,
|
||||
() -> WatchCreateRequest
|
||||
.newBuilder()
|
||||
.setWatchType(WatchType.WATCH_TYPE_GRAPH_CHANGE)
|
||||
.build()
|
||||
);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onNext(WatchResponse watchResponse) {
|
||||
this.listener.onNext(watchResponse);
|
||||
}
|
||||
}
|
||||
|
||||
private class ShardGroupWatcher extends AbstractWatcher<WatchResponse> {
|
||||
|
||||
private ShardGroupWatcher(Listener listener) {
|
||||
super(listener,
|
||||
() -> WatchCreateRequest
|
||||
.newBuilder()
|
||||
.setWatchType(WatchType.WATCH_TYPE_SHARD_GROUP_CHANGE)
|
||||
.build()
|
||||
);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onNext(WatchResponse watchResponse) {
|
||||
this.listener.onNext(watchResponse);
|
||||
}
|
||||
}
|
||||
|
||||
private class PartitionWatcher extends AbstractWatcher<PartitionEvent> {
|
||||
|
||||
private PartitionWatcher(Listener listener) {
|
||||
super(listener,
|
||||
() -> WatchCreateRequest
|
||||
.newBuilder()
|
||||
.setWatchType(WatchType.WATCH_TYPE_PARTITION_CHANGE)
|
||||
.build()
|
||||
);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onNext(WatchResponse watchResponse) {
|
||||
WatchPartitionResponse res = watchResponse.getPartitionResponse();
|
||||
PartitionEvent event = new PartitionEvent(res.getGraph(), res.getPartitionId(),
|
||||
PartitionEvent.ChangeType.grpcTypeOf(
|
||||
res.getChangeType()));
|
||||
this.listener.onNext(event);
|
||||
}
|
||||
}
|
||||
|
||||
private class NodeWatcher extends AbstractWatcher<NodeEvent> {
|
||||
|
||||
private NodeWatcher(Listener listener) {
|
||||
super(listener,
|
||||
() -> WatchCreateRequest
|
||||
.newBuilder()
|
||||
.setWatchType(WatchType.WATCH_TYPE_STORE_NODE_CHANGE)
|
||||
.build()
|
||||
);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onNext(WatchResponse watchResponse) {
|
||||
WatchNodeResponse res = watchResponse.getNodeResponse();
|
||||
NodeEvent event = new NodeEvent(res.getGraph(), res.getNodeId(),
|
||||
NodeEvent.EventType.grpcTypeOf(res.getNodeEventType()));
|
||||
this.listener.onNext(event);
|
||||
}
|
||||
}
|
||||
|
||||
private abstract class AbstractWatcher<T> implements Watcher, StreamObserver<WatchResponse> {
|
||||
|
||||
Listener<T> listener;
|
||||
StreamObserver<WatchRequest> reqStream;
|
||||
Supplier<WatchCreateRequest> requestSupplier;
|
||||
|
||||
private AbstractWatcher(Listener<T> listener,
|
||||
Supplier<WatchCreateRequest> requestSupplier) {
|
||||
this.listener = listener;
|
||||
this.requestSupplier = requestSupplier;
|
||||
this.init();
|
||||
}
|
||||
|
||||
void init() {
|
||||
this.reqStream = PDWatchImpl.this.stub.watch(this);
|
||||
this.reqStream.onNext(WatchRequest.newBuilder().setCreateRequest(
|
||||
this.requestSupplier.get()
|
||||
).build());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
this.reqStream.onCompleted();
|
||||
}
|
||||
|
||||
@Override
|
||||
public abstract void onNext(WatchResponse watchResponse);
|
||||
|
||||
@Override
|
||||
public void onError(Throwable throwable) {
|
||||
|
||||
this.listener.onError(throwable);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onCompleted() {
|
||||
this.listener.onCompleted();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,50 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.hugegraph.pd.pulse;
|
||||
|
||||
import java.util.function.Consumer;
|
||||
|
||||
import org.apache.hugegraph.pd.grpc.pulse.PulseResponse;
|
||||
|
||||
public class PartitionNotice implements PulseServerNotice<PulseResponse> {
|
||||
|
||||
private final long noticeId;
|
||||
private final Consumer<Long> ackConsumer;
|
||||
private final PulseResponse content;
|
||||
|
||||
public PartitionNotice(long noticeId, Consumer<Long> ackConsumer, PulseResponse content) {
|
||||
this.noticeId = noticeId;
|
||||
this.ackConsumer = ackConsumer;
|
||||
this.content = content;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void ack() {
|
||||
this.ackConsumer.accept(this.noticeId);
|
||||
}
|
||||
|
||||
@Override
|
||||
public long getNoticeId() {
|
||||
return this.noticeId;
|
||||
}
|
||||
|
||||
@Override
|
||||
public PulseResponse getContent() {
|
||||
return this.content;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,36 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.hugegraph.pd.pulse;
|
||||
|
||||
public interface PulseServerNotice<T> {
|
||||
|
||||
/**
|
||||
* @throws RuntimeException when failed to send ack-message to pd-server
|
||||
*/
|
||||
void ack();
|
||||
|
||||
long getNoticeId();
|
||||
|
||||
/**
|
||||
* Return a response object of gRPC stream.
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
T getContent();
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,100 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.hugegraph.pd.watch;
|
||||
|
||||
import java.util.Objects;
|
||||
|
||||
import org.apache.hugegraph.pd.grpc.watch.NodeEventType;
|
||||
|
||||
public class NodeEvent {
|
||||
|
||||
private final String graph;
|
||||
private final long nodeId;
|
||||
private final EventType eventType;
|
||||
|
||||
public NodeEvent(String graph, long nodeId, EventType eventType) {
|
||||
this.graph = graph;
|
||||
this.nodeId = nodeId;
|
||||
this.eventType = eventType;
|
||||
}
|
||||
|
||||
public String getGraph() {
|
||||
return graph;
|
||||
}
|
||||
|
||||
public long getNodeId() {
|
||||
return nodeId;
|
||||
}
|
||||
|
||||
public EventType getEventType() {
|
||||
return eventType;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object o) {
|
||||
if (this == o) {
|
||||
return true;
|
||||
}
|
||||
if (o == null || getClass() != o.getClass()) {
|
||||
return false;
|
||||
}
|
||||
NodeEvent nodeEvent = (NodeEvent) o;
|
||||
return nodeId == nodeEvent.nodeId && Objects.equals(graph,
|
||||
nodeEvent.graph) &&
|
||||
eventType == nodeEvent.eventType;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
return Objects.hash(graph, nodeId, eventType);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "NodeEvent{" +
|
||||
"graph='" + graph + '\'' +
|
||||
", nodeId=" + nodeId +
|
||||
", eventType=" + eventType +
|
||||
'}';
|
||||
}
|
||||
|
||||
public enum EventType {
|
||||
UNKNOWN,
|
||||
NODE_ONLINE,
|
||||
NODE_OFFLINE,
|
||||
NODE_RAFT_CHANGE,
|
||||
NODE_PD_LEADER_CHANGE;
|
||||
|
||||
public static EventType grpcTypeOf(NodeEventType grpcType) {
|
||||
switch (grpcType) {
|
||||
case NODE_EVENT_TYPE_NODE_ONLINE:
|
||||
return NODE_ONLINE;
|
||||
case NODE_EVENT_TYPE_NODE_OFFLINE:
|
||||
return NODE_OFFLINE;
|
||||
case NODE_EVENT_TYPE_NODE_RAFT_CHANGE:
|
||||
return NODE_RAFT_CHANGE;
|
||||
case NODE_EVENT_TYPE_PD_LEADER_CHANGE:
|
||||
return NODE_PD_LEADER_CHANGE;
|
||||
default:
|
||||
return UNKNOWN;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,22 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.hugegraph.pd.watch;
|
||||
|
||||
public class PDWatcher {
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,94 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.hugegraph.pd.watch;
|
||||
|
||||
import java.util.Objects;
|
||||
|
||||
import org.apache.hugegraph.pd.grpc.watch.WatchChangeType;
|
||||
|
||||
public class PartitionEvent {
|
||||
|
||||
private final String graph;
|
||||
private final int partitionId;
|
||||
private final ChangeType changeType;
|
||||
|
||||
public PartitionEvent(String graph, int partitionId, ChangeType changeType) {
|
||||
this.graph = graph;
|
||||
this.partitionId = partitionId;
|
||||
this.changeType = changeType;
|
||||
}
|
||||
|
||||
public String getGraph() {
|
||||
return this.graph;
|
||||
}
|
||||
|
||||
public int getPartitionId() {
|
||||
return this.partitionId;
|
||||
}
|
||||
|
||||
public ChangeType getChangeType() {
|
||||
return this.changeType;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object o) {
|
||||
if (this == o) {
|
||||
return true;
|
||||
}
|
||||
if (o == null || getClass() != o.getClass()) {
|
||||
return false;
|
||||
}
|
||||
PartitionEvent that = (PartitionEvent) o;
|
||||
return partitionId == that.partitionId && Objects.equals(graph, that.graph) &&
|
||||
changeType == that.changeType;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
return Objects.hash(graph, partitionId, changeType);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "PartitionEvent{" +
|
||||
"graph='" + graph + '\'' +
|
||||
", partitionId=" + partitionId +
|
||||
", changeType=" + changeType +
|
||||
'}';
|
||||
}
|
||||
|
||||
public enum ChangeType {
|
||||
UNKNOWN,
|
||||
ADD,
|
||||
ALTER,
|
||||
DEL;
|
||||
|
||||
public static ChangeType grpcTypeOf(WatchChangeType grpcType) {
|
||||
switch (grpcType) {
|
||||
case WATCH_CHANGE_TYPE_ADD:
|
||||
return ADD;
|
||||
case WATCH_CHANGE_TYPE_ALTER:
|
||||
return ALTER;
|
||||
case WATCH_CHANGE_TYPE_DEL:
|
||||
return DEL;
|
||||
default:
|
||||
return UNKNOWN;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,30 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.hugegraph.pd.watch;
|
||||
|
||||
enum WatchType {
|
||||
|
||||
PARTITION_CHANGE(10);
|
||||
|
||||
private final int value;
|
||||
|
||||
WatchType(int value) {
|
||||
this.value = value;
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
Reference in New Issue