dc70e52977369e7123567dab831b68e2945b9c02
[policy/drools-pdp.git] /
1 /*-
2  * ============LICENSE_START=======================================================
3  * policy-endpoints
4  * ================================================================================
5  * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
6  * ================================================================================
7  * Licensed under the Apache License, Version 2.0 (the "License");
8  * you may not use this file except in compliance with the License.
9  * You may obtain a copy of the License at
10  * 
11  *      http://www.apache.org/licenses/LICENSE-2.0
12  * 
13  * Unless required by applicable law or agreed to in writing, software
14  * distributed under the License is distributed on an "AS IS" BASIS,
15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16  * See the License for the specific language governing permissions and
17  * limitations under the License.
18  * ============LICENSE_END=========================================================
19  */
20
21 package org.onap.policy.drools.event.comm.bus.internal;
22
23 import java.util.List;
24 import java.util.Map;
25
26 import org.onap.policy.drools.event.comm.Topic;
27 import org.onap.policy.drools.event.comm.bus.DmaapTopicSource;
28 import org.slf4j.Logger;
29 import org.slf4j.LoggerFactory;
30
31 /**
32  * This topic reader implementation specializes in reading messages
33  * over DMAAP topic and notifying its listeners
34  */
35 public class SingleThreadedDmaapTopicSource extends SingleThreadedBusTopicSource
36                                             implements DmaapTopicSource, Runnable {     
37
38         private static Logger logger = LoggerFactory.getLogger(SingleThreadedDmaapTopicSource.class);
39         
40         protected boolean allowSelfSignedCerts;
41         protected final String userName;
42         protected final String password;
43         
44         protected String environment = null;
45         protected String aftEnvironment = null;
46         protected String partner = null;
47         protected String latitude = null;
48         protected String longitude = null;
49         
50         protected Map<String,String> additionalProps = null;
51
52         
53         /**
54          * 
55          * @param servers DMaaP servers
56          * @param topic DMaaP Topic to be monitored
57          * @param apiKey DMaaP API Key (optional)
58          * @param apiSecret DMaaP API Secret (optional)
59          * @param consumerGroup DMaaP Reader Consumer Group
60          * @param consumerInstance DMaaP Reader Instance
61          * @param fetchTimeout DMaaP fetch timeout
62          * @param fetchLimit DMaaP fetch limit
63          * @param environment DME2 Environment
64          * @param aftEnvironment DME2 AFT Environment
65          * @param partner DME2 Partner
66          * @param latitude DME2 Latitude
67          * @param longitude DME2 Longitude
68          * @param additionalProps Additional properties to pass to DME2
69          * @param useHttps does connection use HTTPS?
70          * @param allowSelfSignedCerts are self-signed certificates allow
71          * 
72          * @throws IllegalArgumentException An invalid parameter passed in
73          */
74         public SingleThreadedDmaapTopicSource(List<String> servers, String topic, 
75                                                                                         String apiKey, String apiSecret,
76                                                                                         String userName, String password,
77                                                                                         String consumerGroup, String consumerInstance,
78                                                                                         int fetchTimeout, int fetchLimit,
79                                                                                         String environment, String aftEnvironment, String partner,
80                                                                                         String latitude, String longitude, Map<String,String> additionalProps,
81                                                                                         boolean useHttps, boolean allowSelfSignedCerts)
82                         throws IllegalArgumentException {
83                         
84                 super(servers, topic, apiKey, apiSecret, 
85                           consumerGroup, consumerInstance, 
86                           fetchTimeout, fetchLimit, useHttps,allowSelfSignedCerts);
87                 
88                 this.userName = userName;
89                 this.password = password;
90                 
91                 this.environment = environment;
92                 this.aftEnvironment = aftEnvironment;
93                 this.partner = partner;
94                 
95                 this.latitude = latitude;
96                 this.longitude = longitude;
97                 
98                 this.additionalProps = additionalProps;
99                 try {
100                         this.init();
101                 } catch (Exception e) {
102                         logger.error("ERROR during init of topic {}", this.topic);
103                         throw new IllegalArgumentException(e);
104                 }
105         }
106
107         /**
108          * 
109          * @param servers DMaaP servers
110          * @param topic DMaaP Topic to be monitored
111          * @param apiKey DMaaP API Key (optional)
112          * @param apiSecret DMaaP API Secret (optional)
113          * @param consumerGroup DMaaP Reader Consumer Group
114          * @param consumerInstance DMaaP Reader Instance
115          * @param fetchTimeout DMaaP fetch timeout
116          * @param fetchLimit DMaaP fetch limit
117          * @param useHttps does connection use HTTPS?
118          * @param allowSelfSignedCerts are self-signed certificates allow
119          * @throws IllegalArgumentException An invalid parameter passed in
120          */
121         public SingleThreadedDmaapTopicSource(List<String> servers, String topic, 
122                                                       String apiKey, String apiSecret,
123                                                       String userName, String password,
124                                                       String consumerGroup, String consumerInstance, 
125                                                       int fetchTimeout, int fetchLimit, boolean useHttps, boolean allowSelfSignedCerts)
126                         throws IllegalArgumentException {
127                 
128                 
129                 super(servers, topic, apiKey, apiSecret, 
130                           consumerGroup, consumerInstance, 
131                           fetchTimeout, fetchLimit, useHttps, allowSelfSignedCerts);
132                 
133                 this.userName = userName;
134                 this.password = password;               
135                 
136                 try {
137                         this.init();
138                 } catch (Exception e) {
139                         logger.warn("dmaap-source: cannot create topic {} because of {}", topic, e.getMessage(), e);
140                         throw new IllegalArgumentException(e);
141                 }
142         }
143         
144
145         /**
146          * Initialize the Cambria or MR Client
147          */
148         @Override
149         public void init() throws Exception {
150                 if (this.userName == null || this.userName.isEmpty() || 
151                                 this.password == null || this.password.isEmpty()) {
152                                 this.consumer =
153                                                 new BusConsumer.CambriaConsumerWrapper(this.servers, this.topic, 
154                                                                                            this.apiKey, this.apiSecret,
155                                                                                            this.consumerGroup, this.consumerInstance,
156                                                                                            this.fetchTimeout, this.fetchLimit,
157                                                                                            this.useHttps, this.allowSelfSignedCerts);
158                 } else if ((this.environment == null    || this.environment.isEmpty()) &&
159                                    (this.aftEnvironment == null || this.aftEnvironment.isEmpty()) &&
160                                    (this.latitude == null          || this.latitude.isEmpty()) &&
161                                    (this.longitude == null         || this.longitude.isEmpty()) &&
162                                    (this.partner == null           || this.partner.isEmpty())) {
163                         this.consumer =
164                                         new BusConsumer.DmaapAafConsumerWrapper(this.servers, this.topic, 
165                                                                                     this.apiKey, this.apiSecret,
166                                                                                     this.userName, this.password,
167                                                                                     this.consumerGroup, this.consumerInstance,
168                                                                                     this.fetchTimeout, this.fetchLimit, this.useHttps);
169                 } else {
170                         this.consumer =
171                                         new BusConsumer.DmaapDmeConsumerWrapper(this.servers, this.topic, 
172                                                                                     this.apiKey, this.apiSecret,
173                                                                                     this.userName, this.password,
174                                                                                     this.consumerGroup, this.consumerInstance,
175                                                                                     this.fetchTimeout, this.fetchLimit,
176                                                                                     this.environment, this.aftEnvironment, this.partner,
177                                                                                     this.latitude, this.longitude, this.additionalProps, this.useHttps);
178                 }
179                         
180                 logger.info("{}: INITTED", this);
181         }
182         
183         /**
184          * {@inheritDoc}
185          */
186         @Override
187         public CommInfrastructure getTopicCommInfrastructure() {
188                 return Topic.CommInfrastructure.DMAAP;
189         }
190
191         @Override
192         public String toString() {
193                 StringBuilder builder = new StringBuilder();
194                 builder.append("SingleThreadedDmaapTopicSource [userName=").append(userName).append(", password=")
195                                 .append((password == null || password.isEmpty()) ? "-" : password.length())
196                                 .append(", getTopicCommInfrastructure()=").append(getTopicCommInfrastructure())
197                                 .append(", toString()=").append(super.toString()).append("]");
198                 return builder.toString();
199         }
200
201
202 }