GH-91 waiting for the JobConsumers to be ready before publishing job

Author: francoisledroff

aem/aio_aem_events/README.md

@@ -5,11 +5,20 @@ configurations, enabling AEM and Adobe I/O Events integration.
 
 ## Configuration
 
-* The package expects [`Workspace`](./src/main/java/com/adobe/io/workspace/ocd/WorkspaceConfig.java)
-  configuration to be resolved from CloudManager/System environment variables
-    * cf
-      the [Configuring OSGi for Adobe Experience Manager as a Cloud Service](https://experienceleague.adobe.com/docs/experience-manager-cloud-service/content/implementing/deploying/configuring-osgi.html%3Flang%3Den#environment-specific-configuration-values)
-      guide
+This package expects your Adobe Developer Console configured with OSGI configuration
+see [core_aem documentation](../core_aem/README.md)
+
+## Status Checks
+
+This packages holds bundles that come with status servlet:
+* from [/bin/aio/workspace.json](http://localhost:4502/bin/aio/workspace.json) 
+you can `GET` the status of your workspace configuration (see [core_aem documentation](../core_aem/README.md))
+
+http://localhost:4502/bin/aio/provider_config.json
+http://localhost:4502/bin/aio/provider.json
+http://localhost:4502/bin/aio/publish_event.json
+http://localhost:4502/bin/aio/event_metadata.json
+http://localhost:4502/bin/aio/osgi_event_metadata.json
 
 ## Builds
 

aem/core_aem/src/main/java/com/adobe/aio/aem/util/Util.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2017 Adobe. All rights reserved.
+ * This file is licensed to you under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License. You may obtain a copy
+ * of the License at http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under
+ * the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR REPRESENTATIONS
+ * OF ANY KIND, either express or implied. See the License for the specific language
+ * governing permissions and limitations under the License.
+ */
+package com.adobe.aio.aem.util;
+
+import com.adobe.aio.exception.AIOException;
+import java.time.Instant;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.function.Supplier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class Util {
+
+  private Util(){}
+  private static final long TIME_OUT_MILLI = 30000L;
+  private final static Logger log = LoggerFactory.getLogger(Util.class);
+
+  public static void waitFor(Supplier<Boolean> isReady, String supplierDescription) {
+    long start = Instant.now().toEpochMilli();
+    while (!isReady.get()){
+      try {
+        Thread.sleep(ThreadLocalRandom.current().nextInt( 50,500));
+      } catch (InterruptedException e) {
+        throw new AIOException("InterruptedException, while waiting for "+supplierDescription, e);
+      }
+      log.debug("waiting for {} to be ready ...", supplierDescription);
+      if (Instant.now().toEpochMilli()-start> TIME_OUT_MILLI){
+        throw new AIOException("Timeout error, while waiting for "+supplierDescription);
+      }
+    }
+  }
+
+}

aem/events_mgmt_aem/src/main/java/com/adobe/aio/aem/event/management/EventMetadataRegistrationJobConsumer.java

@@ -15,6 +15,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.sling.event.jobs.Job;
 import org.apache.sling.event.jobs.consumer.JobConsumer;
+import org.osgi.service.component.annotations.Activate;
 import org.osgi.service.component.annotations.Component;
 import org.osgi.service.component.annotations.Reference;
 import org.slf4j.Logger;
@@ -40,6 +41,12 @@ public class EventMetadataRegistrationJobConsumer implements JobConsumer {
   EventMetadataStatusSupplier eventMetadataStatusSupplier;
   private String lastErrorMessage;
 
+  @Activate
+  protected void activate() {
+    log.info("activating");
+    eventMetadataStatusSupplier.setJobConsumerReady(true);
+  }
+
   @Override
   public JobResult process(final Job job) {
     if (job.getProperty(AIO_EVENT_CODE_PROPERTY) != null) {

aem/events_mgmt_aem/src/main/java/com/adobe/aio/aem/event/management/EventMetadataStatusSupplier.java

@@ -18,5 +18,7 @@
 @ProviderType
 public interface EventMetadataStatusSupplier extends StatusSupplier {
 
+  Boolean isJobConsumerReady();
+  void setJobConsumerReady(Boolean ready);
   void addStatus(String eventCode, EventMetadataStatus eventMetadataStatus);
 }

aem/events_mgmt_aem/src/main/java/com/adobe/aio/aem/event/management/internal/EventMetadataStatusSupplierImpl.java

@@ -13,12 +13,10 @@
 
 import com.adobe.aio.aem.event.management.EventMetadataStatus;
 import com.adobe.aio.aem.event.management.EventMetadataStatusSupplier;
-import com.adobe.aio.aem.event.management.ocd.ApiManagementConfig;
 import com.adobe.aio.aem.status.Status;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
-import org.osgi.service.component.annotations.Activate;
 import org.osgi.service.component.annotations.Component;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -32,10 +30,16 @@ public class EventMetadataStatusSupplierImpl implements EventMetadataStatusSuppl
 
   private final Logger log = LoggerFactory.getLogger(getClass());
   private final Map<String, EventMetadataStatus> eventMetadataStatusByEventCode = new ConcurrentHashMap<>();
+  private Boolean isJobConsumerReady = false;
 
-  @Activate
-  protected void activate(ApiManagementConfig config) {
-    log.info("activating");
+  @Override
+  public Boolean isJobConsumerReady() {
+    return this.isJobConsumerReady;
+  }
+
+  @Override
+  public void setJobConsumerReady(Boolean ready) {
+    this.isJobConsumerReady = ready;
   }
 
   @Override

aem/events_mgmt_aem/src/main/java/com/adobe/aio/aem/event/management/internal/EventMetadataSupplierImpl.java

@@ -12,12 +12,16 @@
 package com.adobe.aio.aem.event.management.internal;
 
 import com.adobe.aio.aem.event.management.EventMetadataRegistrationJobConsumer;
+import com.adobe.aio.aem.event.management.EventMetadataStatusSupplier;
 import com.adobe.aio.aem.event.management.EventMetadataSupplier;
+import com.adobe.aio.aem.event.management.EventProviderRegistrationService;
 import com.adobe.aio.aem.event.management.ocd.EventMetadataConfig;
+import com.adobe.aio.aem.util.Util;
 import com.adobe.aio.event.management.model.EventMetadata;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.ThreadLocalRandom;
 import org.apache.sling.event.jobs.JobManager;
 import org.osgi.service.component.annotations.Activate;
 import org.osgi.service.component.annotations.Component;
@@ -40,12 +44,20 @@ public class EventMetadataSupplierImpl implements EventMetadataSupplier {
   @Reference
   JobManager jobManager;
 
+  @Reference
+  EventMetadataStatusSupplier eventMetadataStatusSupplier;
+
+  @Reference
+  EventProviderRegistrationService eventProviderRegistrationService;
+
   @Activate
   protected void activate(EventMetadataConfig eventMetadataConfig) {
     Map<String, Object> jobProperties = new HashMap();
     jobProperties.put(EventMetadataRegistrationJobConsumer.AIO_EVENT_CODE_PROPERTY,
         eventMetadataConfig.aio_event_code());
     try {
+      Util.waitFor(eventMetadataStatusSupplier::isJobConsumerReady,
+          "Adobe I/O Events Metadata Registration Job Consumer");
       EventMetadata configuredEventMetadata = EventMetadata.builder()
           .description(eventMetadataConfig.aio_event_description())
           .label(eventMetadataConfig.aio_event_label())
@@ -67,4 +79,5 @@ protected void activate(EventMetadataConfig eventMetadataConfig) {
     }
   }
 
+
 }

aem/events_mgmt_aem/src/main/java/com/adobe/aio/aem/event/management/internal/EventProviderRegistrationServiceImpl.java

@@ -95,7 +95,7 @@ private Provider registerProvider() {
   }
 
   @Override
-  public Provider getRegisteredProvider() {
+  public synchronized Provider getRegisteredProvider() {
     if (!isProviderRegistered()) {
       log.warn("The Adobe I/O Events Provider was not registered yet, trying registering it...");
       return registerProvider();

aem/events_osgi_mapping/src/main/java/com/adobe/aio/aem/event/osgimapping/EventHandlerRegistrationJobConsumer.java

@@ -75,6 +75,7 @@ public class EventHandlerRegistrationJobConsumer implements JobConsumer {
   protected void activate(BundleContext context, Map<String, Object> config) {
     log.info("activating");
     this.bundleContext = context;
+    osgiEventMappingStatusSupplier.setJobConsumerReady(true);
   }
 
   @Override
@@ -87,6 +88,13 @@ public JobResult process(final Job job) {
           workspaceSupplier.getWorkspace().validateAll();
           // we don't want to register sling event handlers if the provider can't be registered
           eventProviderRegistrationService.getRegisteredProvider();
+          /**
+           * Return the proper JobResult based on the work done...
+           *
+           * > OK : Processed successfully
+           * > FAILED: Processed unsuccessfully and reschedule
+           * > CANCEL: Processed unsuccessfully and do NOT reschedule
+           */
 
           OsgiEventMapping osgiEventMapping = new ObjectMapper().readValue
               ((String) job.getProperty(AIO_OSGI_EVENT_MAPPING_PROPERTY), OsgiEventMapping.class);

aem/events_osgi_mapping/src/main/java/com/adobe/aio/aem/event/osgimapping/OsgiEventMappingStatusSupplier.java

@@ -18,5 +18,7 @@
 @ProviderType
 public interface OsgiEventMappingStatusSupplier extends StatusSupplier {
 
+  boolean isJobConsumerReady();
+  void setJobConsumerReady(boolean ready);
   void addStatus(String eventCode, OsgiEventMappingStatus eventMetadataStatus);
 }

aem/events_osgi_mapping/src/main/java/com/adobe/aio/aem/event/osgimapping/internal/OsgiEventMappingStatusSupplierImpl.java

@@ -32,6 +32,17 @@ public class OsgiEventMappingStatusSupplierImpl implements OsgiEventMappingStatu
   private final Logger log = LoggerFactory.getLogger(getClass());
   private final Map<String, OsgiEventMappingStatus> osgiEventMappingStatusByEventCode = new ConcurrentHashMap<>();
 
+  private boolean isJobConsumerReady = false;
+
+  @Override
+  public boolean isJobConsumerReady() {
+    return this.isJobConsumerReady;
+  }
+
+  @Override
+  public void setJobConsumerReady(boolean ready) {
+    this.isJobConsumerReady =ready;
+  }
 
   @Override
   public void addStatus(String eventCode, OsgiEventMappingStatus eventMetadataStatus) {