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
10 * http://www.apache.org/licenses/LICENSE-2.0
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
16 * ============LICENSE_END==========================================================================
18 package org.onap.ccsdk.features.sdnr.wt.devicemanager.onf.notifications;
20 import java.util.concurrent.BlockingQueue;
21 import java.util.concurrent.ExecutorService;
22 import java.util.concurrent.Executors;
23 import java.util.concurrent.LinkedBlockingQueue;
26 * Provide a thread that is receiving and process notifications.
28 * @param <T> represents the object that is provided with a notification and forwarded to the NotificationActor<T>.
31 public class NotificationWorker<T> implements AutoCloseable {
33 private final BlockingQueue<T> workQueue;
34 private final ExecutorService service;
35 private final NotificationActor<T> actor;
37 public NotificationWorker(int numWorkers, int workQueueSize, NotificationActor<T> actorObject) {
38 workQueue = new LinkedBlockingQueue<>(workQueueSize);
39 service = Executors.newFixedThreadPool(numWorkers);
42 for (int i = 0; i < numWorkers; i++) {
43 service.submit(new Worker<>(workQueue, actor));
47 public void put(T item) {
50 } catch (InterruptedException ex) {
51 Thread.currentThread().interrupt();
56 public void close() throws Exception {
57 // Auto-generated method stub
60 private static class Worker<T> implements Runnable {
61 private final BlockingQueue<T> workQueue;
62 private final NotificationActor<T> actor;
65 public Worker(BlockingQueue<T> workQueue, NotificationActor<T> actor) {
66 this.workQueue = workQueue;
72 while (!Thread.currentThread().isInterrupted()) {
74 T item = workQueue.take();
77 actor.notificationActor(item);
79 } catch (InterruptedException ex) {
80 Thread.currentThread().interrupt();