appc-asdc-listener/appc-asdc-listener-bundle/src/main/java/org/openecomp/appc/sdc/listener/AsdcCallback.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * openECOMP : APP-C
   4  * ================================================================================
   5  * Copyright (C) 2017 AT&T Intellectual Property. All rights
   6  *                                              reserved.
   7  * ================================================================================
   8  * Licensed under the Apache License, Version 2.0 (the "License");
   9  * you may not use this file except in compliance with the License.
  10  * You may obtain a copy of the License at
  11  *
  12  *      http://www.apache.org/licenses/LICENSE-2.0
  13  *
  14  * Unless required by applicable law or agreed to in writing, software
  15  * distributed under the License is distributed on an "AS IS" BASIS,
  16  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  17  * See the License for the specific language governing permissions and
  18  * limitations under the License.
  19  * ============LICENSE_END=========================================================
  20  */
  21
  22 package org.openecomp.appc.sdc.listener;
  23
  24 import com.att.eelf.configuration.EELFLogger;
  25 import com.att.eelf.configuration.EELFManager;
  26 import org.openecomp.appc.adapter.message.EventSender;
  27 import org.openecomp.sdc.api.IDistributionClient;
  28 import org.openecomp.sdc.api.consumer.INotificationCallback;
  29 import org.openecomp.sdc.api.notification.IArtifactInfo;
  30 import org.openecomp.sdc.api.notification.INotificationData;
  31 import org.openecomp.sdc.api.notification.IResourceInstance;
  32 import org.apache.commons.lang3.concurrent.BasicThreadFactory;
  33 import org.osgi.framework.BundleContext;
  34 import org.osgi.framework.FrameworkUtil;
  35 import org.osgi.framework.ServiceReference;
  36
  37 import java.net.URI;
  38 import java.util.concurrent.ArrayBlockingQueue;
  39 import java.util.concurrent.ThreadPoolExecutor;
  40 import java.util.concurrent.TimeUnit;
  41 import java.util.concurrent.atomic.AtomicBoolean;
  42
  43 public class AsdcCallback implements INotificationCallback {
  44
  45     private final EELFLogger logger = EELFManager.getInstance().getLogger(AsdcCallback.class);
  46
  47     private URI storeUri;
  48     private IDistributionClient client;
  49
  50     private EventSender eventSender = null;
  51
  52     private ThreadPoolExecutor executor;
  53     private int threadCount = 10;
  54
  55     private AtomicBoolean isRunning = new AtomicBoolean(false);
  56
  57
  58     public AsdcCallback(URI storeUri, IDistributionClient client) {
  59         this.storeUri = storeUri;
  60         this.client = client;
  61
  62         // Create the thread pool
  63         executor = new ThreadPoolExecutor(threadCount, threadCount, 1, TimeUnit.SECONDS,
  64             new ArrayBlockingQueue<Runnable>(threadCount * 2));
  65
  66         // Custom Named thread factory
  67         BasicThreadFactory threadFactory = new BasicThreadFactory.Builder().namingPattern("Appc-Listener-%d").build();
  68         executor.setThreadFactory(threadFactory);
  69
  70         isRunning.set(true);
  71     }
  72
  73     @Override
  74     public void activateCallback(INotificationData data) {
  75         if (null == eventSender) {
  76             try {
  77                 BundleContext bctx = FrameworkUtil.getBundle(EventSender.class).getBundleContext();
  78                 ServiceReference sref = bctx.getServiceReference(EventSender.class);
  79                 eventSender = (EventSender) bctx.getService(sref);
  80             } catch (Throwable t) {
  81                 logger.error("AsdcCallback failed on initializing EventSender", t);
  82             }
  83         }
  84
  85         if (isRunning.get()) {
  86             for (IResourceInstance resource : data.getResources()) {
  87                 for (IArtifactInfo artifact : resource.getArtifacts()) {
  88                     logger.info(Util.toAsdcStoreDocumentInput(data, resource, artifact, "abc"));
  89                     if (executor.getQueue().size() >= threadCount) {
  90                         // log warning about job backlog
  91                     }
  92                     executor.submit(new DownloadAndStoreOp(client, eventSender, data, resource, artifact, storeUri));
  93                 }
  94             }
  95         } else {
  96             // TODO - return a failed result so asdc knows we are shut down
  97         }
  98     }
  99
 100     public void stop() {
 101         stop(10);
 102     }
 103
 104     public void stop(int waitSec) {
 105         isRunning.set(false);
 106         logger.info(String.format("Stopping the ASDC listener and waiting up to %ds for %d pending jobs", waitSec,
 107             executor.getQueue().size()));
 108         boolean cleanShutdown = false;
 109         executor.shutdown();
 110         try {
 111             cleanShutdown = executor.awaitTermination(waitSec, TimeUnit.SECONDS);
 112             executor.shutdownNow(); // In case of timeout
 113         } catch (InterruptedException e) {
 114             e.printStackTrace();
 115         }
 116         logger.info(String.format("Attempting to shutdown cleanly: %s", cleanShutdown ? "SUCCESS" : "FAILURE"));
 117         logger.info("Shutdown complete.");
 118     }
 119
 120 }