Add link eviction to gRPC Link SB.
- Now assumes (remote) LinkProvider to periodically report existing Link.
- Note: This eviction mechanism can be removed, once gRPC Link SB service was
remodelled using streaming RPC.
Change-Id: I98f05f849b876cff9bbdb648e7ac79f900f4bfcb
diff --git a/incubator/rpc-grpc/src/main/java/org/onosproject/incubator/rpc/grpc/LinkProviderServiceServerProxy.java b/incubator/rpc-grpc/src/main/java/org/onosproject/incubator/rpc/grpc/LinkProviderServiceServerProxy.java
index 608f6e3..435c650 100644
--- a/incubator/rpc-grpc/src/main/java/org/onosproject/incubator/rpc/grpc/LinkProviderServiceServerProxy.java
+++ b/incubator/rpc-grpc/src/main/java/org/onosproject/incubator/rpc/grpc/LinkProviderServiceServerProxy.java
@@ -17,8 +17,14 @@
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.cache.RemovalListeners.asynchronous;
import static org.onosproject.net.DeviceId.deviceId;
+import static org.onosproject.net.LinkKey.linkKey;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.lang3.tuple.Pair;
import org.onosproject.grpc.net.Link.ConnectPoint.ElementIdCase;
import org.onosproject.grpc.net.Link.LinkType;
import org.onosproject.grpc.net.link.LinkProviderServiceRpcGrpc.LinkProviderServiceRpc;
@@ -29,6 +35,7 @@
import org.onosproject.net.ConnectPoint;
import org.onosproject.net.DeviceId;
import org.onosproject.net.Link;
+import org.onosproject.net.LinkKey;
import org.onosproject.net.PortNumber;
import org.onosproject.net.SparseAnnotations;
import org.onosproject.net.link.DefaultLinkDescription;
@@ -38,9 +45,13 @@
import org.slf4j.LoggerFactory;
import com.google.api.client.repackaged.com.google.common.annotations.Beta;
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.RemovalNotification;
import io.grpc.stub.StreamObserver;
+// Only single instance will be created and bound to gRPC LinkProviderService
/**
* Server-side implementation of gRPC version of LinkProviderService.
*/
@@ -48,15 +59,39 @@
final class LinkProviderServiceServerProxy
implements LinkProviderServiceRpc {
+ /**
+ * Silence time in seconds, until link gets treated as vanished.
+ */
+ private static final int EVICT_LIMIT = 3 * 3;
+
private final Logger log = LoggerFactory.getLogger(getClass());
private final GrpcRemoteServiceServer server;
// TODO implement aging mechanism to automatically remove
// stale links reported by dead client, etc.
+ /**
+ * Evicting Cache to track last seen time.
+ */
+ private final Cache<Pair<String, LinkKey>, LinkDescription> lastSeen;
LinkProviderServiceServerProxy(GrpcRemoteServiceServer server) {
this.server = checkNotNull(server);
+ ScheduledExecutorService executor = server.getSharedExecutor();
+ lastSeen = CacheBuilder.newBuilder()
+ .expireAfterWrite(EVICT_LIMIT, TimeUnit.SECONDS)
+ .removalListener(asynchronous(this::onRemove, executor))
+ .build();
+
+ executor.scheduleWithFixedDelay(lastSeen::cleanUp,
+ EVICT_LIMIT, EVICT_LIMIT, TimeUnit.SECONDS);
+ }
+
+ private void onRemove(RemovalNotification<Pair<String, LinkKey>, LinkDescription> n) {
+ if (n.wasEvicted()) {
+ getLinkProviderServiceFor(n.getKey().getLeft())
+ .linkVanished(n.getValue());
+ }
}
/**
@@ -92,6 +127,7 @@
LinkDescription linkDescription = translate(request.getLinkDescription());
linkProviderService.linkDetected(linkDescription);
+ lastSeen.put(Pair.of(scheme, linkKey(linkDescription)), linkDescription);
}
@Override
@@ -123,6 +159,7 @@
case LINK_DESCRIPTION:
LinkDescription desc = translate(request.getLinkDescription());
getLinkProviderServiceFor(scheme).linkVanished(desc);
+ lastSeen.invalidate(Pair.of(scheme, linkKey(desc)));
break;
case SUBJECT_NOT_SET:
default: