Add watch feature

arion_master/src/main/java/com/futurewei/arionmaster/config/ApplicationServerConfig.java

@@ -82,14 +82,12 @@ HazelcastInstance hazelcastInstance(Config config) {
     @Scope("singleton")
     public ClientConfig clientConfig() throws Exception {
         ClientConfig clientConfig = new ClientConfig();
+        clientConfig.getSerializationConfig().addDataSerializableFactory(1, new ArionDataSerializableFactory());
         if (kubernetesconfig) {
             clientConfig.getNetworkConfig().getKubernetesConfig().setEnabled(true)
                     .setProperty("namespace", namespace)
                     .setProperty("service-name", serviceName);
-            ClientUserCodeDeploymentConfig clientUserCodeDeploymentConfig = new ClientUserCodeDeploymentConfig();
-            clientUserCodeDeploymentConfig.addClass("com.futurewei.arionmaster.model.NeighborRule");
-            clientUserCodeDeploymentConfig.setEnabled(true);
-            clientConfig.setUserCodeDeploymentConfig(clientUserCodeDeploymentConfig);
+
         } else {
             clientConfig
                     .getNetworkConfig()

arion_master/src/main/java/com/futurewei/arionmaster/grpc/GrpcServerService.java

@@ -15,19 +15,17 @@ free of charge, to any person obtaining a copy of this software and associated d
 */
 package com.futurewei.arionmaster.grpc;
 
-import com.futurewei.alcor.schema.Common;
-import com.futurewei.alcor.schema.GoalStateProvisionerGrpc;
-import com.futurewei.alcor.schema.Goalstate;
-import com.futurewei.alcor.schema.Goalstateprovisioner;
-import com.futurewei.arionmaster.controller.NeighborStateController;
+import com.futurewei.alcor.schema.*;
 import com.futurewei.arionmaster.service.GoalStatePersistenceService;
 import io.grpc.stub.StreamObserver;
 import net.devh.boot.grpc.server.service.GrpcService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 
-import java.util.logging.Level;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
 
 @GrpcService
 public class GrpcServerService extends GoalStateProvisionerGrpc.GoalStateProvisionerImplBase {
@@ -85,3 +83,7 @@ public void onCompleted() {
         };
     }
 }
+
+
+
+

arion_master/src/main/java/com/futurewei/arionmaster/grpc/Watch.java

@@ -0,0 +1,61 @@
+/*
+MIT License
+Copyright(c) 2022 Futurewei Cloud
+
+    Permission is hereby granted,
+    free of charge, to any person obtaining a copy of this software and associated documentation files(the "Software"), to deal in the Software without restriction,
+    including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and / or sell copies of the Software, and to permit persons
+    to whom the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+    WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+*/
+package com.futurewei.arionmaster.grpc;
+
+import com.futurewei.alcor.schema.Arionmaster;
+import com.futurewei.alcor.schema.WatchGrpc;
+import com.futurewei.arionmaster.service.impl.Watcher;
+import io.grpc.stub.StreamObserver;
+import net.devh.boot.grpc.server.service.GrpcService;
+import org.springframework.beans.factory.annotation.Autowired;
+
+@GrpcService
+public class Watch extends WatchGrpc.WatchImplBase{
+
+
+
+    private String mapName = "com.futurewei.common.model.NeighborRule";
+
+    @Autowired
+    private Watcher watcher;
+
+
+    @Override
+    public StreamObserver<Arionmaster.ArionWingRequest> watch(final StreamObserver<Arionmaster.NeighborRule> responseObserver) {
+        return new StreamObserver<Arionmaster.ArionWingRequest>() {
+            private Runnable cancellation = () -> {};
+            @Override
+            public void onNext(Arionmaster.ArionWingRequest req) {
+                cancellation = watcher.watch(req, mapName, req.getIp(), r -> {
+                    synchronized (responseObserver) {
+                        responseObserver.onNext(r);
+                    }
+                });
+            }
+
+            @Override
+            public void onError(Throwable throwable) {
+                cancellation.run();
+                responseObserver.onError(throwable);
+            }
+
+            @Override
+            public void onCompleted() {
+                responseObserver.onCompleted();
+            }
+        };
+    }
+}

arion_master/src/main/java/com/futurewei/arionmaster/service/impl/Watcher.java

@@ -0,0 +1,123 @@
+/*
+MIT License
+Copyright(c) 2022 Futurewei Cloud
+
+    Permission is hereby granted,
+    free of charge, to any person obtaining a copy of this software and associated documentation files(the "Software"), to deal in the Software without restriction,
+    including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and / or sell copies of the Software, and to permit persons
+    to whom the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+    WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+*/
+package com.futurewei.arionmaster.service.impl;
+
+import com.futurewei.alcor.schema.Arionmaster;
+import com.futurewei.alcor.schema.Common;
+import com.futurewei.arionmaster.grpc.GrpcServerService;
+import com.futurewei.common.model.NeighborRule;
+import com.futurewei.common.service.NeighborRuleService;
+import com.hazelcast.client.HazelcastClient;
+import com.hazelcast.client.config.ClientConfig;
+import com.hazelcast.config.QueryCacheConfig;
+import com.hazelcast.core.EntryEvent;
+import com.hazelcast.core.EntryListener;
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.map.IMap;
+import com.hazelcast.map.MapEvent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import java.util.function.Consumer;
+
+@Service
+public final class Watcher {
+
+    private static final Logger logger = LoggerFactory.getLogger(Watcher.class);
+
+    @Autowired HazelcastInstance hazelcastInstance;
+
+    public Runnable watch(Arionmaster.ArionWingRequest req, String mapName, String cacheName, Consumer<Arionmaster.NeighborRule> resConsumer) {
+        IMap<String, NeighborRule> map = (IMap) hazelcastInstance.getMap(mapName);
+        var cache = map.getQueryCache(cacheName, new NeighborRuleListener(resConsumer), new NeighborRuleService(req.getGroup(), req.getRev()), true);
+        return () -> {
+            cache.destroy();
+        };
+    }
+}
+
+class NeighborRuleListener implements EntryListener {
+
+    private static final Logger logger = LoggerFactory.getLogger(NeighborRuleListener.class);
+
+    private Consumer<Arionmaster.NeighborRule> resConsumer;
+
+
+    public NeighborRuleListener (Consumer<Arionmaster.NeighborRule> resConsumer) {
+        this.resConsumer = resConsumer;
+    }
+
+    @Override
+    public void mapEvicted(MapEvent mapEvent) {
+
+    }
+
+    @Override
+    public void mapCleared(MapEvent mapEvent) {
+
+    }
+
+    @Override
+    public void entryUpdated(EntryEvent entryEvent) {
+        try {
+            resConsumer.accept(buildNeighborRule((NeighborRule) entryEvent.getValue(), Common.OperationType.CREATE));
+        } catch (Exception e) {
+            logger.info(e.getMessage());
+        }
+    }
+
+    @Override
+    public void entryRemoved(EntryEvent entryEvent) {
+        try {
+            resConsumer.accept(buildNeighborRule((NeighborRule) entryEvent.getValue(), Common.OperationType.DELETE));
+        } catch (Exception e) {
+            logger.info(e.getMessage());
+        }
+    }
+
+    @Override
+    public void entryExpired(EntryEvent entryEvent) {
+
+    }
+
+    @Override
+    public void entryEvicted(EntryEvent entryEvent) {
+
+    }
+
+    @Override
+    public void entryAdded(EntryEvent entryEvent) {
+        try {
+            resConsumer.accept(buildNeighborRule((NeighborRule) entryEvent.getValue(), Common.OperationType.CREATE));
+        } catch (Exception e) {
+            logger.info(e.getMessage());
+        }
+
+    }
+
+    public Arionmaster.NeighborRule buildNeighborRule(NeighborRule neighborRule, Common.OperationType operationType) throws Exception {
+        Arionmaster.NeighborRule.Builder neighborRuleBuilder = Arionmaster.NeighborRule.newBuilder();
+        neighborRuleBuilder.setOperationType(operationType);
+        neighborRuleBuilder.setIp(neighborRule.getIp());
+        neighborRuleBuilder.setMac(neighborRule.getMac());
+        neighborRuleBuilder.setHostip(neighborRule.getHostIp());
+        neighborRuleBuilder.setHostmac(neighborRule.getHostMac());
+        neighborRuleBuilder.setArionwingGroup(neighborRule.getArionGroup());
+        neighborRuleBuilder.setVersion(neighborRule.getVersion());
+        return neighborRuleBuilder.build();
+    }
+}

common/src/main/java/com/futurewei/common/model/ArionDataSerializableFactory.java

@@ -17,6 +17,7 @@ free of charge, to any person obtaining a copy of this software and associated d
 
 import com.hazelcast.nio.serialization.IdentifiedDataSerializable;
 
+
 public class ArionDataSerializableFactory implements com.hazelcast.nio.serialization.DataSerializableFactory {
 
         public static final int FACTORY_ID = 1;
@@ -44,3 +45,6 @@ public IdentifiedDataSerializable create(int typeId) {
             }
         }
 }
+
+
+

common/src/main/java/com/futurewei/common/service/NeighborRuleService.java

@@ -0,0 +1,43 @@
+/*
+MIT License
+Copyright(c) 2020 Futurewei Cloud
+
+    Permission is hereby granted,
+    free of charge, to any person obtaining a copy of this software and associated documentation files(the "Software"), to deal in the Software without restriction,
+    including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and / or sell copies of the Software, and to permit persons
+    to whom the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+    WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+*/
+package com.futurewei.common.service;
+
+import com.futurewei.common.model.NeighborRule;
+import com.hazelcast.query.Predicate;
+
+import java.util.Map;
+
+public class NeighborRuleService implements Predicate<String, NeighborRule> {
+
+    private String group;
+    private long rev;
+
+    public NeighborRuleService(String group, long rev) {
+        this.group = group;
+        this.rev = rev;
+    }
+
+    @Override
+    public boolean apply(Map.Entry<String, NeighborRule> entry) {
+        var neighborRule = entry.getValue();
+        if (neighborRule.getArionGroup().equals(group) && neighborRule.getVersion() >= rev) {
+            return true;
+        }
+
+        return false;
+
+    }
+}