2 * ============LICENSE_START=======================================================
4 * ================================================================================
5 * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
7 * Modifications Copyright (C) 2019 IBM.
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
13 * http://www.apache.org/licenses/LICENSE-2.0
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 * ============LICENSE_END=========================================================
23 package org.onap.dmaap.dbcapi.service;
25 import org.onap.dmaap.dbcapi.util.RandomInteger;
27 import java.util.ArrayList;
28 import java.util.Iterator;
29 import java.util.List;
32 import javax.ws.rs.core.Response.Status;
34 import org.apache.log4j.Logger;
35 import org.json.simple.JSONArray;
36 import org.json.simple.JSONObject;
37 import org.json.simple.parser.JSONParser;
38 import org.json.simple.parser.ParseException;
39 import org.onap.dmaap.dbcapi.client.DrProvConnection;
40 import org.onap.dmaap.dbcapi.database.DatabaseClass;
41 import org.onap.dmaap.dbcapi.logging.BaseLoggingClass;
42 import org.onap.dmaap.dbcapi.model.ApiError;
43 import org.onap.dmaap.dbcapi.model.DR_Pub;
44 import org.onap.dmaap.dbcapi.model.DR_Sub;
45 import org.onap.dmaap.dbcapi.model.Feed;
46 import org.onap.dmaap.dbcapi.model.MR_Client;
47 import org.onap.dmaap.dbcapi.model.DmaapObject.DmaapObject_Status;
48 import org.onap.dmaap.dbcapi.util.DmaapConfig;
50 public class FeedService extends BaseLoggingClass {
52 private Map<String, Feed> feeds = DatabaseClass.getFeeds();
53 private Map<String, DR_Sub> dr_subs = DatabaseClass.getDr_subs();
54 private DR_PubService pubService = new DR_PubService();
55 private DR_SubService subService = new DR_SubService();
56 private DcaeLocationService dcaeLocations = new DcaeLocationService();
57 private String deleteHandling;
58 private String unit_test;
60 public FeedService() {
61 logger.info( "new FeedService");
62 DmaapConfig p = (DmaapConfig)DmaapConfig.getConfig();
63 deleteHandling = p.getProperty("Feed.deleteHandling", "DeleteOnDR");
64 unit_test = p.getProperty( "UnitTest", "No" );
68 public Map<String, Feed> getFeeds() {
72 private void getSubObjects( Feed f ) {
73 ArrayList<DR_Pub> pubs = pubService.getDr_PubsByFeedId( f.getFeedId() );
75 ArrayList<DR_Sub> subs = subService.getDr_SubsByFeedId( f.getFeedId() );
79 public List<Feed> getAllFeeds(){
80 return getAllFeeds(null, null, null);
83 public List<Feed> getAllFeeds( String name, String ver, String match ) {
84 logger.info( "getAllFeeds: name=" + name + " ver=" + ver + " match=" + match);
85 ArrayList<Feed> fatFeeds = new ArrayList<Feed>();
86 for( Feed f: feeds.values() ) {
89 if ( match != null && "startsWith".equals(match) ) {
90 if ( ! f.getFeedName().startsWith( name ) ) {
91 logger.info( "getAllFeeds: feedName=" + f.getFeedName() + " doesn't start with=" + name);
94 } else if ( match != null && match.equals("contains") ) {
95 if ( ! f.getFeedName().contains( name ) ) {
96 logger.info( "getAllFeeds: feedName=" + f.getFeedName() + " doesn't contain=" + name);
100 if ( ! f.getFeedName().equals( name ) ) {
101 logger.info( "getAllFeeds: feedName=" + f.getFeedName() + " doesn't equal=" + name);
107 if ( keep && ver != null ) {
108 if ( ! f.getFeedVersion().equals(ver)) {
109 logger.info( "getAllFeeds: feedVersion=" + f.getFeedName() + " doesn't match " + ver);
112 logger.info( "getAllFeeds: feedVersion=" + f.getFeedName() + " matches " + ver);
125 private Feed _getFeed( String key, ApiError err, boolean flag ) {
126 Feed f = feeds.get( key );
127 if ( f != null && ( flag || f.getStatus() != DmaapObject_Status.DELETED ) ) {
130 err.setCode(Status.NOT_FOUND.getStatusCode());
131 err.setMessage("feed not found");
132 err.setFields("feedId=" + key );
138 public Feed getFeed( String key, ApiError err ) {
139 return _getFeed( key, err, false );
141 public Feed getFeedPure( String key, ApiError err ) {
142 return _getFeed( key, err, true );
145 public Feed getFeedByName( String name, String ver, ApiError err ) {
146 for( Feed f: feeds.values() ) {
147 if ( f.getFeedName().equals( name ) && f.getFeedVersion().equals(ver) ) {
153 err.setCode(Status.NOT_FOUND.getStatusCode());
154 err.setMessage("feed not found");
155 err.setFields("feedName=" + name + " and ver=" + ver );
160 private boolean savePubs( Feed f ) {
161 return savePubs( f, f );
163 // need to save the Pub objects independently and copy pubId from original request
164 private boolean savePubs( Feed fnew, Feed req ) {
166 DR_PubService pubSvc = new DR_PubService();
167 ArrayList<DR_Pub> reqPubs = req.getPubs();
168 ArrayList<DR_Pub> newPubs = fnew.getPubs();
172 int nSize = newPubs.size();
173 int rSize = reqPubs.size();
174 logger.info( "reqPubs size=" + rSize + " newPubs size=" + nSize );
175 if ( nSize != rSize ) {
176 errorLogger.error( "Resulting set of publishers do not match requested set of publishers " + nSize + " vs " + rSize );
177 fnew.setStatus( DmaapObject_Status.INVALID);
180 // NOTE: when i > 1 newPubs are in reverse order from reqPubs
181 for( int i = 0; i < reqPubs.size(); i++ ) {
182 DR_Pub reqPub = reqPubs.get(i);
183 ApiError err = new ApiError();
184 if ( pubSvc.getDr_Pub( reqPub.getPubId(), err ) == null ) {
185 DR_Pub newPub = newPubs.get(nSize - i - 1);
186 reqPub.setPubId(newPub.getPubId());
187 reqPub.setFeedId(newPub.getFeedId());
188 reqPub.setStatus(DmaapObject_Status.VALID);
189 if ( reqPub.getDcaeLocationName() == null ) {
190 reqPub.setDcaeLocationName("notSpecified");
192 pubSvc.addDr_Pub( reqPub );
197 fnew.setPubs(reqPubs);
198 fnew.setStatus(DmaapObject_Status.VALID);
203 private boolean saveSubs( Feed f ) {
204 return saveSubs( f, f );
206 // need to save the Sub objects independently
207 private boolean saveSubs( Feed fnew, Feed req ) {
208 ArrayList<DR_Sub> subs = req.getSubs();
209 if ( subs == null || subs.size() == 0 ) {
210 logger.info( "No subs specified");
212 DR_SubService subSvc = new DR_SubService( fnew.getSubscribeURL() );
213 ApiError err = new ApiError();
214 for( int i = 0; i < subs.size(); i++ ) {
215 DR_Sub sub = subs.get(i);
216 if ( subSvc.getDr_Sub( sub.getSubId(), err) == null ) {
217 subs.set( i, subSvc.addDr_Sub(sub, err));
218 if ( ! err.is2xx()) {
219 logger.error( "i=" + i + " url=" + sub.getDeliveryURL() + " err=" + err.getCode() );
229 fnew.setStatus(DmaapObject_Status.VALID);
234 public Feed addFeed( Feed req, ApiError err ) {
236 // at least 1 pub is required by DR, so create a default pub if none is specified
237 if ( req.getPubs().size() == 0 ) {
238 logger.info( "No pubs specified - creating tmp pub");
239 ArrayList<DR_Pub> pubs = new ArrayList<DR_Pub>();
240 pubs.add( new DR_Pub( dcaeLocations.getCentralLocation())
242 .setRandomPassword());
247 DrProvConnection prov = new DrProvConnection();
248 prov.makeFeedConnection();
249 String resp = prov.doPostFeed( req, err );
250 if ( unit_test.equals( "Yes" ) ) {
251 // assume resp is null, so need to simulate it
252 resp = simulateResp( req, "POST" );
254 logger.info( "resp=" + resp );
255 if ( resp == null ) {
256 switch( err.getCode() ) {
258 err.setFields( "feedName=" + req.getFeedName() + " + feedVersion=" + req.getFeedVersion() );
262 err.setMessage("API deployment/configuration error - contact support");
263 err.setFields( "PROV_AUTH_ADDRESSES");
264 logger.error( "Prov response: 403. " + err.getMessage() + " regarding " + err.getFields() );
268 err.setMessage( "Unexpected response from DR backend" );
269 err.setFields("response");
276 Feed fnew = new Feed( resp );
277 logger.info( "fnew status is:" + fnew.getStatus() );
278 if ( ! fnew.isStatusValid()) {
280 err.setMessage( "Unexpected response from DR backend" );
281 err.setFields("response");
285 //saveChildren( fnew, req );
286 if ( ! savePubs( fnew, req ) || ! saveSubs( fnew, req ) ) {
287 err.setCode(Status.BAD_REQUEST.getStatusCode());
288 err.setMessage("Unable to save Pub or Sub objects");
291 fnew.setFormatUuid(req.getFormatUuid());
293 feeds.put( fnew.getFeedId(), fnew );
297 public Feed updateFeed( Feed req, ApiError err ) {
299 // at least 1 pub is required by DR, so create a default pub if none is specified
300 if ( req.getPubs().size() == 0 ) {
301 logger.info( "No pubs specified - creating tmp pub");
302 ArrayList<DR_Pub> pubs = new ArrayList<DR_Pub>();
303 pubs.add( new DR_Pub( dcaeLocations.getCentralLocation())
305 .setRandomPassword());
309 DrProvConnection prov = new DrProvConnection();
310 prov.makeFeedConnection( req.getFeedId() );
311 String resp = prov.doPutFeed( req, err );
312 if ( unit_test.equals( "Yes" ) ) {
313 // assume resp is null, so need to simulate it
314 resp = simulateResp( req, "PUT" );
317 logger.info( "resp=" + resp );
318 if ( resp == null ) {
319 switch( err.getCode() ) {
321 err.setFields( "feedName=" + req.getFeedName() + " + feedVersion=" + req.getFeedVersion() );
325 err.setMessage("API deployment/configuration error - contact support");
326 err.setFields( "PROV_AUTH_ADDRESSES");
330 err.setMessage( "Unexpected response from DR backend" );
331 err.setFields("response");
337 Feed fnew = new Feed( resp );
338 logger.info( "fnew status is:" + fnew.getStatus() );
339 if ( ! fnew.isStatusValid()) {
341 err.setMessage( "Unexpected response from DR backend" );
342 err.setFields("response");
346 if ( ! savePubs( fnew, req ) || ! saveSubs( fnew, req ) ) {
347 err.setCode(Status.BAD_REQUEST.getStatusCode());
348 err.setMessage("Unable to save Pub or Sub objects");
351 fnew.setFormatUuid(req.getFormatUuid());
353 feeds.put( fnew.getFeedId(), fnew );
359 // DR does not actually delete a feed, so we provide two behaviors:
360 // 1) clean up the feed by removing all subs and pubs, mark it here as DELETED.
361 // then client can add it back if desired.
362 // 2) Call the DR Delete function. Feed with the same name and version can never be added again
364 public Feed removeFeed( Feed req, ApiError err ) {
365 return removeFeed( req, err, true );
368 public Feed removeFeed( Feed req, ApiError err, boolean hitDR ) {
370 // strip pubs and subs from feed first no matter what
371 ArrayList<DR_Pub> pubs = pubService.getDr_PubsByFeedId( req.getFeedId() );
372 for( DR_Pub pub: pubs ) {
373 pubService.removeDr_Pub(pub.getPubId(), err, hitDR);
374 if ( ! err.is2xx()) {
378 ArrayList<DR_Sub> subs = subService.getDr_SubsByFeedId( req.getFeedId() );
379 for ( DR_Sub sub: subs ) {
380 subService.removeDr_Sub(sub.getSubId(), err, hitDR);
381 if ( ! err.is2xx()) {
387 return feeds.remove(req.getFeedId());
390 if ( deleteHandling.equalsIgnoreCase("DeleteOnDR")) {
391 DrProvConnection prov = new DrProvConnection();
392 prov.makeFeedConnection( req.getFeedId() );
393 String resp = prov.doDeleteFeed( req, err );
394 if ( unit_test.equals( "Yes" ) ) {
395 // assume resp is null, so need to simulate it
396 resp = simulateDelResp( req );
398 logger.info( "resp=" + resp );
399 if ( resp == null ) {
400 switch( err.getCode() ) {
402 err.setFields( "feedName=" + req.getFeedName() + " + feedVersion=" + req.getFeedVersion() );
406 err.setMessage("API deployment/configuration error - contact support");
407 err.setFields( "PROV_AUTH_ADDRESSES");
411 err.setMessage( "Unexpected response from DR backend" );
412 err.setFields("response");
414 return req; // return back the requested feed - implies it wasn't removed
416 return feeds.remove(req.getFeedId());
419 logger.info( "Disable pubs for deleted feed - creating tmp pub");
420 ArrayList<DR_Pub> tmppub = new ArrayList<DR_Pub>();
421 tmppub.add( new DR_Pub( dcaeLocations.getCentralLocation())
423 .setRandomPassword());
426 Feed fnew = updateFeed( req, err );
427 if ( ! err.is2xx()) {
430 fnew.setStatus(DmaapObject_Status.DELETED);
431 feeds.put( fnew.getFeedId(), fnew );
440 * sync will retrieve current config from DR and add it to the DB
441 * when hard = true, then first git rid of current DR provisioning data (from the DB)
443 public void sync( boolean hard, ApiError err ) {
447 ArrayList<Feed> flist = new ArrayList<Feed>(this.getAllFeeds());
448 for ( Iterator<Feed> it = flist.iterator(); it.hasNext(); ) {
451 @SuppressWarnings("unused")
452 Feed old = removeFeed( f, err, false );
459 DrProvConnection prov = new DrProvConnection();
460 prov.makeDumpConnection();
461 String resp = prov.doGetDump( err );
465 logger.debug("sync: resp from DR is: " + resp);
467 JSONParser parser = new JSONParser();
470 jsonObj = (JSONObject) parser.parse( resp );
471 } catch ( ParseException pe ) {
472 logger.error( "Error parsing provisioning data: " + resp );
479 JSONArray feedsArray = (JSONArray) jsonObj.get( "feeds");
480 for( i = 0; i < feedsArray.size(); i++ ) {
481 JSONObject entry = (JSONObject) feedsArray.get(i);
482 Feed fnew = new Feed( entry.toJSONString() );
484 logger.info( "fnew status is:" + fnew.getStatus() );
485 if ( ! fnew.isStatusValid()) {
487 err.setMessage( "Unexpected response from DR backend" );
488 err.setFields("response");
492 if ( ! savePubs( fnew ) ) {
493 err.setCode(Status.BAD_REQUEST.getStatusCode());
494 err.setMessage("Unable to save Pub or Sub objects");
497 fnew.setFormatUuid(fnew.getFormatUuid());
499 feeds.put( fnew.getFeedId(), fnew );
503 JSONArray subArray = (JSONArray) jsonObj.get( "subscriptions");
504 for( i = 0; i < subArray.size(); i++ ) {
505 JSONObject entry = (JSONObject) subArray.get(i);
506 DR_Sub snew = new DR_Sub( entry.toJSONString() );
508 logger.info( "snew status is:" + snew.getStatus() );
509 if ( ! snew.isStatusValid()) {
511 err.setMessage( "Unexpected response from DR backend" );
512 err.setFields("response");
516 dr_subs.put( snew.getSubId(), snew );
524 private String simulateResp( Feed f, String action ){
525 String server = "localhost";
527 if ( action.equals( "POST" ) ) {
528 RandomInteger ran = new RandomInteger(10000);
529 feedid = Integer.toString( ran.next() );
530 } else if ( action.equals( "PUT" ) ) {
531 feedid = f.getFeedId();
535 String ret = String.format(
536 "{\"suspend\":false,\"groupid\":0,\"description\":\"%s\",\"version\":\"1.0\",\"authorization\":",
537 f.getFeedDescription() );
539 String endpoints = "{\"endpoint_addrs\":[],\"classification\":\"unclassified\",\"endpoint_ids\":[";
541 for( DR_Pub pub: f.getPubs()) {
542 endpoints += String.format( "%s{\"password\":\"%s\",\"id\":\"%s\"}",
543 sep, pub.getUserpwd(), pub.getUsername() );
550 ret += String.format(
551 "\"name\":\"%s\",\"business_description\":\"\",\"publisher\":\"sim\",\"links\":{\"subscribe\":\"https://%s/subscribe/%s\",\"log\":\"https://%s/feedlog/%s\",\"publish\":\"https://%s/publish/%s\",\"self\":\"https://%s/feed/%s\"}}",
559 logger.info( "simulateResp ret=" + ret );
562 private String simulateDelResp( Feed f ){
563 String server = "localhost";
564 String feedid = f.getFeedId();
565 String ret = String.format(
566 "{\"suspend\":true,\"groupid\":0,\"description\":\"%s\",\"version\":\"1.0\",\"authorization\":{\"endpoint_addrs\":[],\"classification\":\"unclassified\",\"endpoint_ids\":[{\"password\":\"topSecret123\",\"id\":\"sim\"}]},\"name\":\"%s\",\"business_description\":\"\",\"publisher\":\"sim\",\"links\":{\"subscribe\":\"https://%s/subscribe/%s\",\"log\":\"https://%s/feedlog/%s\",\"publish\":\"https://%s/publish/%s\",\"self\":\"https://%s/feed/%s\"}}",
567 f.getFeedDescription(),