diff --git a/validator/.idea/runConfigurations/Registrar.xml b/validator/.idea/runConfigurations/Registrar.xml
index ac2276ae1a..b4aab06c4a 100644
--- a/validator/.idea/runConfigurations/Registrar.xml
+++ b/validator/.idea/runConfigurations/Registrar.xml
@@ -1,8 +1,8 @@
 <component name="ProjectRunConfigurationManager">
   <configuration default="false" name="Registrar" type="Application" factoryName="Application">
     <option name="MAIN_CLASS_NAME" value="com.google.daq.mqtt.registrar.Registrar" />
-    <module name="daq-validator.validator" />
-    <option name="PROGRAM_PARAMETERS" value="-s udmi_site_model -p bos-udmi-hub" />
+    <module name="validator" />
+    <option name="PROGRAM_PARAMETERS" value="-u -s udmi_site_model -p bos-daq-testing" />
     <option name="WORKING_DIRECTORY" value="$PROJECT_DIR$/.." />
     <extension name="coverage">
       <pattern>
diff --git a/validator/build.gradle b/validator/build.gradle
index 01c238f9df..189a41a756 100644
--- a/validator/build.gradle
+++ b/validator/build.gradle
@@ -31,7 +31,7 @@ sourceSets {
 
 checkstyle {
     ignoreFailures = false
-    maxWarnings = 61
+    maxWarnings = 48
 }
 checkstyleMain.source = 'src/main/java'
 
diff --git a/validator/src/main/java/com/google/daq/mqtt/registrar/Registrar.java b/validator/src/main/java/com/google/daq/mqtt/registrar/Registrar.java
index 4cb1f743c1..7a088aa362 100644
--- a/validator/src/main/java/com/google/daq/mqtt/registrar/Registrar.java
+++ b/validator/src/main/java/com/google/daq/mqtt/registrar/Registrar.java
@@ -27,6 +27,7 @@
 import java.io.FileInputStream;
 import java.io.FilenameFilter;
 import java.io.InputStream;
+import java.lang.reflect.Field;
 import java.math.BigInteger;
 import java.net.URI;
 import java.time.Duration;
@@ -49,6 +50,7 @@
 import java.util.stream.Collectors;
 import udmi.schema.Config;
 import udmi.schema.Envelope.SubFolder;
+import udmi.schema.Metadata;
 
 public class Registrar {
 
@@ -75,12 +77,14 @@ public class Registrar {
   private static final String SCHEMA_NAME = "UDMI";
   private static final String SWARM_SUBFOLDER = "swarm";
   private static final long PROCESSING_TIMEOUT_MIN = 60;
+  private static final String CONFIG_SUB_TYPE = "config";
+  private static final String MODEL_SUB_TYPE = "model";
   private final Map<String, JsonSchema> schemas = new HashMap<>();
   private final String generation = getGenerationString();
   private CloudIotManager cloudIotManager;
   private File siteDir;
   private File schemaBase;
-  private PubSubPusher configPusher;
+  private PubSubPusher updatePusher;
   private PubSubPusher feedPusher;
   private Map<String, LocalDevice> localDevices;
   private File summaryFile;
@@ -130,9 +134,6 @@ private static boolean processArgs(List<String> argList, Registrar registrar) {
         case "-s":
           registrar.setSitePath(argList.remove(0));
           break;
-        case "-c":
-          registrar.setConfigTopic(argList.remove(0));
-          break;
         case "-f":
           registrar.setFeedTopic(argList.remove(0));
           break;
@@ -173,11 +174,6 @@ private void setFeedTopic(String feedTopic) {
     }
   }
 
-  private void setConfigTopic(String configTopic) {
-    System.err.println("Sending updates to config topic " + configTopic);
-    configPusher = new PubSubPusher(projectId, configTopic);
-  }
-
   private void writeErrors() throws Exception {
     Map<String, Map<String, String>> errorSummary = new TreeMap<>();
     DeviceExceptionManager dem = new DeviceExceptionManager(siteDir);
@@ -234,6 +230,10 @@ private void initializeCloudProject() {
             cloudIotManager.getProjectId(),
             cloudIotManager.getCloudRegion(),
             cloudIotManager.getRegistryId()));
+
+    if (cloudIotManager.getUpdateTopic() != null) {
+      updatePusher = new PubSubPusher(projectId, cloudIotManager.getUpdateTopic());
+    }
   }
 
   private String getGenerationString() {
@@ -312,7 +312,7 @@ private void processLocalDevice(String localName, AtomicInteger processedDeviceC
       localDevice.writeConfigFile();
       if (cloudDevices != null) {
         updateCloudIoT(localName, localDevice);
-        sendConfigMessages(localDevice);
+        sendUpdateMessages(localDevice);
         if (cloudDevices.contains(localName)) {
           sendFeedMessage(localDevice);
         }
@@ -434,31 +434,47 @@ private Device fetchDevice(String localName) {
     }
   }
 
-  private void sendConfigMessages(LocalDevice localDevice) {
-    if (configPusher == null) {
-      return;
+  private void sendUpdateMessages(LocalDevice localDevice) {
+    if (updatePusher != null) {
+      System.err.println("Sending model/config update for " + localDevice.getDeviceId());
+      sendUpdateMessage(localDevice, SubFolder.SYSTEM);
+      sendUpdateMessage(localDevice, SubFolder.POINTSET);
+      sendUpdateMessage(localDevice, SubFolder.GATEWAY);
+      sendUpdateMessage(localDevice, SubFolder.LOCALNET);
     }
-    System.err.println("Sending config messages for " + localDevice.getDeviceId());
+  }
 
-    Config deviceConfig = localDevice.deviceConfigObject();
-    sendConfigMessage(localDevice, SubFolder.SYSTEM, deviceConfig.system);
-    sendConfigMessage(localDevice, SubFolder.POINTSET, deviceConfig.pointset);
-    sendConfigMessage(localDevice, SubFolder.GATEWAY, deviceConfig.gateway);
-    sendConfigMessage(localDevice, SubFolder.LOCALNET, deviceConfig.localnet);
+  private void sendUpdateMessage(LocalDevice localDevice, SubFolder subFolder) {
+    sendUpdateMessage(localDevice, MODEL_SUB_TYPE, subFolder, localDevice.getMetadata());
+    sendUpdateMessage(localDevice, CONFIG_SUB_TYPE, subFolder, localDevice.deviceConfigObject());
   }
 
-  private void sendConfigMessage(LocalDevice localDevice, SubFolder subFolder, Object subConfig) {
+  private void sendUpdateMessage(LocalDevice localDevice, String subType, SubFolder subfolder, Object target) {
+    String fieldName = subfolder.toString().toLowerCase();
+    try {
+      Field declaredField = target.getClass().getDeclaredField(fieldName);
+      sendSubMessage(localDevice, subType, subfolder, declaredField.get(target));
+    } catch (Exception e) {
+      throw new RuntimeException(String.format("Getting field %s from target %s", fieldName, target.getClass().getSimpleName()));
+    }
+  }
+
+  private void sendSubMessage(LocalDevice localDevice, String subType, SubFolder subFolder,
+      Object subConfig) {
     try {
       Map<String, String> attributes = new HashMap<>();
       attributes.put("deviceId", localDevice.getDeviceId());
       attributes.put("deviceNumId", localDevice.getDeviceNumId());
       attributes.put("deviceRegistryId", cloudIotManager.getRegistryId());
       attributes.put("projectId", cloudIotManager.getProjectId());
+      attributes.put("subType", subType);
       attributes.put("subFolder", subFolder.value());
       String messageString = OBJECT_MAPPER.writeValueAsString(subConfig);
-      configPusher.sendMessage(attributes, messageString);
-    } catch (JsonProcessingException e) {
-      throw new RuntimeException("While sending config " + subFolder, e);
+      updatePusher.sendMessage(attributes, messageString);
+    } catch (Exception e) {
+      throw new RuntimeException(
+          String.format("Sending %s/%s messages for %s%n", subType, subFolder,
+              localDevice.getDeviceId()), e);
     }
   }
 
@@ -483,8 +499,8 @@ private void bindGatewayDevices(Map<String, LocalDevice> localDevices, Set<Strin
   }
 
   private void shutdown() {
-    if (configPusher != null) {
-      configPusher.shutdown();
+    if (updatePusher != null) {
+      updatePusher.shutdown();
     }
     if (feedPusher != null) {
       feedPusher.shutdown();
diff --git a/validator/src/main/java/com/google/daq/mqtt/util/CloudDeviceSettings.java b/validator/src/main/java/com/google/daq/mqtt/util/CloudDeviceSettings.java
index 30604d3563..a5b6a77341 100644
--- a/validator/src/main/java/com/google/daq/mqtt/util/CloudDeviceSettings.java
+++ b/validator/src/main/java/com/google/daq/mqtt/util/CloudDeviceSettings.java
@@ -3,6 +3,9 @@
 import com.google.api.services.cloudiot.v1.model.DeviceCredential;
 import java.util.List;
 
+/**
+ * Bucket of settings to use for a cloud device entry.
+ */
 public class CloudDeviceSettings {
 
   public List<DeviceCredential> credentials;
diff --git a/validator/src/main/java/com/google/daq/mqtt/util/CloudIotConfig.java b/validator/src/main/java/com/google/daq/mqtt/util/CloudIotConfig.java
index f813e46ab3..7ef8f33ca1 100644
--- a/validator/src/main/java/com/google/daq/mqtt/util/CloudIotConfig.java
+++ b/validator/src/main/java/com/google/daq/mqtt/util/CloudIotConfig.java
@@ -9,6 +9,7 @@ public class CloudIotConfig {
   public String cloud_region;
   public String reflect_region;
   public String site_name;
+  public String update_topic;
   public String alt_project;
   public String alt_registry;
   public boolean block_unknown;
diff --git a/validator/src/main/java/com/google/daq/mqtt/util/CloudIotManager.java b/validator/src/main/java/com/google/daq/mqtt/util/CloudIotManager.java
index dd1b68b6f1..08e5832e6f 100644
--- a/validator/src/main/java/com/google/daq/mqtt/util/CloudIotManager.java
+++ b/validator/src/main/java/com/google/daq/mqtt/util/CloudIotManager.java
@@ -52,13 +52,19 @@ public class CloudIotManager {
   private final String registryId;
   private final String projectId;
   private final String cloudRegion;
-
+  private final Map<String, Device> deviceMap = new ConcurrentHashMap<>();
+  private final String schemaName;
   private CloudIot cloudIotService;
   private String projectPath;
   private CloudIot.Projects.Locations.Registries cloudIotRegistries;
-  private final Map<String, Device> deviceMap = new ConcurrentHashMap<>();
-  private final String schemaName;
 
+  /**
+   * Create a new CloudIoTManager.
+   *
+   * @param projectId     project id
+   * @param iotConfigFile configuration file
+   * @param schemaName    schema name to use for this device
+   */
   public CloudIotManager(String projectId, File iotConfigFile, String schemaName) {
     this.projectId = projectId;
     this.schemaName = schemaName;
@@ -68,10 +74,17 @@ public CloudIotManager(String projectId, File iotConfigFile, String schemaName)
     initializeCloudIoT();
   }
 
+  /**
+   * Validate the given configuration.
+   *
+   * @param cloudIotConfig configuration to validate
+   * @param projectId      expected project id
+   * @return validated config (for chaining)
+   */
   public static CloudIotConfig validate(CloudIotConfig cloudIotConfig, String projectId) {
     if (projectId.equals(cloudIotConfig.alt_project)) {
-      System.err.printf("Using alt_registry %s for alt_project %s\n",
-          cloudIotConfig.alt_registry, cloudIotConfig.alt_project);
+      System.err.printf("Using alt_registry %s for alt_project %s\n", cloudIotConfig.alt_registry,
+          cloudIotConfig.alt_project);
       cloudIotConfig.alt_project = null;
       cloudIotConfig.registry_id = cloudIotConfig.alt_registry;
       cloudIotConfig.alt_registry = null;
@@ -82,6 +95,13 @@ public static CloudIotConfig validate(CloudIotConfig cloudIotConfig, String proj
     return cloudIotConfig;
   }
 
+  /**
+   * Make an auth credential.
+   *
+   * @param keyFormat key format to use
+   * @param keyData   key data
+   * @return public key device credential
+   */
   public static DeviceCredential makeCredentials(String keyFormat, String keyData) {
     PublicKeyCredential publicKeyCredential = new PublicKeyCredential();
     publicKeyCredential.setFormat(keyFormat);
@@ -104,14 +124,12 @@ private void initializeCloudIoT() {
     projectPath = "projects/" + projectId + "/locations/" + cloudRegion;
     try {
       System.err.println("Initializing with default credentials...");
-      GoogleCredentials credential =
-          GoogleCredentials.getApplicationDefault().createScoped(CloudIotScopes.all());
+      GoogleCredentials credential = GoogleCredentials.getApplicationDefault()
+          .createScoped(CloudIotScopes.all());
       JsonFactory jsonFactory = JacksonFactory.getDefaultInstance();
       HttpRequestInitializer init = new HttpCredentialsAdapter(credential);
-      cloudIotService =
-          new CloudIot.Builder(GoogleNetHttpTransport.newTrustedTransport(), jsonFactory, init)
-              .setApplicationName("com.google.iot.bos")
-              .build();
+      cloudIotService = new CloudIot.Builder(GoogleNetHttpTransport.newTrustedTransport(),
+          jsonFactory, init).setApplicationName("com.google.iot.bos").build();
       cloudIotRegistries = cloudIotService.projects().locations().registries();
       System.err.println("Created service for project " + projectPath);
     } catch (Exception e) {
@@ -119,6 +137,13 @@ private void initializeCloudIoT() {
     }
   }
 
+  /**
+   * Register the given device in a cloud registry.
+   *
+   * @param deviceId device to register
+   * @param settings settings for the device
+   * @return true if this is a new device entry
+   */
   public boolean registerDevice(String deviceId, CloudDeviceSettings settings) {
     try {
       Preconditions.checkNotNull(cloudIotService, "CloudIoT service not initialized");
@@ -141,13 +166,18 @@ private void writeDeviceConfig(String deviceId, String config) {
     try {
       cloudIotRegistries.devices().modifyCloudToDeviceConfig(getDevicePath(deviceId),
           new ModifyCloudToDeviceConfigRequest().setBinaryData(
-              Base64.getEncoder().encodeToString(config.getBytes()))
-      ).execute();
+              Base64.getEncoder().encodeToString(config.getBytes()))).execute();
     } catch (Exception e) {
       throw new RuntimeException("While modifying device config", e);
     }
   }
 
+  /**
+   * Set the blocked state of a given device.
+   *
+   * @param deviceId target device
+   * @param blocked  should this device be blocked?
+   */
   public void blockDevice(String deviceId, boolean blocked) {
     try {
       Device device = new Device();
@@ -160,8 +190,7 @@ public void blockDevice(String deviceId, boolean blocked) {
     }
   }
 
-  private Device makeDevice(String deviceId, CloudDeviceSettings settings,
-      Device oldDevice) {
+  private Device makeDevice(String deviceId, CloudDeviceSettings settings, Device oldDevice) {
     Map<String, String> metadataMap = oldDevice == null ? null : oldDevice.getMetadata();
     if (metadataMap == null) {
       metadataMap = new HashMap<>();
@@ -178,11 +207,8 @@ private Device makeDevice(String deviceId, CloudDeviceSettings settings,
       metadataMap.put(KEY_BYTES_KEY, keyBase64);
       metadataMap.put(KEY_ALGORITHM_KEY, settings.keyAlgorithm);
     }
-    return new Device()
-        .setId(deviceId)
-        .setGatewayConfig(getGatewayConfig(settings))
-        .setCredentials(getCredentials(settings))
-        .setMetadata(metadataMap);
+    return new Device().setId(deviceId).setGatewayConfig(getGatewayConfig(settings))
+        .setCredentials(getCredentials(settings)).setMetadata(metadataMap);
   }
 
   private List<DeviceCredential> getCredentials(CloudDeviceSettings settings) {
@@ -199,40 +225,36 @@ private GatewayConfig getGatewayConfig(CloudDeviceSettings settings) {
 
   private void createDevice(String deviceId, CloudDeviceSettings settings) throws IOException {
     try {
-      cloudIotRegistries.devices().create(getRegistryPath(),
-          makeDevice(deviceId, settings, null)).execute();
+      cloudIotRegistries.devices().create(getRegistryPath(), makeDevice(deviceId, settings, null))
+          .execute();
     } catch (GoogleJsonResponseException e) {
       throw new RuntimeException("Remote error creating device " + deviceId, e);
     }
   }
 
-  private void updateDevice(String deviceId, CloudDeviceSettings settings,
-      Device oldDevice) {
+  private void updateDevice(String deviceId, CloudDeviceSettings settings, Device oldDevice) {
     try {
-      Device device = makeDevice(deviceId, settings, oldDevice)
-          .setId(null)
-          .setNumId(null);
-      cloudIotRegistries
-          .devices()
-          .patch(getDevicePath(deviceId), device).setUpdateMask(DEVICE_UPDATE_MASK)
-          .execute();
+      Device device = makeDevice(deviceId, settings, oldDevice).setId(null).setNumId(null);
+      cloudIotRegistries.devices().patch(getDevicePath(deviceId), device)
+          .setUpdateMask(DEVICE_UPDATE_MASK).execute();
     } catch (Exception e) {
       throw new RuntimeException("Remote error patching device " + deviceId, e);
     }
   }
 
+  /**
+   * Fetch the list of registered devices.
+   *
+   * @return registered device list
+   */
   public Set<String> fetchDeviceList() {
     Preconditions.checkNotNull(cloudIotService, "CloudIoT service not initialized");
     Set<Device> allDevices = new HashSet<>();
     String nextPageToken = null;
     try {
       do {
-        ListDevicesResponse response = cloudIotRegistries
-            .devices()
-            .list(getRegistryPath())
-            .setPageToken(nextPageToken)
-            .setPageSize(LIST_PAGE_SIZE)
-            .execute();
+        ListDevicesResponse response = cloudIotRegistries.devices().list(getRegistryPath())
+            .setPageToken(nextPageToken).setPageSize(LIST_PAGE_SIZE).execute();
         List<Device> devices = response.getDevices();
         allDevices.addAll(devices == null ? ImmutableList.of() : devices);
         System.err.printf("Retrieved %d devices from registry...%n", allDevices.size());
@@ -260,18 +282,47 @@ private Device fetchDeviceRaw(String deviceId) {
     }
   }
 
+  /**
+   * Get target registry.
+   *
+   * @return Target registry
+   */
   public String getRegistryId() {
     return registryId;
   }
 
+  /**
+   * Get target project.
+   *
+   * @return Target GCP project
+   */
   public String getProjectId() {
     return projectId;
   }
 
+  /**
+   * Get target site.
+   *
+   * @return Name for the site (building name)
+   */
   public String getSiteName() {
     return cloudIotConfig.site_name;
   }
 
+  /**
+   * Get update topic.
+   *
+   * @return Topic name to use for sending update messages
+   */
+  public String getUpdateTopic() {
+    return cloudIotConfig.update_topic;
+  }
+
+  /**
+   * Get cloud region.
+   *
+   * @return Cloud region (for the registry)
+   */
   public Object getCloudRegion() {
     return cloudRegion;
   }
@@ -285,6 +336,12 @@ private BindDeviceToGatewayRequest getBindRequest(String deviceId, String gatewa
     return new BindDeviceToGatewayRequest().setDeviceId(deviceId).setGatewayId(gatewayId);
   }
 
+  /**
+   * Get the configuration for a device.
+   *
+   * @param deviceId target device
+   * @return device configuration
+   */
   public String getDeviceConfig(String deviceId) {
     try {
       List<DeviceConfig> deviceConfigs = cloudIotRegistries.devices().configVersions()
@@ -298,6 +355,12 @@ public String getDeviceConfig(String deviceId) {
     }
   }
 
+  /**
+   * Set the device configuration.
+   *
+   * @param deviceId target device
+   * @param data     configuration to set
+   */
   public void setDeviceConfig(String deviceId, String data) {
     try {
       ModifyCloudToDeviceConfigRequest req = new ModifyCloudToDeviceConfigRequest();
@@ -306,8 +369,8 @@ public void setDeviceConfig(String deviceId, String data) {
           .encodeToString(data.getBytes(StandardCharsets.UTF_8.name()));
       req.setBinaryData(encPayload);
 
-      cloudIotRegistries.devices()
-          .modifyCloudToDeviceConfig(getDevicePath(deviceId), req).execute();
+      cloudIotRegistries.devices().modifyCloudToDeviceConfig(getDevicePath(deviceId), req)
+          .execute();
     } catch (Exception e) {
       throw new RuntimeException("While setting device config for " + deviceId);
     }
diff --git a/validator/src/main/java/com/google/daq/mqtt/util/PubSubClient.java b/validator/src/main/java/com/google/daq/mqtt/util/PubSubClient.java
index 188034f5e1..45c9fa8a53 100644
--- a/validator/src/main/java/com/google/daq/mqtt/util/PubSubClient.java
+++ b/validator/src/main/java/com/google/daq/mqtt/util/PubSubClient.java
@@ -13,15 +13,12 @@
 import com.google.cloud.pubsub.v1.Publisher;
 import com.google.cloud.pubsub.v1.Subscriber;
 import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
-import com.google.cloud.pubsub.v1.SubscriptionAdminClient.ListSubscriptionsPagedResponse;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.Timestamp;
-import com.google.pubsub.v1.ProjectName;
 import com.google.pubsub.v1.ProjectSubscriptionName;
 import com.google.pubsub.v1.ProjectTopicName;
 import com.google.pubsub.v1.PubsubMessage;
 import com.google.pubsub.v1.SeekRequest;
-import com.google.pubsub.v1.Subscription;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.TreeMap;
@@ -30,6 +27,9 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BiConsumer;
 
+/**
+ * Message publisher that uses PubSub.
+ */
 public class PubSubClient implements MessagePublisher {
 
   private static final String CONNECT_ERROR_FORMAT = "While connecting to project %s";
@@ -53,6 +53,13 @@ public class PubSubClient implements MessagePublisher {
   private final Subscriber subscriber;
   private final Publisher publisher;
 
+  /**
+   * Create a new PubSub client.
+   *
+   * @param projectId  target project id
+   * @param registryId target registry id
+   * @param name       target subscription name
+   */
   public PubSubClient(String projectId, String registryId, String name) {
     try {
       this.projectId = projectId;
@@ -78,10 +85,20 @@ private SeekRequest getCurrentTimeSeekRequest(String subscription) {
     return SeekRequest.newBuilder().setSubscription(subscription).setTime(timestamp).build();
   }
 
+  /**
+   * Check if the client is active.
+   *
+   * @return true if the client is currently active
+   */
   public boolean isActive() {
     return active.get();
   }
 
+  /**
+   * Process the given message.
+   *
+   * @param handler the handler to use for processing the message
+   */
   @SuppressWarnings("unchecked")
   public void processMessage(BiConsumer<Map<String, Object>, Map<String, String>> handler) {
     try {
diff --git a/validator/src/main/java/com/google/daq/mqtt/util/PubSubPusher.java b/validator/src/main/java/com/google/daq/mqtt/util/PubSubPusher.java
index 2c32266b63..e24d0975a9 100644
--- a/validator/src/main/java/com/google/daq/mqtt/util/PubSubPusher.java
+++ b/validator/src/main/java/com/google/daq/mqtt/util/PubSubPusher.java
@@ -44,6 +44,13 @@ public PubSubPusher(String projectId, String outTopic) {
     }
   }
 
+  /**
+   * Send a message.
+   *
+   * @param attributes message attributes
+   * @param body       message body
+   * @return message ID of the published message
+   */
   public String sendMessage(Map<String, String> attributes, String body) {
     try {
       PubsubMessage message = PubsubMessage.newBuilder()
@@ -57,6 +64,9 @@ public String sendMessage(Map<String, String> attributes, String body) {
     }
   }
 
+  /**
+   * Shutdown this pusher.
+   */
   public void shutdown() {
     try {
       publisher.publishAllOutstanding();
@@ -67,6 +77,11 @@ public void shutdown() {
     }
   }
 
+  /**
+   * Check if the default subscription for this topic is empty.
+   *
+   * @return true if the related subscription is empty
+   */
   public boolean isEmpty() {
     String subscriptionName = ProjectSubscriptionName.format(projectId, outTopic);
     System.err.println("Using PubSub subscription " + subscriptionName);
diff --git a/validator/src/main/java/com/google/daq/mqtt/validator/Validator.java b/validator/src/main/java/com/google/daq/mqtt/validator/Validator.java
index 177b014346..ea66ecaced 100644
--- a/validator/src/main/java/com/google/daq/mqtt/validator/Validator.java
+++ b/validator/src/main/java/com/google/daq/mqtt/validator/Validator.java
@@ -108,6 +108,7 @@ public class Validator {
   private CloudIotManager cloudIotManager;
   private String siteDir;
   private String deviceId;
+  private MessagePublisher client;
 
   /**
    * Validator for streaming message validator.
@@ -149,6 +150,7 @@ public static void main(String[] args) {
         default:
           throw new RuntimeException("Unknown target spec " + targetSpec);
       }
+      validator.messageLoop();
     } catch (ExceptionMap processingException) {
       System.exit(2);
     } catch (Exception e) {
@@ -260,19 +262,21 @@ private BiConsumer<Map<String, Object>, Map<String, String>> messageValidator()
 
   private void validatePubSub(String instName) {
     String registryId = cloudIotConfig.registry_id;
-    PubSubClient client = new PubSubClient(projectId, registryId, instName);
-    messageLoop(client);
+    client = new PubSubClient(projectId, registryId, instName);
   }
 
   private void validateReflector(String instName) {
     deviceId = NO_SITE.equals(instName) ? null : instName;
     String keyFile = new File(siteDir, GCP_REFLECT_KEY_PKCS8).getAbsolutePath();
     System.err.println("Loading reflector key file from " + keyFile);
-    IotCoreClient client = new IotCoreClient(projectId, cloudIotConfig, keyFile);
-    messageLoop(client);
+    client = new IotCoreClient(projectId, cloudIotConfig, keyFile);
   }
 
-  private void messageLoop(MessagePublisher client) {
+  private void messageLoop() {
+    if (client == null) {
+      System.err.println("No message publisher defined.");
+      return;
+    }
     System.err.println(
         "Entering message loop on " + client.getSubscriptionId() + " with device " + deviceId);
     BiConsumer<Map<String, Object>, Map<String, String>> validator = messageValidator();
@@ -281,7 +285,7 @@ private void messageLoop(MessagePublisher client) {
       try {
         if (!initialized) {
           initialized = true;
-          sendInitializationQuery(client);
+          sendInitializationQuery();
         }
         client.processMessage(validator);
       } catch (Exception e) {
@@ -291,7 +295,7 @@ private void messageLoop(MessagePublisher client) {
     System.err.println("Message loop complete");
   }
 
-  private void sendInitializationQuery(MessagePublisher client) {
+  private void sendInitializationQuery() {
     if (deviceId != null) {
       System.err.println("Sending initialization query messages for device " + deviceId);
       client.publish(deviceId, STATE_QUERY_TOPIC, EMPTY_MESSAGE);