diff options
Diffstat (limited to 'dmaap-bc/src/main/java/org/onap/dmaap/dbcapi/service/FeedService.java')
-rw-r--r-- | dmaap-bc/src/main/java/org/onap/dmaap/dbcapi/service/FeedService.java | 572 |
1 files changed, 572 insertions, 0 deletions
diff --git a/dmaap-bc/src/main/java/org/onap/dmaap/dbcapi/service/FeedService.java b/dmaap-bc/src/main/java/org/onap/dmaap/dbcapi/service/FeedService.java new file mode 100644 index 0000000..19b0267 --- /dev/null +++ b/dmaap-bc/src/main/java/org/onap/dmaap/dbcapi/service/FeedService.java @@ -0,0 +1,572 @@ +/*- + * ============LICENSE_START======================================================= + * org.onap.dmaap + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved. + * + * Modifications Copyright (C) 2019 IBM. + * ================================================================================ + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * ============LICENSE_END========================================================= + */ + +package org.onap.dmaap.dbcapi.service; + +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import javax.ws.rs.core.Response.Status; +import org.json.simple.JSONArray; +import org.json.simple.JSONObject; +import org.json.simple.parser.JSONParser; +import org.json.simple.parser.ParseException; +import org.onap.dmaap.dbcapi.client.DrProvConnection; +import org.onap.dmaap.dbcapi.database.DatabaseClass; +import org.onap.dmaap.dbcapi.logging.BaseLoggingClass; +import org.onap.dmaap.dbcapi.model.ApiError; +import org.onap.dmaap.dbcapi.model.DR_Pub; +import org.onap.dmaap.dbcapi.model.DR_Sub; +import org.onap.dmaap.dbcapi.model.DmaapObject.DmaapObject_Status; +import org.onap.dmaap.dbcapi.model.Feed; +import org.onap.dmaap.dbcapi.util.DmaapConfig; +import org.onap.dmaap.dbcapi.util.RandomInteger; + +public class FeedService extends BaseLoggingClass { + + private Map<String, Feed> feeds = DatabaseClass.getFeeds(); + private Map<String, DR_Sub> dr_subs = DatabaseClass.getDr_subs(); + private DR_PubService pubService = new DR_PubService(); + private DR_SubService subService = new DR_SubService(); + private DcaeLocationService dcaeLocations = new DcaeLocationService(); + private String deleteHandling; + private String unit_test; + + public FeedService() { + logger.info( "new FeedService"); + DmaapConfig p = (DmaapConfig)DmaapConfig.getConfig(); + deleteHandling = p.getProperty("Feed.deleteHandling", "DeleteOnDR"); + unit_test = p.getProperty( "UnitTest", "No" ); + + } + + public Map<String, Feed> getFeeds() { + return feeds; + } + + private void getSubObjects( Feed f ) { + ArrayList<DR_Pub> pubs = pubService.getDr_PubsByFeedId( f.getFeedId() ); + f.setPubs(pubs); + ArrayList<DR_Sub> subs = subService.getDr_SubsByFeedId( f.getFeedId() ); + f.setSubs(subs); + } + + public List<Feed> getAllFeeds(){ + return getAllFeeds(null, null, null); + } + + public List<Feed> getAllFeeds( String name, String ver, String match ) { + logger.info( "getAllFeeds: name=" + name + " ver=" + ver + " match=" + match); + ArrayList<Feed> fatFeeds = new ArrayList<Feed>(); + for( Feed f: feeds.values() ) { + boolean keep = true; + if ( name != null ) { + if ( match != null && "startsWith".equals(match) ) { + if ( ! f.getFeedName().startsWith( name ) ) { + logger.info( "getAllFeeds: feedName=" + f.getFeedName() + " doesn't start with=" + name); + keep = false; + } + } else if ( match != null && match.equals("contains") ) { + if ( ! f.getFeedName().contains( name ) ) { + logger.info( "getAllFeeds: feedName=" + f.getFeedName() + " doesn't contain=" + name); + keep = false; + } + } else { + if ( ! f.getFeedName().equals( name ) ) { + logger.info( "getAllFeeds: feedName=" + f.getFeedName() + " doesn't equal=" + name); + keep = false; + } + } + + } + if ( keep && ver != null ) { + if ( ! f.getFeedVersion().equals(ver)) { + logger.info( "getAllFeeds: feedVersion=" + f.getFeedName() + " doesn't match " + ver); + keep = false; + } else { + logger.info( "getAllFeeds: feedVersion=" + f.getFeedName() + " matches " + ver); + } + } + + if (keep){ + getSubObjects(f); + fatFeeds.add(f); + } + } + return fatFeeds; + } + + + private Feed _getFeed( String key, ApiError err, boolean flag ) { + Feed f = feeds.get( key ); + if ( f != null && ( flag || f.getStatus() != DmaapObject_Status.DELETED ) ) { + getSubObjects( f ); + } else { + err.setCode(Status.NOT_FOUND.getStatusCode()); + err.setMessage("feed not found"); + err.setFields("feedId=" + key ); + return null; + } + err.setCode(200); + return f; + } + public Feed getFeed( String key, ApiError err ) { + return _getFeed( key, err, false ); + } + public Feed getFeedPure( String key, ApiError err ) { + return _getFeed( key, err, true ); + } + + public Feed getFeedByName( String name, String ver, ApiError err ) { + for( Feed f: feeds.values() ) { + if ( f.getFeedName().equals( name ) && f.getFeedVersion().equals(ver) ) { + getSubObjects(f); + return f; + } + + } + err.setCode(Status.NOT_FOUND.getStatusCode()); + err.setMessage("feed not found"); + err.setFields("feedName=" + name + " and ver=" + ver ); + return null; + + } + + private boolean savePubs( Feed f ) { + return savePubs( f, f ); + } + // need to save the Pub objects independently and copy pubId from original request + private boolean savePubs( Feed fnew, Feed req ) { + // save any pubs + DR_PubService pubSvc = new DR_PubService(); + ArrayList<DR_Pub> reqPubs = req.getPubs(); + ArrayList<DR_Pub> newPubs = fnew.getPubs(); + + + + int nSize = newPubs.size(); + int rSize = reqPubs.size(); + logger.info( "reqPubs size=" + rSize + " newPubs size=" + nSize ); + if ( nSize != rSize ) { + errorLogger.error( "Resulting set of publishers do not match requested set of publishers " + nSize + " vs " + rSize ); + fnew.setStatus( DmaapObject_Status.INVALID); + return false; + } + // NOTE: when i > 1 newPubs are in reverse order from reqPubs + for( int i = 0; i < reqPubs.size(); i++ ) { + DR_Pub reqPub = reqPubs.get(i); + ApiError err = new ApiError(); + if ( pubSvc.getDr_Pub( reqPub.getPubId(), err ) == null ) { + DR_Pub newPub = newPubs.get(nSize - i - 1); + reqPub.setPubId(newPub.getPubId()); + reqPub.setFeedId(newPub.getFeedId()); + reqPub.setStatus(DmaapObject_Status.VALID); + if ( reqPub.getDcaeLocationName() == null ) { + reqPub.setDcaeLocationName("notSpecified"); + } + pubSvc.addDr_Pub( reqPub ); + } + + } + + fnew.setPubs(reqPubs); + fnew.setStatus(DmaapObject_Status.VALID); + return true; + + } + + private boolean saveSubs( Feed f ) { + return saveSubs( f, f ); + } + // need to save the Sub objects independently + private boolean saveSubs( Feed fnew, Feed req ) { + ArrayList<DR_Sub> subs = req.getSubs(); + if ( subs == null || subs.size() == 0 ) { + logger.info( "No subs specified"); + } else { + DR_SubService subSvc = new DR_SubService( fnew.getSubscribeURL() ); + ApiError err = new ApiError(); + for( int i = 0; i < subs.size(); i++ ) { + DR_Sub sub = subs.get(i); + if ( subSvc.getDr_Sub( sub.getSubId(), err) == null ) { + subs.set( i, subSvc.addDr_Sub(sub, err)); + if ( ! err.is2xx()) { + logger.error( "i=" + i + " url=" + sub.getDeliveryURL() + " err=" + err.getCode() ); + return false; + } + } + + } + fnew.setSubs(subs); + } + + + fnew.setStatus(DmaapObject_Status.VALID); + return true; + + } + + public Feed addFeed( Feed req, ApiError err ) { + + // at least 1 pub is required by DR, so create a default pub if none is specified + if ( req.getPubs().size() == 0 ) { + logger.info( "No pubs specified - creating tmp pub"); + ArrayList<DR_Pub> pubs = new ArrayList<DR_Pub>(); + pubs.add( new DR_Pub( dcaeLocations.getCentralLocation()) + .setRandomUserName() + .setRandomPassword()); + req.setPubs(pubs); + } + + + DrProvConnection prov = new DrProvConnection(); + prov.makeFeedConnection(); + String resp = prov.doPostFeed( req, err ); + if ( unit_test.equals( "Yes" ) ) { + // assume resp is null, so need to simulate it + resp = simulateResp( req, "POST" ); + } + logger.info( "resp=" + resp ); + if ( resp == null ) { + switch( err.getCode() ) { + case 400: + err.setFields( "feedName=" + req.getFeedName() + " + feedVersion=" + req.getFeedVersion() ); + break; + case 403: + err.setCode(500); + err.setMessage("API deployment/configuration error - contact support"); + err.setFields( "PROV_AUTH_ADDRESSES"); + logger.error( "Prov response: 403. " + err.getMessage() + " regarding " + err.getFields() ); + break; + default: + err.setCode(500); + err.setMessage( "Unexpected response from DR backend" ); + err.setFields("response"); + } + return null; + + } + + + Feed fnew = new Feed( resp ); + logger.info( "fnew status is:" + fnew.getStatus() ); + if ( ! fnew.isStatusValid()) { + err.setCode(500); + err.setMessage( "Unexpected response from DR backend" ); + err.setFields("response"); + return null; + } + + //saveChildren( fnew, req ); + if ( ! savePubs( fnew, req ) || ! saveSubs( fnew, req ) ) { + err.setCode(Status.BAD_REQUEST.getStatusCode()); + err.setMessage("Unable to save Pub or Sub objects"); + return null; + } + fnew.setFormatUuid(req.getFormatUuid()); + fnew.setLastMod(); + feeds.put( fnew.getFeedId(), fnew ); + return fnew; + } + + public Feed updateFeed( Feed req, ApiError err ) { + + // at least 1 pub is required by DR, so create a default pub if none is specified + if ( req.getPubs().size() == 0 ) { + logger.info( "No pubs specified - creating tmp pub"); + ArrayList<DR_Pub> pubs = new ArrayList<DR_Pub>(); + pubs.add( new DR_Pub( dcaeLocations.getCentralLocation()) + .setRandomUserName() + .setRandomPassword()); + req.setPubs(pubs); + } + + DrProvConnection prov = new DrProvConnection(); + prov.makeFeedConnection( req.getFeedId() ); + String resp = prov.doPutFeed( req, err ); + if ( unit_test.equals( "Yes" ) ) { + // assume resp is null, so need to simulate it + resp = simulateResp( req, "PUT" ); + err.setCode(200); + } + logger.info( "resp=" + resp ); + if ( resp == null ) { + switch( err.getCode() ) { + case 400: + err.setFields( "feedName=" + req.getFeedName() + " + feedVersion=" + req.getFeedVersion() ); + break; + case 403: + err.setCode(500); + err.setMessage("API deployment/configuration error - contact support"); + err.setFields( "PROV_AUTH_ADDRESSES"); + break; + default: + err.setCode(500); + err.setMessage( "Unexpected response from DR backend" ); + err.setFields("response"); + } + return null; + } + + + Feed fnew = new Feed( resp ); + logger.info( "fnew status is:" + fnew.getStatus() ); + if ( ! fnew.isStatusValid()) { + err.setCode(500); + err.setMessage( "Unexpected response from DR backend" ); + err.setFields("response"); + return null; + } + + if ( ! savePubs( fnew, req ) || ! saveSubs( fnew, req ) ) { + err.setCode(Status.BAD_REQUEST.getStatusCode()); + err.setMessage("Unable to save Pub or Sub objects"); + return null; + } + fnew.setFormatUuid(req.getFormatUuid()); + fnew.setLastMod(); + feeds.put( fnew.getFeedId(), fnew ); + return fnew; + } + + + // + // DR does not actually delete a feed, so we provide two behaviors: + // 1) clean up the feed by removing all subs and pubs, mark it here as DELETED. + // then client can add it back if desired. + // 2) Call the DR Delete function. Feed with the same name and version can never be added again + // + public Feed removeFeed( Feed req, ApiError err ) { + return removeFeed( req, err, true ); + } + + public Feed removeFeed( Feed req, ApiError err, boolean hitDR ) { + + // strip pubs and subs from feed first no matter what + ArrayList<DR_Pub> pubs = pubService.getDr_PubsByFeedId( req.getFeedId() ); + for( DR_Pub pub: pubs ) { + pubService.removeDr_Pub(pub.getPubId(), err, hitDR); + if ( ! err.is2xx()) { + return req; + } + } + ArrayList<DR_Sub> subs = subService.getDr_SubsByFeedId( req.getFeedId() ); + for ( DR_Sub sub: subs ) { + subService.removeDr_Sub(sub.getSubId(), err, hitDR); + if ( ! err.is2xx()) { + return req; + } + } + + if ( ! hitDR ) { + return feeds.remove(req.getFeedId()); + } + + if ( deleteHandling.equalsIgnoreCase("DeleteOnDR")) { + DrProvConnection prov = new DrProvConnection(); + prov.makeFeedConnection( req.getFeedId() ); + String resp = prov.doDeleteFeed( req, err ); + if ( unit_test.equals( "Yes" ) ) { + // assume resp is null, so need to simulate it + resp = simulateDelResp( req ); + } + logger.info( "resp=" + resp ); + if ( resp == null ) { + switch( err.getCode() ) { + case 400: + err.setFields( "feedName=" + req.getFeedName() + " + feedVersion=" + req.getFeedVersion() ); + break; + case 403: + err.setCode(500); + err.setMessage("API deployment/configuration error - contact support"); + err.setFields( "PROV_AUTH_ADDRESSES"); + break; + default: + err.setCode(500); + err.setMessage( "Unexpected response from DR backend" ); + err.setFields("response"); + } + return req; // return back the requested feed - implies it wasn't removed + } + return feeds.remove(req.getFeedId()); + } else { + + logger.info( "Disable pubs for deleted feed - creating tmp pub"); + ArrayList<DR_Pub> tmppub = new ArrayList<DR_Pub>(); + tmppub.add( new DR_Pub( dcaeLocations.getCentralLocation()) + .setRandomUserName() + .setRandomPassword()); + req.setPubs(tmppub); + req.setSubs(null); + Feed fnew = updateFeed( req, err ); + if ( ! err.is2xx()) { + return req; + } + fnew.setStatus(DmaapObject_Status.DELETED); + feeds.put( fnew.getFeedId(), fnew ); + return null; + } + + + } + + + /* + * sync will retrieve current config from DR and add it to the DB + * when hard = true, then first git rid of current DR provisioning data (from the DB) + */ + public void sync( boolean hard, ApiError err ) { + + if ( hard ) { + + ArrayList<Feed> flist = new ArrayList<Feed>(this.getAllFeeds()); + for ( Iterator<Feed> it = flist.iterator(); it.hasNext(); ) { + Feed f = it.next(); + + @SuppressWarnings("unused") + Feed old = removeFeed( f, err, false ); + if (! err.is2xx()) { + return; + } + } + } + + DrProvConnection prov = new DrProvConnection(); + prov.makeDumpConnection(); + String resp = prov.doGetDump( err ); + if (! err.is2xx()) { + return; + } + logger.debug("sync: resp from DR is: " + resp); + + JSONParser parser = new JSONParser(); + JSONObject jsonObj; + try { + jsonObj = (JSONObject) parser.parse( resp ); + } catch ( ParseException pe ) { + logger.error( "Error parsing provisioning data: " + resp ); + err.setCode(500); + return; + } + + int i; + + JSONArray feedsArray = (JSONArray) jsonObj.get( "feeds"); + for( i = 0; i < feedsArray.size(); i++ ) { + JSONObject entry = (JSONObject) feedsArray.get(i); + Feed fnew = new Feed( entry.toJSONString() ); + + logger.info( "fnew status is:" + fnew.getStatus() ); + if ( ! fnew.isStatusValid()) { + err.setCode(500); + err.setMessage( "Unexpected response from DR backend" ); + err.setFields("response"); + return; + } + + if ( ! savePubs( fnew ) ) { + err.setCode(Status.BAD_REQUEST.getStatusCode()); + err.setMessage("Unable to save Pub or Sub objects"); + return; + } + fnew.setFormatUuid(fnew.getFormatUuid()); + fnew.setLastMod(); + feeds.put( fnew.getFeedId(), fnew ); + + } + + JSONArray subArray = (JSONArray) jsonObj.get( "subscriptions"); + for( i = 0; i < subArray.size(); i++ ) { + JSONObject entry = (JSONObject) subArray.get(i); + DR_Sub snew = new DR_Sub( entry.toJSONString() ); + + logger.info( "snew status is:" + snew.getStatus() ); + if ( ! snew.isStatusValid()) { + err.setCode(500); + err.setMessage( "Unexpected response from DR backend" ); + err.setFields("response"); + return; + } + + dr_subs.put( snew.getSubId(), snew ); + + } + err.setCode(200); + return; + + } + + private String simulateResp( Feed f, String action ){ + String server = "localhost"; + String feedid; + if ( action.equals( "POST" ) ) { + RandomInteger ran = new RandomInteger(10000); + feedid = Integer.toString( ran.next() ); + } else if ( action.equals( "PUT" ) ) { + feedid = f.getFeedId(); + } else { + feedid = "99"; + } + String ret = String.format( +"{\"suspend\":false,\"groupid\":0,\"description\":\"%s\",\"version\":\"1.0\",\"authorization\":", + f.getFeedDescription() ); + + String endpoints = "{\"endpoint_addrs\":[],\"classification\":\"unclassified\",\"endpoint_ids\":["; + String sep = ""; + for( DR_Pub pub: f.getPubs()) { + endpoints += String.format( "%s{\"password\":\"%s\",\"id\":\"%s\"}", + sep, pub.getUserpwd(), pub.getUsername() ); + sep = ","; + + } + endpoints += "]},"; + ret += endpoints; + + ret += String.format( + "\"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\"}}", + + f.getFeedName(), + server, feedid, + server, feedid, + server, feedid, + server, feedid + ); + logger.info( "simulateResp ret=" + ret ); + return ret; + } + private String simulateDelResp( Feed f ){ + String server = "localhost"; + String feedid = f.getFeedId(); + String ret = String.format( +"{\"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\"}}", + f.getFeedDescription(), + f.getFeedName(), + server, feedid, + server, feedid, + server, feedid, + server, feedid + + ); + return ret; + } +} |