private String feedContentType;
private String subContentType;
private String unit_test;
+ private String provURI;
private HttpsURLConnection uc;
behalfHeader = p.getProperty( "DR.onBehalfHeader", "X-DMAAP-DR-ON-BEHALF-OF");
feedContentType = p.getProperty( "DR.feedContentType", "application/vnd.dmaap-dr.feed");
subContentType = p.getProperty( "DR.subContentType", "application/vnd.dmaap-dr.subscription");
+ provURI = p.getProperty( "DR.ProvisioningURI", "/internal/prov");
logger.info( "provURL=" + provURL + " provApi=" + provApi + " behalfHeader=" + behalfHeader
+ " feedContentType=" + feedContentType + " subContentType=" + subContentType );
unit_test = p.getProperty( "UnitTest", "No" );
sub, nodep );
return makeConnection( provURL + uri );
}
+ public boolean makeDumpConnection() {
+ String url = provURL + provURI;
+ return makeConnection( url );
+ }
public boolean makeNodesConnection( String varName ) {
String uri = String.format("/internal/api/%s", varName);
}
+ // add double-quotes around a value
+ // hope his is easier to read than in-line escaping...
+ private String dq( String v ) {
+ return ( "\"" + v + "\"");
+ }
+ private String dq( String k, String v) {
+ return( dq(k) + ":" + dq(v));
+ }
+ private String dqc( String k, String v) {
+ return( dq(k) + ":" + dq(v) + ",");
+ }
+
+ private String dumpSimulation() {
+ logger.info( "enter dumpSimulation()");
+ String responseBody =
+ "{"
+ + dq("feeds") + ":["
+ + "{" + dq( "suspend") + ":false,"
+ + dq( "groupid") + ":0,"
+ + dqc( "description", "Some description" )
+ + dqc( "version", "m1.1")
+ + dq( "authorization") + ":"
+ + "{" + dq( "endpoint_addrs" ) + ":[],"
+ + dq( "classification", "unclassified")
+ + dq( "endpoint_ids") + ":[{"
+ + dqc( "password", "dradmin" )
+ + dq( "id", "dradmin")
+ + "}]}"
+ + dq( "last_mod") + ":1553738110000,"
+ + dq( "deleted") + ":false,"
+ + dq( "feedid") + ":1,"
+ + dqc( "name", "Default PM Feed")
+ + dq( "business_description") + ":\"\","
+ + dqc( "publisher", "onap")
+ + dq( "links") + ":{"
+ + dqc( "subscribe", "https://dmaap-dr-prov/subscribe/1")
+ + dqc( "log", "https://dmaap-dr-prov/feedlog/1")
+ + dqc( "publish", "https://dmaap-dr-prov/publish/1")
+ + dq( "self", "https:/dmaap-dr-prov/feed/1")
+ + "}"
+ + dq( "created_date") + ":1553738110000 }"
+ + "],"
+ + dq( "groups") + ":["
+ + "],"
+ + dq( "subscriptions") + ":["
+ + "],"
+ + dq( "ingress") + ":["
+ + "],"
+ + dq( "egress") + ":{"
+ + "},"
+ + dq( "routing") + ":["
+ + "],"
+ + "}";
+ return responseBody;
+ }
+
+ public String doGetDump( ApiError err ) {
+ logger.info( "entry: doGetDump() " );
+
+ String responsemessage = null;
+ String responseBody = null;
+
+ try {
+
+ uc.setRequestMethod("GET");
+ int rc = -1;
+
+
+ try {
+ uc.connect();
+
+
+ } catch (ProtocolException pe) {
+
+ // Rcvd error instead of 100-Continue
+ try {
+ // work around glitch in Java 1.7.0.21 and likely others
+ // without this, Java will connect multiple times to the server to run the same request
+ uc.setDoOutput(false);
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ }
+ }
+
+ rc = uc.getResponseCode();
+ logger.info( "http response code:" + rc );
+ responsemessage = uc.getResponseMessage();
+ logger.info( "responsemessage=" + responsemessage );
+
+
+
+ if (responsemessage == null) {
+
+ // work around for glitch in Java 1.7.0.21 and likely others
+ // When Expect: 100 is set and a non-100 response is received, the response message is not set but the response code is
+ String h0 = uc.getHeaderField(0);
+ if (h0 != null) {
+ int i = h0.indexOf(' ');
+ int j = h0.indexOf(' ', i + 1);
+ if (i != -1 && j != -1) {
+ responsemessage = h0.substring(j + 1);
+ }
+ }
+ }
+
+ err.setCode(rc); // may not really be an error, but we save rc
+ if (rc == 200 ) {
+ responseBody = bodyToString( uc.getInputStream() );
+ logger.info( "responseBody=" + responseBody );
+ } else {
+ err.setMessage(responsemessage);
+ }
+
+
+ } catch (ConnectException ce) {
+ if ( unit_test.equals( "Yes" ) ) {
+ err.setCode(200);
+ err.setMessage( "simulated response");
+ logger.info( "artificial 200 response from doGetNodes because unit_test =" + unit_test );
+ responseBody = dumpSimulation();
+
+ } else {
+ errorLogger.error( DmaapbcLogMessageEnum.HTTP_CONNECTION_EXCEPTION, provURL, ce.getMessage() );
+ err.setCode( 500 );
+ err.setMessage("Backend connection refused");
+ logger.error(ce.getMessage(), ce);
+ }
+ } catch (Exception e) {
+ if ( unit_test.equals( "Yes" ) ) {
+ err.setCode(200);
+ err.setMessage( "simulated response");
+ logger.info( "artificial 200 response from doGetNodes because unit_test =" + unit_test );
+ responseBody = dumpSimulation();
+
+ } else {
+ logger.error("Unable to read response ", e.getMessage());
+ }
+ } finally {
+
+ if ( uc != null ) uc.disconnect();
+ }
+
+ return responseBody;
+
+ }
}
}
return responseBuilder.success(nfeed);
}
+
+ @PUT
+ @ApiOperation( value = "sync feeds to existing DR",
+ notes = "When Bus Controller is deployed after DR, then it is possible"
+ + "that DR has previous provisioning data that needs to be imported"
+ + "into Bus Controller.",
+ response = Feed.class )
+ @ApiResponses( value = {
+ @ApiResponse( code = 200, message = "Success", response = Feed.class),
+ @ApiResponse( code = 400, message = "Error", response = ApiError.class )
+ })
+ @Path( "/sync")
+ public Response syncFeeds (
+ @QueryParam("hard") String hardParam
+ ) {
+ ApiService resp = new ApiService();
+
+ FeedService feedService = new FeedService();
+ boolean hard = false;
+ if ( hardParam != null && hardParam.equalsIgnoreCase("true")) {
+ hard = true;
+ }
+ feedService.sync( hard, resp.getErr() );
+ if ( resp.getErr().is2xx()) {
+ List<Feed> nfeeds = feedService.getAllFeeds();
+ GenericEntity<List<Feed>> list = new GenericEntity<List<Feed>>(nfeeds) {
+ };
+ return responseBuilder.success(list);
+ }
+ return responseBuilder.error(resp.getErr());
+ }
+
}
}
public DR_Pub removeDr_Pub( String pubId, ApiError err ) {
-
+ return removeDr_Pub( pubId, err, true );
+ }
+
+
+ public DR_Pub removeDr_Pub( String pubId, ApiError err, boolean hitDR ) {
DR_Pub pub = dr_pubs.get( pubId );
if ( pub == null ) {
err.setCode(Status.NOT_FOUND.getStatusCode());
}
public void removeDr_Sub( String key, ApiError apiError ) {
+ removeDr_Sub( key, apiError, true );
+ return;
+ }
+
+ public void removeDr_Sub( String key, ApiError apiError, boolean hitDR ) {
logger.debug( "enter removeDR_Subs()");
DR_Sub sub = dr_subs.get( key );
apiError.setCode(Status.NOT_FOUND.getStatusCode());
apiError.setFields( "subId");
apiError.setMessage("subId " + key + " not found");
- } else {
- DrProvConnection prov = new DrProvConnection();
- prov.makeSubPutConnection( key );
- String resp = prov.doDeleteDr_Sub( sub, apiError );
- logger.debug( "resp=" + resp );
+ } else {
+ if ( hitDR ) {
+ DrProvConnection prov = new DrProvConnection();
+ prov.makeSubPutConnection( key );
+ String resp = prov.doDeleteDr_Sub( sub, apiError );
+ logger.debug( "resp=" + resp );
+ } else {
+ apiError.setCode(200);
+ }
if ( apiError.is2xx() || unit_test.equals( "Yes" ) ) {
dr_subs.remove(key);
import org.onap.dmaap.dbcapi.util.RandomInteger;
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.apache.log4j.Logger;
+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.DR_Pub;
import org.onap.dmaap.dbcapi.model.DR_Sub;
import org.onap.dmaap.dbcapi.model.Feed;
+import org.onap.dmaap.dbcapi.model.MR_Client;
import org.onap.dmaap.dbcapi.model.DmaapObject.DmaapObject_Status;
import org.onap.dmaap.dbcapi.util.DmaapConfig;
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();
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);
// 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);
+ 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);
+ 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();
}
+
+
+ /*
+ * 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";
}
+
+ @Test
+ public void syncTestHard() {
+ ApiError err = new ApiError();
+ ds.sync( true, err );
+
+ assert( 200 == err.getCode());
+ }
+
}
Code Review / dmaap / dbcapi.git / commitdiff