Tweaking bind code

Author: grafnu

udmis/src/main/java/com/google/bos/udmi/service/access/ClearBladeIotAccessProvider.java

@@ -246,7 +246,7 @@ protected DeviceManagerInterface getDeviceManager(int monitorSec) {
   }
 
   private CloudModel bindDevicesToGateway(String registryId, String gatewayId,
-      CloudModel cloudModel, Consumer<Integer> progress) {
+      CloudModel cloudModel, Consumer<String> progress) {
     Set<String> deviceIds = ReflectProcessor.isLegacyRequest(cloudModel)
         ? cloudModel.device_ids.keySet() : getDeviceIds(cloudModel.gateway);
     boolean toBind = cloudModel.operation == BIND;
@@ -529,10 +529,11 @@ private Map<String, CloudModel> augmentGatewayModels(String registryId,
     HashMap<String, String> proxyDeviceGateways = new HashMap<>();
     Set<Entry<String, CloudModel>> gateways =
         boundDevices.entrySet().stream().filter(this::isGateway).collect(Collectors.toSet());
-    AtomicInteger count = new AtomicInteger(-gateways.size());
+    AtomicInteger count = new AtomicInteger();
     gateways.forEach(entry -> {
       augmentGatewayModel(registryId, entry, proxyDeviceGateways);
-      progress.accept(format("Processed %d gateways...", count.incrementAndGet()));
+      progress.accept(format("Augmented gateway %s (%d/%d)",
+          entry.getKey(), count.incrementAndGet(), gateways.size()));
     });
     boundDevices.entrySet()
         .forEach(entry -> augmentProxiedModel(entry, proxyDeviceGateways));
@@ -566,7 +567,7 @@ private GatewayModel makeGatewayModel(HashMap<String, CloudModel> boundDevices)
 
   @Override
   public CloudModel modelDevice(String registryId, String deviceId, CloudModel cloudModel,
-      Consumer<Integer> progress) {
+      Consumer<String> progress) {
     String devicePath = getDeviceName(registryId, deviceId);
     Operation operation = cloudModel.operation;
     Resource_type type = ofNullable(cloudModel.resource_type).orElse(Resource_type.DEVICE);
@@ -640,7 +641,7 @@ private StateNotificationConfig stateNotificationConfig() {
   }
 
   private CloudModel unbindAndDelete(String registryId, Device device, CloudModel request,
-      Consumer<Integer> progress) {
+      Consumer<String> progress) {
     try {
       final Set<String> unbindIds = ReflectProcessor.isLegacyRequest(request)
           ? legacyFindGateways(registryId, device)
@@ -685,7 +686,7 @@ private Set<String> legacyFindGateways(String registryId, Device device) {
   }
 
   private CloudModel unbindAndDeleteCore(String registryId, Device device, Set<String> unbindIds,
-      Consumer<Integer> progress) {
+      Consumer<String> progress) {
     String deviceId = requireNonNull(device.toBuilder().getId(), "unspecified device id");
     try {
       ifNotNullThen(unbindIds, ids -> unbindGatewayDevices(registryId, device, ids, progress));
@@ -704,20 +705,25 @@ private CloudModel unbindAndDeleteCore(String registryId, Device device, Set<Str
   }
 
   private void bindDevicesGateways(String registryId, Set<String> gatewayIds,
-      Set<String> deviceIds, boolean toBind, Consumer<Integer> progress) {
+      Set<String> deviceIds, boolean toBind, Consumer<String> progress) {
     String opCode = toBind ? "Binding" : "Unbinding";
     info("%s %s: gateways %s devices %s", opCode, registryId, gatewayIds, deviceIds);
     AtomicInteger opCount = new AtomicInteger(0);
+    ifNotNullThen(progress,
+        p -> p.accept(format("%s %d devices on %d gateways", opCode, deviceIds.size(),
+            gatewayIds.size())));
     gatewayIds.forEach(gatewayId -> {
       deviceIds.forEach(deviceId -> {
         ifTrueThen(opCount.incrementAndGet() % OP_PROGRESS_BATCH_SIZE == 0,
-            () -> ifNotNullThen(progress, p -> p.accept(opCount.get())));
+            () -> ifNotNullThen(progress,
+                p -> p.accept(format("%s %d devices...", opCode, opCount.get()))));
         ifTrueThen(toBind,
             () -> bindDevice(registryId, gatewayId, deviceId),
             () -> unbindDevice(registryId, gatewayId, deviceId));
       });
     });
-    ifTrueThen(opCount.get() > 0, () -> ifNotNullThen(progress, p -> p.accept(opCount.get())));
+    ifTrueThen(opCount.get() > 0, () -> ifNotNullThen(progress, p -> p.accept(
+        format("Completed binding %d devices.", opCount.get()))));
   }
 
   private void unbindDevice(String registryId, String gatewayId, String proxyId) {
@@ -734,7 +740,7 @@ private void unbindDevice(String registryId, String gatewayId, String proxyId) {
   }
 
   private boolean unbindGatewayDevices(String registryId, Device gatewayDevice,
-      Set<String> unbindIds, Consumer<Integer> progress) {
+      Set<String> unbindIds, Consumer<String> progress) {
     try {
       ImmutableSet<String> gatewayIds = ImmutableSet.of(gatewayDevice.toBuilder().getId());
       bindDevicesGateways(registryId, gatewayIds, unbindIds, false, progress);

udmis/src/main/java/com/google/bos/udmi/service/access/DynamicIotAccessProvider.java

@@ -152,7 +152,7 @@ public CloudModel listDevices(String registryId, Consumer<String> progress) {
 
   @Override
   public CloudModel modelDevice(String registryId, String deviceId, CloudModel cloudModel,
-      Consumer<Integer> progress) {
+      Consumer<String> progress) {
     debug("%s iot device %s/%s, %s %s", cloudModel.operation, registryId, deviceId,
         cloudModel.blocked, cloudModel.num_id);
     return getProviderFor(registryId, deviceId).modelDevice(registryId, deviceId, cloudModel,

udmis/src/main/java/com/google/bos/udmi/service/access/ImplicitIotAccessProvider.java

@@ -326,7 +326,7 @@ private Map<String, CloudModel> listBoundDevices(String registryId, String gatew
 
   @Override
   public CloudModel modelDevice(String registryId, String deviceId, CloudModel cloudModel,
-      Consumer<Integer> progress) {
+      Consumer<String> progress) {
     Operation operation = cloudModel.operation;
     Resource_type type = ofNullable(cloudModel.resource_type).orElse(Resource_type.DEVICE);
     checkState(type == DEVICE || type == GATEWAY, "unexpected resource type " + type);

udmis/src/main/java/com/google/bos/udmi/service/access/IotAccessProvider.java

@@ -62,7 +62,7 @@ static IotAccessProvider from(IotAccess iotAccess) {
   CloudModel listDevices(String registryId, Consumer<String> progress);
 
   CloudModel modelDevice(String registryId, String deviceId,
-      CloudModel cloudModel, Consumer<Integer> progress);
+      CloudModel cloudModel, Consumer<String> progress);
 
   CloudModel modelRegistry(String registryId, String deviceId, CloudModel cloudModel);
 

udmis/src/main/java/com/google/bos/udmi/service/access/LocalIotAccessProvider.java

@@ -89,7 +89,7 @@ public CloudModel listDevices(String registryId, Consumer<String> progress) {
 
   @Override
   public CloudModel modelDevice(String registryId, String deviceId, CloudModel cloudModel,
-      Consumer<Integer> progress) {
+      Consumer<String> progress) {
     throw new RuntimeException("Not yet implemented");
   }
 

udmis/src/main/java/com/google/bos/udmi/service/access/PubSubIotAccessProvider.java

@@ -144,7 +144,7 @@ public CloudModel listDevices(String registryId, Consumer<String> progress) {
 
   @Override
   public CloudModel modelDevice(String registryId, String deviceId, CloudModel cloudModel,
-      Consumer<Integer> progress) {
+      Consumer<String> progress) {
     throw new RuntimeException("modelDevice not implemented for PubSub");
   }
 

udmis/src/main/java/com/google/bos/udmi/service/core/ReflectProcessor.java

@@ -260,7 +260,7 @@ private CloudModel queryCloudDevice(Envelope attributes) {
 
   private CloudModel queryCloudRegistry(Envelope attributes) {
     CloudModel cloudModel = iotAccess.listDevices(attributes.deviceRegistryId,
-        progress -> reflectUdmiLog(attributes, format("Fetched %d devices...", progress)));
+        progress -> reflectUdmiLog(attributes, progress));
     cloudModel.operation = READ;
     return cloudModel;
   }
@@ -303,7 +303,7 @@ private CloudModel updateModel(Envelope attributes, CloudModel request) {
       return iotAccess.modelRegistry(deviceRegistryId, deviceId, request);
     } else {
       return iotAccess.modelDevice(deviceRegistryId, deviceId, request, progress ->
-          reflectUdmiLog(attributes, format("Processed %d entries for %s...", progress, deviceId)));
+          reflectUdmiLog(attributes, progress));
     }
   }
 

validator/src/main/java/com/google/daq/mqtt/registrar/Registrar.java

@@ -173,6 +173,7 @@ public class Registrar {
   private SiteModel siteModel;
   private boolean queryOnly;
   private boolean strictWarnings;
+  private boolean doNotUpdate;
 
   /**
    * Main entry point for registrar.
@@ -277,21 +278,26 @@ private void setRunnerThreads(String argValue) {
   @CommandLineOption(short_form = "-d", description = "Delete (known) devices")
   private void setDeleteDevices() {
     checkNotNull(projectId, "delete devices specified with no target project");
-    this.deleteDevices = true;
-    this.updateCloudIoT = true;
+    deleteDevices = true;
+    updateCloudIoT = true;
   }
 
   @CommandLineOption(short_form = "-i", description = "Instantiate extra (unknown) devices")
   private void setInstantiateExtras() {
-    this.instantiateExtras = true;
-    this.updateCloudIoT = true;
+    instantiateExtras = true;
+    updateCloudIoT = true;
   }
 
   @CommandLineOption(short_form = "-x", description = "Expunge (unknown) devices")
   private void setExpungeDevices() {
     checkNotNull(projectId, "expunge devices specified with no target project");
-    this.expungeDevices = true;
-    this.updateCloudIoT = true;
+    expungeDevices = true;
+    updateCloudIoT = true;
+  }
+
+  @CommandLineOption(short_form = "-z", description = "Do not update existing devices")
+  private void setDoNotUpdate() {
+    doNotUpdate = true;
   }
 
   @CommandLineOption(short_form = "-e", arg_name = "suffix", description = "Set registry suffix")
@@ -787,6 +793,11 @@ private boolean processLocalDevice(String localName, AtomicInteger processedDevi
       System.err.println("Skipping active device " + localDevice.getDeviceId());
       return false;
     }
+
+    if (doNotUpdate && cloudModels.containsKey(localName)) {
+      return false;
+    }
+
     Instant start = Instant.now();
     int count = processedDeviceCount.incrementAndGet();
     boolean created = false;