Update license header in appc-event-listener files
[appc.git] / appc-event-listener / appc-event-listener-bundle / src / main / java / org / onap / appc / listener / impl / ControllerImpl.java
1 /*-
2  * ============LICENSE_START=======================================================
3  * ONAP : APPC
4  * ================================================================================
5  * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
6  * ================================================================================
7  * Copyright (C) 2017 Amdocs
8  * =============================================================================
9  * Licensed under the Apache License, Version 2.0 (the "License");
10  * you may not use this file except in compliance with the License.
11  * You may obtain a copy of the License at
12  * 
13  *      http://www.apache.org/licenses/LICENSE-2.0
14  * 
15  * Unless required by applicable law or agreed to in writing, software
16  * distributed under the License is distributed on an "AS IS" BASIS,
17  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
18  * See the License for the specific language governing permissions and
19  * limitations under the License.
20  * 
21  * ============LICENSE_END=========================================================
22  */
23
24 package org.onap.appc.listener.impl;
25
26 import java.util.HashMap;
27 import java.util.Iterator;
28 import java.util.Map;
29 import java.util.Set;
30 import java.util.concurrent.ArrayBlockingQueue;
31 import java.util.concurrent.ThreadPoolExecutor;
32 import java.util.concurrent.TimeUnit;
33
34 import org.apache.commons.lang3.concurrent.BasicThreadFactory;
35 import org.onap.appc.listener.Controller;
36 import org.onap.appc.listener.Listener;
37 import org.onap.appc.listener.ListenerProperties;
38
39 import com.att.eelf.configuration.EELFLogger;
40 import com.att.eelf.configuration.EELFManager;
41
42
43 /**
44  * A common implementation of a controller. This controller should not need to be modified to implement new listeners
45  *
46  */
47 public class ControllerImpl implements Controller {
48
49     private final EELFLogger LOG = EELFManager.getInstance().getLogger(ControllerImpl.class);
50
51     private int LISTENER_COUNT = 1;
52
53     private Map<ListenerProperties, Listener> listeners = null;
54
55     private ThreadPoolExecutor executor;
56
57     /**
58      * Creates a Controller with the set of listener properties which will be used to start listener threads.
59      *
60      * @param properties
61      *            A non null Set of ListenerProperties
62      */
63     public ControllerImpl(Set<ListenerProperties> properties) {
64         listeners = new HashMap<ListenerProperties, Listener>();
65         for (ListenerProperties props : properties) {
66             if (props.getClass() != null) {
67                 listeners.put(props, null);
68             } else {
69                 LOG.error(String.format(
70                     "The ListenerProperties %s has no Listener class associated with it and will not run.", props));
71                 properties.remove(props);
72             }
73         }
74
75         LISTENER_COUNT = properties.size();
76
77         // Only create executor if listeners are configured
78         if (LISTENER_COUNT > 0) {
79         executor = new ThreadPoolExecutor(LISTENER_COUNT, LISTENER_COUNT, 1, TimeUnit.SECONDS,
80             new ArrayBlockingQueue<Runnable>(LISTENER_COUNT));
81
82         // Custom Named thread factory
83         BasicThreadFactory threadFactory = new BasicThreadFactory.Builder().namingPattern("Appc-Listener-%d").build();
84         executor.setThreadFactory(threadFactory);
85         }
86     }
87
88     @Override
89     public void start() {
90         LOG.info("Starting DMaaP Controller.");
91         for (ListenerProperties props : listeners.keySet()) {
92             try {
93                 if (props.isDisabled()) {
94                     LOG.warn(String.format("The listener %s is disabled and will not be run", props.getPrefix()));
95                 } else {
96                     Listener l = props.getListenerClass().getConstructor(ListenerProperties.class).newInstance(props);
97                     l.setListenerId(props.getPrefix());
98                     listeners.put(props, l);
99                     executor.execute(l);
100                 }
101             } catch (Exception e) {
102                 LOG.error(String.format("Exception while starting listener %s.", props), e);
103             }
104         }
105     }
106
107     @Override
108     public void stop(boolean stopNow) {
109         LOG.info("Stopping DMaaP Controller.");
110         Iterator<Listener> itr = listeners.values().iterator();
111         while (itr.hasNext()) {
112             Listener l = itr.next();
113             if (stopNow && l != null) {
114                 l.stopNow();
115             } else if(l!=null){
116                 l.stop();
117             }
118             itr.remove();
119         }
120         // disable new tasks from being submitted
121         if(executor != null) {
122             executor.shutdown();
123             int timeout=300;
124             try {
125                 if (!executor.awaitTermination(timeout, TimeUnit.SECONDS)) {
126                     LOG.error("Not all tasks completed execution after " + timeout + " seconds. " +
127                             "Attempting to stop all actively executing tasks.");
128                     executor.shutdownNow();
129                 }
130                 if (!executor.awaitTermination(timeout, TimeUnit.SECONDS)) {
131                     LOG.error("Could not terminate all tasks after " + (timeout*2) + " seconds.");
132                 }
133             } catch (InterruptedException e) {
134                 executor.shutdownNow();
135                 Thread.currentThread().interrupt();
136             }
137         }
138     }
139
140     @Override
141     public Map<ListenerProperties, Listener> getListeners() {
142         return listeners;
143     }
144 }