sdnr/wt/devicemanager-onf/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/devicemanager/onf/notifications/NotificationWorker.java

   1 /*
   2  * ============LICENSE_START========================================================================
   3  * ONAP : ccsdk feature sdnr wt
   4  * =================================================================================================
   5  * Copyright (C) 2019 highstreet technologies GmbH Intellectual Property. All rights reserved.
   6  * =================================================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
   8  * in compliance with the License. You may obtain a copy of the License at
   9  *
  10  * http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software distributed under the License
  13  * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
  14  * or implied. See the License for the specific language governing permissions and limitations under
  15  * the License.
  16  * ============LICENSE_END==========================================================================
  17  */
  18 package org.onap.ccsdk.features.sdnr.wt.devicemanager.onf.notifications;
  19
  20 import java.util.concurrent.BlockingQueue;
  21 import java.util.concurrent.ExecutorService;
  22 import java.util.concurrent.Executors;
  23 import java.util.concurrent.LinkedBlockingQueue;
  24
  25 /**
  26  * Provide a thread that is receiving and process notifications.
  27  * @param <T> represents the object that is provided with a notification and
  28  * forwarded to the NotificationActor<T>.
  29  */
  30
  31 public class NotificationWorker<T> implements AutoCloseable {
  32
  33     private final BlockingQueue<T> workQueue;
  34     private final ExecutorService service;
  35     private final NotificationActor<T> actor;
  36
  37     public NotificationWorker(int numWorkers, int workQueueSize, NotificationActor<T> actorObject) {
  38         workQueue = new LinkedBlockingQueue<>(workQueueSize);
  39         service = Executors.newFixedThreadPool(numWorkers);
  40         actor = actorObject;
  41
  42         for (int i=0; i < numWorkers; i++) {
  43             service.submit(new Worker<>(workQueue, actor));
  44         }
  45     }
  46
  47     public void put(T item) {
  48         try {
  49             workQueue.put(item);
  50         } catch (InterruptedException ex) {
  51             Thread.currentThread().interrupt();
  52         }
  53     }
  54
  55     @Override
  56     public void close() throws Exception {
  57         // Auto-generated method stub
  58     }
  59
  60     private static class Worker<T> implements Runnable {
  61         private final BlockingQueue<T> workQueue;
  62         private final NotificationActor<T> actor;
  63
  64
  65         public Worker(BlockingQueue<T> workQueue, NotificationActor<T> actor) {
  66             this.workQueue = workQueue;
  67             this.actor = actor;
  68         }
  69
  70         @Override
  71         public void run() {
  72             while (!Thread.currentThread().isInterrupted()) {
  73                 try {
  74                     T item = workQueue.take();
  75                     if (item != null) {
  76                         // Process item
  77                         actor.notificationActor(item);
  78                     }
  79                 } catch (InterruptedException ex) {
  80                     Thread.currentThread().interrupt();
  81                     break;
  82                 }
  83             }
  84         }
  85     }
  86
  87
  88
  89 }