Fixing message filter

faucetsdn · Oct 8, 2024 · 723dd66 · 723dd66
1 parent 6e9d1df
commit 723dd66
Show file tree

Hide file tree

Showing 5 changed files with 21 additions and 12 deletions.
diff --git a/bin/pull_pubsub b/bin/pull_pubsub
@@ -36,7 +36,7 @@ echo Pulling from subscription $subscription
 pull_limit=100
 
 while true; do
-    date
+    date -u -Is
     gcloud --format=json --project=$project_id pubsub subscriptions pull $subscription --limit $pull_limit --auto-ack > $TMP_FILE || true
 
     for index in $(seq 0 $((pull_limit-1))); do

diff --git a/common/src/main/java/com/google/udmi/util/JsonUtil.java b/common/src/main/java/com/google/udmi/util/JsonUtil.java
@@ -14,6 +14,7 @@
 import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.SerializationFeature;
+import com.google.gson.internal.bind.util.ISO8601Utils;
 import java.io.File;
 import java.nio.file.Files;
 import java.time.Instant;
@@ -205,6 +206,10 @@ public static String isoConvert(Date timestamp) {
     }
   }
 
+  public static String currentIsoMs() {
+    return ISO8601Utils.format(new Date(), true);
+  }
+
   /**
    * Load a file to given type.
    *

diff --git a/udmis/src/main/java/com/google/bos/udmi/service/pod/ContainerBase.java b/udmis/src/main/java/com/google/bos/udmi/service/pod/ContainerBase.java
@@ -308,7 +308,7 @@ public void notice(String message, Object... args) {
   @Override
   public void output(Level level, String message) {
     PrintStream printStream = level.value() >= Level.WARNING.value() ? System.err : System.out;
-    printStream.printf("%s %s %s: %s %s%n", JsonUtil.isoConvert(), getExecutionContext(),
+    printStream.printf("%s %s %s: %s %s%n", JsonUtil.currentIsoMs(), getExecutionContext(),
         level.name().charAt(0), getSimpleName(), message);
     printStream.flush();
   }

diff --git a/validator/src/main/java/com/google/daq/mqtt/sequencer/SequenceBase.java b/validator/src/main/java/com/google/daq/mqtt/sequencer/SequenceBase.java
@@ -396,8 +396,13 @@ private static IotReflectorClient getAlternateClient() {
     altConfiguration.udmi_namespace = exeConfig.udmi_namespace;
     altConfiguration.alt_registry = null;
 
+    return getReflectorClient(altConfiguration);
+  }
+
+  private static IotReflectorClient getReflectorClient(ExecutionConfiguration altConfiguration) {
     try {
-      return new IotReflectorClient(altConfiguration, getRequiredFunctionsVersion());
+      return new IotReflectorClient(altConfiguration, getRequiredFunctionsVersion(),
+          SequenceBase::messageFilter);
     } catch (Exception e) {
       System.err.println(
           "Could not connect to alternate registry, disabling: " + friendlyStackTrace(e));
@@ -408,6 +413,10 @@ private static IotReflectorClient getAlternateClient() {
     }
   }
 
+  private static boolean messageFilter(Envelope envelope) {
+    return true;
+  }
+
   private static int getRequiredFunctionsVersion() {
     return SequenceRunner.processStage(ALPHA) ? SEQUENCER_FUNCTIONS_ALPHA
         : SEQUENCER_FUNCTIONS_VERSION;
@@ -561,7 +570,7 @@ private static MessagePublisher getReflectorClient() {
           format("IoT Provider '%s' not supported, should be one of: %s", exeConfig.iot_provider,
               CSV_JOINER.join(SEQUENCER_PROVIDERS)));
     }
-    return new IotReflectorClient(exeConfig, getRequiredFunctionsVersion());
+    return getReflectorClient(exeConfig);
   }
 
   private static MessagePublisher altReflector() {
@@ -1191,6 +1200,7 @@ protected void updateConfig(String reason) {
 
   private void updateConfig(String reason, boolean force) {
     assertConfigIsNotPending();
+
     // Add a forced sleep to make sure second-quantized timestamps are unique.
     safeSleep(CONFIG_BARRIER_MS);
 
@@ -1774,11 +1784,8 @@ private synchronized void handleUpdateMessage(String subTypeRaw,
         }
         List<DiffEntry> changes = updateDeviceConfig(config);
         debug(format("Updated config %s %s", isoConvert(config.timestamp), txnId));
-        if (updateCount == 1) {
-          info(format("Initial config #%03d", updateCount), stringify(deviceConfig));
-        } else {
-          info(format("Updated config #%03d", updateCount), changedLines(changes));
-        }
+        String changeUpdate = updateCount == 1 ? stringify(deviceConfig) : changedLines(changes);
+        info(format("Updated config #%03d", updateCount), changeUpdate);
       } else if (converted instanceof State convertedState) {
         String timestamp = isoConvert(convertedState.timestamp);
         if (convertedState.timestamp == null) {

diff --git a/validator/src/main/java/com/google/udmi/util/PubSubReflector.java b/validator/src/main/java/com/google/udmi/util/PubSubReflector.java
@@ -13,9 +13,7 @@
 import static com.google.udmi.util.GeneralUtils.ifNotNullThen;
 import static com.google.udmi.util.JsonUtil.isoConvert;
 import static com.google.udmi.util.JsonUtil.stringify;
-import static com.google.udmi.util.JsonUtil.stringifyTerse;
 import static com.google.udmi.util.JsonUtil.toStringMap;
-import static com.google.udmi.util.JsonUtil.writeFile;
 import static java.lang.String.format;
 import static java.time.Instant.ofEpochSecond;
 import static java.util.Objects.requireNonNull;
@@ -43,7 +41,6 @@
 import com.google.pubsub.v1.ProjectTopicName;
 import com.google.pubsub.v1.PubsubMessage;
 import com.google.pubsub.v1.SeekRequest;
-import java.io.File;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.TreeMap;