code formatting by standard rules

Change-Id: I24dc78e826abf1e0cfeccfbd20d61c6723a0c519
Issue-ID: DMAAP-1211
Signed-off-by: piotr.karas <piotr.karas@nokia.com>

2 files changed, 831 insertions, 842 deletions

diff --git a/src/main/java/org/onap/dmaap/dbcapi/service/MR_ClientService.java b/src/main/java/org/onap/dmaap/dbcapi/service/MR_ClientService.java
index d0407c2..23627b5 100644
--- a/src/main/java/org/onap/dmaap/dbcapi/service/MR_ClientService.java
+++ b/src/main/java/org/onap/dmaap/dbcapi/service/MR_ClientService.java
@@ -9,9 +9,9 @@
  * 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.
@@ -22,288 +22,278 @@
 
 package org.onap.dmaap.dbcapi.service;
 
+import org.onap.dmaap.dbcapi.aaf.AafService;
+import org.onap.dmaap.dbcapi.aaf.AafService.ServiceType;
+import org.onap.dmaap.dbcapi.aaf.AafUserRole;
+import org.onap.dmaap.dbcapi.aaf.DmaapGrant;
+import org.onap.dmaap.dbcapi.aaf.DmaapPerm;
+import org.onap.dmaap.dbcapi.client.MrProvConnection;
+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.DcaeLocation;
+import org.onap.dmaap.dbcapi.model.DmaapObject.DmaapObject_Status;
+import org.onap.dmaap.dbcapi.model.MR_Client;
+import org.onap.dmaap.dbcapi.model.MR_Cluster;
+import org.onap.dmaap.dbcapi.model.Topic;
+import org.onap.dmaap.dbcapi.util.DmaapConfig;
+
+import javax.ws.rs.core.Response.Status;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 
-import javax.ws.rs.core.Response.Status;
+public class MR_ClientService extends BaseLoggingClass {
 
+    private static final String MR_CLIENT_ID = "mrClientId";
+    private int deleteLevel;
+    private Map<String, MR_Client> mr_clients = DatabaseClass.getMr_clients();
+    private Map<String, MR_Cluster> clusters = DatabaseClass.getMr_clusters();
+    private Map<String, DcaeLocation> locations = DatabaseClass.getDcaeLocations();
+    private DmaapService dmaap = new DmaapService();
+    private String centralCname;
 
+    public MR_ClientService() {
+        DmaapConfig p = (DmaapConfig) DmaapConfig.getConfig();
 
+        centralCname = p.getProperty("MR.CentralCname", "MRcname.not.set");
+        deleteLevel = Integer.valueOf(p.getProperty("MR.ClientDeleteLevel", "0"));
+    }
 
+    public List<MR_Client> getAllMr_Clients() {
+        return new ArrayList<>(mr_clients.values());
+    }
 
+    List<MR_Client> getAllMrClients(String fqtn) {
+        ArrayList<MR_Client> results = new ArrayList<>();
+        for (Map.Entry<String, MR_Client> entry : mr_clients.entrySet()) {
+            MR_Client client = entry.getValue();
+            if (fqtn.equals(client.getFqtn())) {
+                results.add(client);
+            }
+        }
+        return results;
+    }
 
+    List<MR_Client> getClientsByLocation(String location) {
+        List<MR_Client> results = new ArrayList<>();
+        for (Map.Entry<String, MR_Client> entry : mr_clients.entrySet()) {
+            MR_Client client = entry.getValue();
+            if (location.equals(client.getDcaeLocationName())) {
+                results.add(client);
+            }
+        }
+        return results;
+    }
 
 
+    public MR_Client getMr_Client(String key, ApiError apiError) {
+        MR_Client c = mr_clients.get(key);
+        if (c == null) {
+            apiError.setCode(Status.NOT_FOUND.getStatusCode());
+            apiError.setFields(MR_CLIENT_ID);
+            apiError.setMessage(MR_CLIENT_ID + " " + key + " not found");
+        } else {
+            apiError.setCode(200);
+        }
+        return c;
+    }
 
-import org.onap.dmaap.dbcapi.aaf.AafService;
-import org.onap.dmaap.dbcapi.aaf.DmaapGrant;
-import org.onap.dmaap.dbcapi.aaf.DmaapPerm;
-import org.onap.dmaap.dbcapi.aaf.AafService.ServiceType;
-import org.onap.dmaap.dbcapi.aaf.AafUserRole;
-import org.onap.dmaap.dbcapi.client.MrProvConnection;
-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.DcaeLocation;
-import org.onap.dmaap.dbcapi.model.MR_Client;
-import org.onap.dmaap.dbcapi.model.MR_Cluster;
-import org.onap.dmaap.dbcapi.model.Topic;
-import org.onap.dmaap.dbcapi.model.DmaapObject.DmaapObject_Status;
-import org.onap.dmaap.dbcapi.util.DmaapConfig;
+    public MR_Client addMr_Client(MR_Client client, Topic topic, ApiError err) {
+        if (client.getDcaeLocationName().isEmpty()) {
+            logger.info("Client  dcaeLocation that doesn't exist or not specified");
+            return null;
+        }
+        // original style: clients specified Role.  This has precedence for backwards
+        //                 compatibility.
+        // ONAP style: clients specify Identity to be assigned to generated Role
+        String role = client.getClientRole();
+        if (role != null) {
+            grantClientRolePerms(client, err);
+        } else if (client.hasClientIdentity()) {
+            if (client.isSubscriber()) {
+                role = topic.getSubscriberRole();
+                assignIdentityToRole(client, role, err);
+            }
+            if (client.isPublisher()) {
+                role = topic.getPublisherRole();
+                assignIdentityToRole(client, role, err);
+            }
+        }
+        if (!client.isStatusValid()) {
+            return null;
+        }
+        String centralFqdn = null;
+        DcaeLocation candidate = locations.get(client.getDcaeLocationName());
+
+        MR_Cluster cluster = clusters.get(client.getDcaeLocationName());
+        if (cluster != null && candidate != null) {
+            if (candidate.isCentral() && !topic.getReplicationCase().involvesFQDN()) {
+                centralFqdn = centralCname;
+            }
+            client.setTopicURL(cluster.genTopicURL(centralFqdn, client.getFqtn()));
+            if (centralFqdn == null) {
+                client.setStatus(addTopicToCluster(cluster, topic, err));
+                if (!err.is2xx() && err.getCode() != 409) {
+                    topic.setFqtn(err.getMessage());
+                    return null;
+                }
+
+            } else {
+                MR_ClusterService clusters = new MR_ClusterService();
+                //  MM should only exist for edge-to-central
+                //  we use a cname for the central target (default resiliency with no replicationGroup set)
+                // but still need to provision topics on all central MRs
+                for (MR_Cluster central : clusters.getCentralClusters()) {
+                    client.setStatus(addTopicToCluster(central, topic, err));
+                    if (!err.is2xx() && err.getCode() != 409) {
+                        topic.setFqtn(err.getMessage());
+                        return null;
+                    }
+                }
+            }
+
+        } else {
+            logger.warn("Client references a dcaeLocation that doesn't exist:" + client.getDcaeLocationName());
+            client.setStatus(DmaapObject_Status.STAGED);
+        }
+
+        mr_clients.put(client.getMrClientId(), client);
+
+        err.setCode(200);
+
+        return client;
+    }
+
+    private DmaapObject_Status addTopicToCluster(MR_Cluster cluster, Topic topic, ApiError err) {
+
+        MrProvConnection prov = new MrProvConnection();
+        logger.info("POST topic " + topic.getFqtn() + " to cluster " + cluster.getFqdn() + " in loc " + cluster.getDcaeLocationName());
+        if (prov.makeTopicConnection(cluster)) {
+            prov.doPostTopic(topic, err);
+            logger.info("response code: " + err.getCode());
+            if (err.is2xx() || err.getCode() == 409) {
+                return DmaapObject_Status.VALID;
+            }
+        }
+        return DmaapObject_Status.INVALID;
+    }
+
+    private void grantClientRolePerms(MR_Client client, ApiError err) {
+        AafService aaf = new AafService(ServiceType.AAF_TopicMgr);
+
+        String instance = ":topic." + client.getFqtn();
+        client.setStatus(DmaapObject_Status.VALID);
+        String role = client.getClientRole();
+        for (String want : client.getAction()) {
+            int rc;
+            DmaapPerm perm = new DmaapPerm(dmaap.getTopicPerm(), instance, want);
+            if (role != null) {
+                DmaapGrant g = new DmaapGrant(perm, role);
+                rc = aaf.addGrant(g);
+                if (rc != 201 && rc != 409) {
+                    client.setStatus(DmaapObject_Status.INVALID);
+                    err.setCode(rc);
+                    err.setMessage("Grant of " + dmaap.getTopicPerm() + "|" + instance + "|" + want + " failed for " + role);
+                    logger.warn(err.getMessage());
+                    return;
+                }
+            } else {
+                logger.warn("No Grant of " + dmaap.getTopicPerm() + "|" + instance + "|" + want + " because role is null ");
+            }
+        }
+    }
+
+    private void assignIdentityToRole(MR_Client client, String role, ApiError err) {
+        AafService aaf = new AafService(ServiceType.AAF_TopicMgr);
+
+        AafUserRole ur = new AafUserRole(client.getClientIdentity(), role);
+        int rc = aaf.addUserRole(ur);
+        if (rc != 201 && rc != 409) {
+            client.setStatus(DmaapObject_Status.INVALID);
+            err.setCode(rc);
+            err.setMessage("Failed to add user " + client.getClientIdentity() + "  to " + role);
+            logger.warn(err.getMessage());
+            return;
+        }
+        client.setStatus(DmaapObject_Status.VALID);
+
+    }
+
+    private void revokeClientPerms(MR_Client client, ApiError err) {
+        AafService aaf = new AafService(ServiceType.AAF_TopicMgr);
+
+        String instance = ":topic." + client.getFqtn();
+        client.setStatus(DmaapObject_Status.VALID);
+        for (String want : client.getAction()) {
+            int rc;
+            DmaapPerm perm = new DmaapPerm(dmaap.getTopicPerm(), instance, want);
+            DmaapGrant g = new DmaapGrant(perm, client.getClientRole());
+            rc = aaf.delGrant(g);
+            if (rc != 200 && rc != 404) {
+                client.setStatus(DmaapObject_Status.INVALID);
+                err.setCode(rc);
+                err.setMessage("Revoke of " + dmaap.getTopicPerm() + "|" + instance + "|" + want + " failed for " + client.getClientRole());
+                logger.warn(err.getMessage());
+                return;
+            }
+        }
+    }
+
+    public MR_Client updateMr_Client(MR_Client client, ApiError apiError) {
+        MR_Client c = mr_clients.get(client.getMrClientId());
+        if (c == null) {
+            apiError.setCode(Status.NOT_FOUND.getStatusCode());
+            apiError.setFields(MR_CLIENT_ID);
+            apiError.setMessage("mrClientId " + client.getMrClientId() + " not found");
+        } else {
+            apiError.setCode(200);
+        }
+        mr_clients.put(client.getMrClientId(), client);
+        return client;
+    }
+
+    public void removeMr_Client(String key, boolean updateTopicView, ApiError apiError) {
+        MR_Client client = mr_clients.get(key);
+        if (client == null) {
+            apiError.setCode(Status.NOT_FOUND.getStatusCode());
+            apiError.setFields(MR_CLIENT_ID);
+            apiError.setMessage("mrClientId " + key + " not found");
+            return;
+        } else {
+            apiError.setCode(200);
+        }
+
+        if (updateTopicView) {
+
+            TopicService topics = new TopicService();
+
+            Topic t = topics.getTopic(client.getFqtn(), apiError);
+            if (t != null) {
+                List<MR_Client> tc = t.getClients();
+                for (MR_Client c : tc) {
+                    if (c.getMrClientId().equals(client.getMrClientId())) {
+                        tc.remove(c);
+                        break;
+                    }
+                }
+                t.setClients(tc);
+                topics.updateTopic(t, apiError);
+            }
+
+        }
+
+        // remove from AAF
+        if (deleteLevel >= 2) {
+            revokeClientPerms(client, apiError);
+            if (!apiError.is2xx()) {
+                return;
+            }
+        }
+        // remove from DB
+        if (deleteLevel >= 1) {
+            mr_clients.remove(key);
+        }
+    }
 
-public class MR_ClientService extends BaseLoggingClass{
-
-	private static final String MR_CLIENT_ID = "mrClientId";
-	private int deleteLevel;
-	private Map<String, MR_Client> mr_clients = DatabaseClass.getMr_clients();
-	private Map<String, MR_Cluster> clusters = DatabaseClass.getMr_clusters();
-	private Map<String, DcaeLocation> locations = DatabaseClass.getDcaeLocations();
-	private DmaapService dmaap = new DmaapService();
-	private String centralCname;
-	
-	public MR_ClientService() {
-		DmaapConfig p = (DmaapConfig)DmaapConfig.getConfig();
-	
-		centralCname = p.getProperty("MR.CentralCname", "MRcname.not.set");
-		deleteLevel = Integer.valueOf(p.getProperty("MR.ClientDeleteLevel", "0" ));
-	}
-		
-	public List<MR_Client> getAllMr_Clients() {
-		return new ArrayList<>(mr_clients.values());
-	}
-	
-	List<MR_Client> getAllMrClients(String fqtn) {
-		ArrayList<MR_Client> results = new ArrayList<>();
-		for (Map.Entry<String, MR_Client> entry : mr_clients.entrySet())
-		{
-			MR_Client client = entry.getValue();
-		    if ( fqtn.equals(client.getFqtn() ) ) {
-		    	results.add( client );
-		    }
-		}
-		return results;
-	}	
-
-	List<MR_Client> getClientsByLocation(String location) {
-		List<MR_Client> results = new ArrayList<>();
-		for (Map.Entry<String, MR_Client> entry : mr_clients.entrySet())
-		{
-			MR_Client client = entry.getValue();
-		    if ( location.equals(client.getDcaeLocationName() ) ) {
-		    	results.add( client );
-		    }
-		}
-		return results;
-	}	
-
-	
-	public MR_Client getMr_Client( String key, ApiError apiError ) {			
-		MR_Client c =  mr_clients.get( key );
-		if ( c == null ) {
-			apiError.setCode(Status.NOT_FOUND.getStatusCode());
-			apiError.setFields(MR_CLIENT_ID);
-			apiError.setMessage(MR_CLIENT_ID+ " " + key + " not found" );
-		} else {
-			apiError.setCode(200);
-		}
-		return c;
-	}
-
-	public MR_Client addMr_Client( MR_Client client, Topic topic, ApiError err ) {
-		if ( client.getDcaeLocationName().isEmpty()) {
-			logger.info( "Client  dcaeLocation that doesn't exist or not specified" );
-			return null;
-		}
-		// original style: clients specified Role.  This has precedence for backwards
-		//                 compatibility.
-		// ONAP style: clients specify Identity to be assigned to generated Role
-		String role = client.getClientRole();
-		if ( role != null ) {
-			grantClientRolePerms(  client,  err);
-		} else if ( client.hasClientIdentity() ){
-			if ( client.isSubscriber() ) {
-				role = topic.getSubscriberRole();
-				assignIdentityToRole( client, role, err );
-			} 
-			if (client.isPublisher() ) {
-				role = topic.getPublisherRole();
-				assignIdentityToRole( client, role, err );
-			}	
-		} 
-		if ( ! client.isStatusValid()) {
-			return null;
-		}
-		String centralFqdn = null;
-		DcaeLocation candidate = locations.get(client.getDcaeLocationName());
-
-		MR_Cluster cluster = clusters.get( client.getDcaeLocationName());
-		if (  cluster != null && candidate != null ) {
-			if ( candidate.isCentral() && ! topic.getReplicationCase().involvesFQDN() ) {
-				centralFqdn = centralCname;
-			}
-			client.setTopicURL(cluster.genTopicURL(centralFqdn, client.getFqtn()));
-			if ( centralFqdn == null ) {
-				client.setStatus( addTopicToCluster( cluster, topic, err));
-				if( ! err.is2xx() && err.getCode() != 409 ) {
-					topic.setFqtn(err.getMessage());
-					return null;
-				}
-			
-			} else {
-				MR_ClusterService clusters = new MR_ClusterService();	
-				//  MM should only exist for edge-to-central
-				//  we use a cname for the central target (default resiliency with no replicationGroup set)
-				// but still need to provision topics on all central MRs
-				for( MR_Cluster central: clusters.getCentralClusters() ) {
-					client.setStatus( addTopicToCluster( central, topic, err));
-					if( ! err.is2xx() && err.getCode() != 409 ) {
-						topic.setFqtn(err.getMessage());
-						return null;
-					}
-				}
-			}
-			
-		} else {
-			logger.warn( "Client references a dcaeLocation that doesn't exist:" + client.getDcaeLocationName());
-			client.setStatus( DmaapObject_Status.STAGED);
-		}
-
-		mr_clients.put( client.getMrClientId(), client );
-
-		err.setCode(200);
-		
-		return client;
-	}
-	
-	private DmaapObject_Status addTopicToCluster( MR_Cluster cluster, Topic topic, ApiError err  ){
-		
-		MrProvConnection prov = new MrProvConnection();
-		logger.info( "POST topic " + topic.getFqtn() + " to cluster " + cluster.getFqdn() + " in loc " + cluster.getDcaeLocationName());
-		if ( prov.makeTopicConnection(cluster)) {
-			prov.doPostTopic(topic, err);
-			logger.info( "response code: " + err.getCode() );
-			if ( err.is2xx() || err.getCode() == 409 ) {
-				return DmaapObject_Status.VALID;
-			} 
-		}
-		return DmaapObject_Status.INVALID;
-	}
-	
-	private void grantClientRolePerms( MR_Client client, ApiError err) {
-		AafService aaf = new AafService(ServiceType.AAF_TopicMgr);
-		
-		String instance = ":topic." + client.getFqtn();
-		client.setStatus( DmaapObject_Status.VALID);
-		String role = client.getClientRole();
-		for( String want : client.getAction() ) {
-			int rc;
-			DmaapPerm perm = new DmaapPerm( dmaap.getTopicPerm(), instance, want );
-			if ( role != null ) {
-				DmaapGrant g = new DmaapGrant( perm, role );
-				rc = aaf.addGrant( g );
-				if ( rc != 201 && rc != 409 ) {
-					client.setStatus( DmaapObject_Status.INVALID);
-					err.setCode(rc);
-					err.setMessage( "Grant of " + dmaap.getTopicPerm() + "|" + instance + "|" + want + " failed for " + role );
-					logger.warn( err.getMessage());
-					return;
-				} 
-			} else {
-				logger.warn( "No Grant of " + dmaap.getTopicPerm() + "|" + instance + "|" + want + " because role is null " );
-			}
-		}
-	}
-	
-	private void assignIdentityToRole( MR_Client client, String role, ApiError err ) {
-		AafService aaf = new AafService(ServiceType.AAF_TopicMgr);
-
-		AafUserRole ur = new AafUserRole( client.getClientIdentity(), role );
-		int rc = aaf.addUserRole( ur );
-		if ( rc != 201 && rc != 409 ) {
-			client.setStatus( DmaapObject_Status.INVALID);
-			err.setCode(rc);
-			err.setMessage( "Failed to add user " + client.getClientIdentity()+  "  to " + role );
-			logger.warn( err.getMessage());
-			return;
-		}
-		client.setStatus( DmaapObject_Status.VALID);
-
-	}
-	private void revokeClientPerms( MR_Client client, ApiError err) {
-		AafService aaf = new AafService(ServiceType.AAF_TopicMgr);
-		
-		String instance = ":topic." + client.getFqtn();
-		client.setStatus( DmaapObject_Status.VALID);
-		for( String want : client.getAction() ) {
-			int rc;
-			DmaapPerm perm = new DmaapPerm( dmaap.getTopicPerm(), instance, want );
-			DmaapGrant g = new DmaapGrant( perm, client.getClientRole() );
-			rc = aaf.delGrant( g );
-			if ( rc != 200 && rc != 404 ) {
-				client.setStatus( DmaapObject_Status.INVALID);
-				err.setCode(rc);
-				err.setMessage( "Revoke of " + dmaap.getTopicPerm() + "|" + instance + "|" + want + " failed for " + client.getClientRole() );
-				logger.warn( err.getMessage());
-				return;
-			} 
-		}
-	}
-	
-	public MR_Client updateMr_Client( MR_Client client, ApiError apiError ) {
-		MR_Client c =  mr_clients.get( client.getMrClientId());
-		if ( c == null ) {
-			apiError.setCode(Status.NOT_FOUND.getStatusCode());
-			apiError.setFields(MR_CLIENT_ID);
-			apiError.setMessage("mrClientId " + client.getMrClientId() + " not found" );
-		} else {
-			apiError.setCode(200);
-		}
-		mr_clients.put( client.getMrClientId(), client );
-		return client;
-	}
-		
-	public void removeMr_Client( String key, boolean updateTopicView, ApiError apiError ) {
-		MR_Client client =  mr_clients.get( key );
-		if ( client == null ) {
-			apiError.setCode(Status.NOT_FOUND.getStatusCode());
-			apiError.setFields(MR_CLIENT_ID);
-			apiError.setMessage("mrClientId " + key + " not found" );
-			return;
-		} else {
-			apiError.setCode(200);
-		}
-		
-		if (updateTopicView) {
-
-			TopicService topics = new TopicService();
-			
-			Topic t = topics.getTopic(client.getFqtn(), apiError );
-			if ( t != null ) {	
-				List<MR_Client> tc = t.getClients();
-				for( MR_Client c: tc) {
-					if ( c.getMrClientId().equals(client.getMrClientId())) {
-						tc.remove(c);
-						break;
-					}
-				}
-				t.setClients(tc);
-				topics.updateTopic( t, apiError );
-			}
-
-		}
-
-		// remove from AAF
-		if ( deleteLevel >= 2 ) {
-			revokeClientPerms( client, apiError );
-			if ( ! apiError.is2xx()) {
-				return;
-			}
-		}
-		// remove from DB 
-		if ( deleteLevel >= 1 ) {
-			mr_clients.remove(key);
-		}
-	}
-	
 }
diff --git a/src/main/java/org/onap/dmaap/dbcapi/service/TopicService.java b/src/main/java/org/onap/dmaap/dbcapi/service/TopicService.java
index 08e58be..83591dd 100644
--- a/src/main/java/org/onap/dmaap/dbcapi/service/TopicService.java
+++ b/src/main/java/org/onap/dmaap/dbcapi/service/TopicService.java
@@ -9,9 +9,9 @@
  * 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.
@@ -22,20 +22,11 @@
 
 package org.onap.dmaap.dbcapi.service;
 
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-import javax.ws.rs.core.Response.Status;
-
 import org.onap.dmaap.dbcapi.aaf.AafNamespace;
 import org.onap.dmaap.dbcapi.aaf.AafRole;
 import org.onap.dmaap.dbcapi.aaf.AafService;
-import org.onap.dmaap.dbcapi.aaf.DmaapGrant;
 import org.onap.dmaap.dbcapi.aaf.AafService.ServiceType;
+import org.onap.dmaap.dbcapi.aaf.DmaapGrant;
 import org.onap.dmaap.dbcapi.aaf.DmaapPerm;
 import org.onap.dmaap.dbcapi.database.DatabaseClass;
 import org.onap.dmaap.dbcapi.logging.BaseLoggingClass;
@@ -52,565 +43,573 @@ import org.onap.dmaap.dbcapi.util.DmaapConfig;
 import org.onap.dmaap.dbcapi.util.Fqdn;
 import org.onap.dmaap.dbcapi.util.Graph;
 
+import javax.ws.rs.core.Response.Status;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
 public class TopicService extends BaseLoggingClass {
 
-	
-
-	// REF: https://wiki.web.att.com/pages/viewpage.action?pageId=519703122
-	private static String defaultGlobalMrHost;
-	
-	private Map<String, Topic> mr_topics;
-	
-	private static DmaapService dmaapSvc = new DmaapService();
-	private MR_ClientService clientService;
-	private MR_ClusterService clusters;
-	private DcaeLocationService locations;
-	private MirrorMakerService	bridge;
-	
-	private static String centralCname;
-	private static boolean createTopicRoles;
-	private boolean strictGraph = true;
-	private boolean mmPerMR;
-
-
-	public TopicService(){
-		this(DatabaseClass.getTopics(), new MR_ClientService(), (DmaapConfig)DmaapConfig.getConfig(),
-				new MR_ClusterService(), new DcaeLocationService(), new MirrorMakerService());
-
-	}
-
-	TopicService(Map<String, Topic> mr_topics,  MR_ClientService clientService, DmaapConfig p,
-				 MR_ClusterService clusters, DcaeLocationService locations, MirrorMakerService	bridge) {
-		this.mr_topics = mr_topics;
-		this.clientService = clientService;
-		defaultGlobalMrHost = p.getProperty("MR.globalHost", "global.host.not.set");
-		centralCname = p.getProperty("MR.CentralCname");
-		createTopicRoles = "true".equalsIgnoreCase(p.getProperty("aaf.CreateTopicRoles", "true"));
-		String unit_test = p.getProperty( "UnitTest", "No" );
-		if ( "Yes".equals(unit_test)) {
-			strictGraph = false;
-		}
-		mmPerMR = "true".equalsIgnoreCase(p.getProperty("MirrorMakerPerMR", "true"));
-		logger.info( "TopicService properties: CentralCname=" + centralCname +
-				"   defaultGlobarlMrHost=" + defaultGlobalMrHost +
-				" createTopicRoles=" + createTopicRoles +
-				" mmPerMR=" + mmPerMR );
-		this.clusters = clusters;
-		this.locations = locations;
-		this.bridge = bridge;
-	}
-
-	public Map<String, Topic> getTopics() {
-		return mr_topics;
-	}
-		
-	public List<Topic> getAllTopics() {
-		return getAllTopics( true );
-	}
-	public List<Topic> getAllTopicsWithoutClients() {
-		return getAllTopics(false);
-	}
-	
-	private List<Topic> getAllTopics( Boolean withClients ) {
-		ArrayList<Topic> topics = new ArrayList<>(mr_topics.values());
-		if ( withClients ) {
-			for( Topic topic: topics ) {
-				topic.setClients(clientService.getAllMrClients(topic.getFqtn()));
-			}
-		}
-		return topics;
-	}
-	
-		
-	public Topic getTopic( String key, ApiError apiError ) {	
-		logger.info( "getTopic: key=" + key);
-		Topic t = mr_topics.get( key );
-		if ( t == null ) {
-			apiError.setCode(Status.NOT_FOUND.getStatusCode());
-			apiError.setFields( "fqtn");
-			apiError.setMessage("topic with fqtn " + key + " not found");
-			return null;
-		}
-		t.setClients( clientService.getAllMrClients( key ));
-		apiError.setCode(Status.OK.getStatusCode());
-		return t;
-	}
-	
-	private void aafTopicSetup(Topic topic, ApiError err ) {
-		
-		String nsr = dmaapSvc.getDmaap().getTopicNsRoot();
-		if ( nsr == null ) {
-			err.setCode(500);
-			err.setMessage("Unable to establish AAF namespace root: (check /dmaap object)"  );
-			err.setFields("topicNsRoot");
-			return;
-		}
-
-		// establish AAF Connection using TopicMgr identity
-		AafService aaf = new AafService(ServiceType.AAF_TopicMgr);
-		
-		AafRole pubRole = null;
-		AafRole subRole = null;
-		
-		// creating Topic Roles was not an original feature.
-		// For backwards compatibility, only do this if the feature is enabled.
-		// Also, if the namespace of the topic is a foreign namespace, (i.e. not the same as our root ns)
-		// then we likely don't have permission to create sub-ns and Roles so don't try.
-		if ( createTopicRoles && topic.getFqtn().startsWith(nsr)) {
-			// create AAF namespace for this topic
-			AafNamespace ns = new AafNamespace( topic.getFqtn(), aaf.getIdentity());
-			{
-				int rc = aaf.addNamespace( ns );
-				if ( rc != 201 && rc != 409 ) {
-					err.setCode(500);
-					err.setMessage("Unexpected response from AAF:" + rc );
-					err.setFields("namespace:" + topic.getFqtn() + " identity="+ aaf.getIdentity());
-					return;
-				}
-			}
-			
-			// create AAF Roles for MR clients of this topic
-			String rn = "publisher";
-			pubRole = new AafRole( topic.getFqtn(), rn );
-			int rc = aaf.addRole( pubRole );
-			if ( rc != 201 && rc != 409 ) {
-				err.setCode(500);
-				err.setMessage("Unexpected response from AAF:" + rc );
-				err.setFields("topic:" + topic.getFqtn() + " role="+ rn);
-				return;
-			}
-			topic.setPublisherRole( pubRole.getFullyQualifiedRole() );
-			
-			rn = "subscriber";
-			subRole = new AafRole( topic.getFqtn(), rn );
-			rc = aaf.addRole( subRole );
-			if ( rc != 201 && rc != 409 ) {
-				err.setCode(500);
-				err.setMessage("Unexpected response from AAF:" + rc );
-				err.setFields("topic:" + topic.getFqtn() + " role="+ rn);
-				return;
-			}
-			topic.setSubscriberRole( subRole.getFullyQualifiedRole() );
-		}
-	
-		// create AAF perms checked by MR
-		String instance = ":topic." + topic.getFqtn();
-		String[] actions = { "pub", "sub", "view" };
-		String t = dmaapSvc.getTopicPerm();
-		for ( String action : actions ){
-			DmaapPerm perm = new DmaapPerm( t, instance, action );
-			int rc = aaf.addPerm( perm );
-			if ( rc != 201 && rc != 409 ) {
-				err.setCode(500);
-				err.setMessage("Unexpected response from AAF:" + rc );
-				err.setFields("t="+t + " instance="+ instance + " action="+ action);
-				return;
-			}
-			if ( createTopicRoles )	{
-				// Grant perms to our default Roles
-				if ( action.equals( "pub") || action.equals( "view") ) {
-					DmaapGrant g = new DmaapGrant( perm, pubRole.getFullyQualifiedRole() );
-					rc = aaf.addGrant( g );
-					if ( rc != 201 && rc != 409 ) {
-						err.setCode(rc);
-						err.setMessage( "Grant of " + perm.toString() + " failed for " + pubRole.getFullyQualifiedRole() );
-						logger.warn( err.getMessage());
-						return;
-					} 
-				}
-				if ( action.equals( "sub") || action.equals( "view") ) {
-					DmaapGrant g = new DmaapGrant( perm, subRole.getFullyQualifiedRole() );
-					rc = aaf.addGrant( g );
-					if ( rc != 201 && rc != 409 ) {
-						err.setCode(rc);
-						err.setMessage( "Grant of " + perm.toString() + " failed for " + subRole.getFullyQualifiedRole() );
-						logger.warn( err.getMessage());
-						return;
-					} 
-				}
-			}
-
-		}
-	}
-
-	public Topic addTopic( Topic topic, ApiError err, Boolean useExisting ) {
-		logger.info( "Entry: addTopic");
-		logger.info( "Topic name=" + topic.getTopicName() + " fqtnStyle=" + topic.getFqtnStyle() );
-		String nFqtn =  topic.genFqtn();
-		logger.info( "FQTN=" + nFqtn );
-		Topic pTopic = getTopic( nFqtn, err );
-		if ( pTopic != null ) {
-			String t = "topic already exists: " + nFqtn;
-			logger.info( t );
-			if (  useExisting ) {
-				err.setCode(Status.OK.getStatusCode());
-				return pTopic;
-			}
-			err.setMessage( t );
-			err.setFields( "fqtn");
-			err.setCode(Status.CONFLICT.getStatusCode());
-			return null;
-		}
-		err.reset();  // err filled with NOT_FOUND is expected case, but don't want to litter...
-
-		topic.setFqtn( nFqtn );
-		
-		aafTopicSetup( topic, err );
-		if ( err.getCode() >= 400 ) {
-			return null;
-		}	
-
-		if ( topic.getReplicationCase().involvesGlobal() ) {
-			if ( topic.getGlobalMrURL() == null ) {
-				topic.setGlobalMrURL(defaultGlobalMrHost);
-			}
-			if ( ! Fqdn.isValid( topic.getGlobalMrURL())) {
-				logger.error( "GlobalMR FQDN not valid: " + topic.getGlobalMrURL());
-				topic.setStatus( DmaapObject_Status.INVALID);
-				err.setCode(500);
-				err.setMessage("Value is not a valid FQDN:" +  topic.getGlobalMrURL() );
-				err.setFields("globalMrURL");
-	
-				return null;
-			}
-		}
-
-
-		if ( topic.getNumClients() > 0 ) {
-			ArrayList<MR_Client> clients = new ArrayList<MR_Client>(topic.getClients());
-		
-	
-			ArrayList<MR_Client> clients2 = new ArrayList<MR_Client>();
-			for ( Iterator<MR_Client> it = clients.iterator(); it.hasNext(); ) {
-				MR_Client c = it.next();
-
-				logger.info( "c fqtn=" + c.getFqtn() + " ID=" + c.getMrClientId() + " url=" + c.getTopicURL());
-				MR_Client nc = new MR_Client( c.getDcaeLocationName(), topic.getFqtn(), c.getClientRole(), c.getAction());
-				nc.setFqtn(topic.getFqtn());
-				nc.setClientIdentity( c.getClientIdentity());
-				logger.info( "nc fqtn=" + nc.getFqtn() + " ID=" + nc.getMrClientId() + " url=" + nc.getTopicURL());
-				clients2.add( clientService.addMr_Client(nc, topic, err));
-				if ( ! err.is2xx()) {
-					return null;
-				}
-			}
-
-			topic.setClients(clients2);
-		}
-
-		Topic ntopic = checkForBridge( topic, err );
-		if ( ntopic == null ) {
-			topic.setStatus( DmaapObject_Status.INVALID);
-			if ( ! err.is2xx()) {
-				return null;
-			}
-		}
-
-		
-		mr_topics.put( nFqtn, ntopic );
-
-		err.setCode(Status.OK.getStatusCode());
-		return ntopic;
-	}
-	
-		
-	public Topic updateTopic( Topic topic, ApiError err ) {
-		logger.info( "updateTopic: entry");
-		logger.info( "updateTopic: topic=" + topic);
-		logger.info( "updateTopic: fqtn=" + topic.getFqtn() );
-		if ( topic.getFqtn().isEmpty()) {
-			return null;
-		}
-		logger.info( "updateTopic: call checkForBridge");
-		Topic ntopic = checkForBridge( topic, err );
-		if ( ntopic == null ) {
-			topic.setStatus( DmaapObject_Status.INVALID);
-			if ( ! err.is2xx() ) {
-				return null;
-			}
-		}
-		if(ntopic != null) {
-			logger.info( "updateTopic: call put");
-			mr_topics.put( ntopic.getFqtn(), ntopic );
-		}
-		err.setCode(Status.OK.getStatusCode());
-		return ntopic;
-	}
-		
-	public Topic removeTopic( String pubId, ApiError apiError ) {
-		Topic topic = mr_topics.get(pubId);
-		if ( topic == null ) {
-			apiError.setCode(Status.NOT_FOUND.getStatusCode());
-			apiError.setMessage("Topic " + pubId + " does not exist");
-			apiError.setFields("fqtn");
-			return null;
-		}
-		ArrayList<MR_Client> clients = new ArrayList<MR_Client>(clientService.getAllMrClients( pubId ));
-		for ( Iterator<MR_Client> it = clients.iterator(); it.hasNext(); ) {
-			MR_Client c = it.next();
-			
-	
-			clientService.removeMr_Client(c.getMrClientId(), false, apiError);
-			if ( ! apiError.is2xx()) {
-				return null;
-			}
-		}
-		apiError.setCode(Status.OK.getStatusCode());
-		return mr_topics.remove(pubId);
-	}	
-	public static ApiError setBridgeClientPerms( MR_Cluster node ) {
-		DmaapConfig p = (DmaapConfig)DmaapConfig.getConfig();
-		String mmProvRole = p.getProperty("MM.ProvRole");
-		String mmAgentRole = p.getProperty("MM.AgentRole");
-		String[] Roles = { mmProvRole, mmAgentRole };
-		String[] actions = { "view", "pub", "sub" };
-		Topic bridgeAdminTopic = new Topic().init();
-		bridgeAdminTopic.setTopicName( dmaapSvc.getBridgeAdminFqtn() );
-		bridgeAdminTopic.setTopicDescription( "RESERVED topic for MirroMaker Provisioning");
-		bridgeAdminTopic.setOwner( "DBC" );
-		
-		ArrayList<MR_Client> clients = new ArrayList<MR_Client>();
-		for( String role: Roles ) {
-			MR_Client client = new MR_Client();
-			client.setAction(actions);
-			client.setClientRole(role);
-			client.setDcaeLocationName( node.getDcaeLocationName());
-			clients.add( client );
-		}
-		bridgeAdminTopic.setClients(clients);
-		
-		TopicService ts = new TopicService();
-		ApiError err = new ApiError();
-		ts.addTopic(bridgeAdminTopic, err, true);
-		
-		if ( err.is2xx() || err.getCode() == 409 ){
-			err.setCode(200);
-			return err;
-		}
-		
-		errorLogger.error( DmaapbcLogMessageEnum.TOPIC_CREATE_ERROR,  bridgeAdminTopic.getFqtn(), Integer.toString(err.getCode()), err.getFields(), err.getMessage());
-		return err;
-	}	
-	
-	
-	public Topic checkForBridge( Topic topic, ApiError err ) {
-		logger.info( "checkForBridge: entry");
-		logger.info( "fqtn=" + topic.getFqtn() + " replicatonType=" + topic.getReplicationCase());
-		if ( topic.getReplicationCase() == ReplicationType.REPLICATION_NONE ) {
-			topic.setStatus( DmaapObject_Status.VALID);
-			return topic;	
-		}
-		
-		boolean anythingWrong = false;
-		
-		Set<String> groups = clusters.getGroups();
-		for ( String g : groups ) {
-			logger.info( "buildBridge for " + topic.getFqtn() + " on group" + g);
-			anythingWrong |= buildBridge( topic, err, g );
-		}
-		if ( anythingWrong ) {
-			topic.setStatus( DmaapObject_Status.INVALID);
-			if ( ! err.is2xx() ) {
-				return null;
-			}	
-		} else {
-			topic.setStatus( DmaapObject_Status.VALID);
-		}
-		return topic;
-	}
-		
-	private boolean buildBridge( Topic topic, ApiError err, String group ) {
-		logger.info( "buildBridge: entry");
-		boolean anythingWrong = false;
-		Graph graph;
-		logger.info( "buildBridge: strictGraph=" + strictGraph );
-		if ( group == null || group.isEmpty() ) {
-			graph = new Graph( topic.getClients(), strictGraph );
-		} else {
-			graph = new Graph( topic.getClients(), strictGraph, group );
-		}
-		logger.info( "buildBridge: graph=" + graph );
-		MR_Cluster groupCentralCluster = null;
-		
-		
-		if ( graph.isEmpty() ) {
-			logger.info( "buildBridge: graph is empty.  return false" );
-			return false;
-		} else if ( group == null &&  topic.getReplicationCase().involvesFQDN() ) {
-			logger.info( "buildBridge: group is null and replicationCaseInvolvesFQDN. return false" );
-			return false;
-		} else if ( ! graph.hasCentral() ) {
-			logger.warn( "Topic " + topic.getFqtn() + " wants to be " + topic.getReplicationCase() + " but has no central clients");
-			return true;
-		} else {
-			groupCentralCluster = clusters.getMr_ClusterByLoc(graph.getCentralLoc());
-		}
-		Collection<String> clientLocations = graph.getKeys();
-		for( String loc : clientLocations ) {
-			logger.info( "loc=" + loc );
-			DcaeLocation location = locations.getDcaeLocation(loc);
-			MR_Cluster cluster = clusters.getMr_ClusterByLoc(loc);
-			logger.info( "cluster=" + cluster + " at "+ cluster.getDcaeLocationName() );
-			logger.info( "location.isCentral()="+location.isCentral() + " getCentralLoc()=" + graph.getCentralLoc() );
-
-			
-				
-			String source = null;
-			String target = null;
-			
-			/*
-			 * Provision Edge to Central bridges...
-			 */
-			if ( ! location.isCentral()  && ! graph.getCentralLoc().equals(cluster.getDcaeLocationName()) ) {
-				switch( topic.getReplicationCase() ) {
-				case REPLICATION_EDGE_TO_CENTRAL:
-				case REPLICATION_EDGE_TO_CENTRAL_TO_GLOBAL:  // NOTE: this is for E2C portion only
-					source = cluster.getFqdn();
-					target = (mmPerMR)? groupCentralCluster.getFqdn() : centralCname;
-					logger.info( "REPLICATION_EDGE_TO_CENTRAL: source=" + source + " target=" +target );
-					break;
-				case REPLICATION_CENTRAL_TO_EDGE:
-				case REPLICATION_GLOBAL_TO_CENTRAL_TO_EDGE:  // NOTE: this is for C2E portion only
-					source = (mmPerMR) ? groupCentralCluster.getFqdn() : centralCname;
-					target = cluster.getFqdn();
-					break;
-				case REPLICATION_CENTRAL_TO_GLOBAL:
-				case REPLICATION_GLOBAL_TO_CENTRAL:
-				case REPLICATION_FQDN_TO_GLOBAL:
-				case REPLICATION_GLOBAL_TO_FQDN:
-					break;
-
-				case REPLICATION_EDGE_TO_FQDN:
-				case REPLICATION_EDGE_TO_FQDN_TO_GLOBAL:  // NOTE: this is for E2C portion only
-					source = cluster.getFqdn();
-					target = groupCentralCluster.getFqdn();
-					break;
-				case REPLICATION_FQDN_TO_EDGE:
-				case REPLICATION_GLOBAL_TO_FQDN_TO_EDGE:  // NOTE: this is for F2E portion only
-					source = groupCentralCluster.getFqdn();
-					target = cluster.getFqdn();
-					break;
-
-				default:
-					logger.error( "Unexpected value for ReplicationType ("+ topic.getReplicationCase() + ") for topic " + topic.getFqtn() );
-					anythingWrong = true;
-					err.setCode(400);
-					err.setFields("topic=" + topic.genFqtn() + " replicationCase="
-							+ topic.getReplicationCase() );
-					err.setMessage("Unexpected value for ReplicationType");
-					continue;
-				}
-
-			} else if ( location.isCentral() && graph.getCentralLoc().equals(cluster.getDcaeLocationName()) ) {
-				/*
-				 * Provision Central to Global bridges
-				 */
-				switch( topic.getReplicationCase() ) {
-
-				case REPLICATION_CENTRAL_TO_GLOBAL:
-				case REPLICATION_EDGE_TO_CENTRAL_TO_GLOBAL:
-					source = centralCname;
-					target = topic.getGlobalMrURL();
-					break;
-				case REPLICATION_GLOBAL_TO_CENTRAL:
-				case REPLICATION_GLOBAL_TO_CENTRAL_TO_EDGE:  // NOTE: this is for G2C portion only
-					source = topic.getGlobalMrURL();
-					target = centralCname;
-					break;
-
-				case REPLICATION_EDGE_TO_FQDN_TO_GLOBAL:  // NOTE: this is for E2F portion only
-					source = groupCentralCluster.getFqdn();
-					target = topic.getGlobalMrURL();
-					break;
-
-				case REPLICATION_FQDN_TO_GLOBAL:
-					source = groupCentralCluster.getFqdn();
-					target = topic.getGlobalMrURL();
-					break;
-					
-				case REPLICATION_GLOBAL_TO_FQDN:
-				case REPLICATION_GLOBAL_TO_FQDN_TO_EDGE:  // NOTE: this is for G2F portion only
-					source = topic.getGlobalMrURL();
-					target = groupCentralCluster.getFqdn();
-					break;
-
-				case REPLICATION_FQDN_TO_EDGE:
-				case REPLICATION_EDGE_TO_FQDN:
-				case REPLICATION_EDGE_TO_CENTRAL:
-				case REPLICATION_CENTRAL_TO_EDGE:
-					break;
-				default:
-					logger.error( "Unexpected value for ReplicationType ("+ topic.getReplicationCase() + ") for topic " + topic.getFqtn() );
-					anythingWrong = true;
-					err.setCode(400);
-					err.setFields("topic=" + topic.genFqtn() + " replicationCase="
-							+ topic.getReplicationCase() );
-					err.setMessage("Unexpected value for ReplicationType");
-					continue;
-				}				
-			} else {
-				logger.warn( "dcaeLocation " + loc + " is neither Edge nor Central so no mmagent provisioning was done");
-				anythingWrong = true;
-				continue;
-			}
-			if ( source != null && target != null ) {
-				try { 
-					logger.info( "Create a MM from " + source + " to " + target );
-					MirrorMaker mm = bridge.findNextMM( source, target, topic.getFqtn());
-					mm.addTopic(topic.getFqtn());
-					bridge.updateMirrorMaker(mm);
-				} catch ( Exception ex ) {
-					err.setCode(500);
-					err.setFields( "mirror_maker.topic");
-					err.setMessage("Unexpected condition: " + ex );
-					anythingWrong = true;
-					break;
-				}
-			}			
-
-			
-		}
-		return  anythingWrong;
-
-	}
-	
-	
-	/*
-	 * Prior to 1707, we only supported EDGE_TO_CENTRAL replication.
-	 * This was determined automatically based on presence of edge publishers and central subscribers.
-	 * The following method is a modification of that original logic, to preserve some backwards compatibility, 
-	 * i.e. to be used when no ReplicationType is specified.
-	 */
-	public ReplicationType reviewTopic( Topic topic ) {
-	
-		
-		if ( topic.getNumClients() > 1 ) {
-			Graph graph = new Graph( topic.getClients(), false );
-			
-			String centralFqdn = new String();
-			if ( graph.hasCentral() ) {
-				DmaapConfig p = (DmaapConfig)DmaapConfig.getConfig();
-				centralFqdn = p.getProperty("MR.CentralCname");
-			}
-
-			Collection<String> locations = graph.getKeys();
-			for( String loc : locations ) {
-				logger.info( "loc=" + loc );
-				MR_Cluster cluster = clusters.getMr_ClusterByLoc(loc);
-				if ( cluster == null ) {
-					logger.info( "No MR cluster for location " + loc );
-					continue;
-				}
-				if ( graph.hasCentral() &&  ! graph.getCentralLoc().equals(cluster.getDcaeLocationName())) {
-					logger.info( "Detected case for EDGE_TO_CENTRAL from " + cluster.getFqdn() + " to " + centralFqdn );
-					return ReplicationType.REPLICATION_EDGE_TO_CENTRAL;
-					
-				}
-				
-			}
-		}
-	
-		return ReplicationType.REPLICATION_NONE;
-	}
+
+    // REF: https://wiki.web.att.com/pages/viewpage.action?pageId=519703122
+    private static String defaultGlobalMrHost;
+
+    private Map<String, Topic> mr_topics;
+
+    private static DmaapService dmaapSvc = new DmaapService();
+    private MR_ClientService clientService;
+    private MR_ClusterService clusters;
+    private DcaeLocationService locations;
+    private MirrorMakerService bridge;
+
+    private static String centralCname;
+    private static boolean createTopicRoles;
+    private boolean strictGraph = true;
+    private boolean mmPerMR;
+
+
+    public TopicService() {
+        this(DatabaseClass.getTopics(), new MR_ClientService(), (DmaapConfig) DmaapConfig.getConfig(),
+                new MR_ClusterService(), new DcaeLocationService(), new MirrorMakerService());
+
+    }
+
+    TopicService(Map<String, Topic> mr_topics, MR_ClientService clientService, DmaapConfig p,
+                 MR_ClusterService clusters, DcaeLocationService locations, MirrorMakerService bridge) {
+        this.mr_topics = mr_topics;
+        this.clientService = clientService;
+        defaultGlobalMrHost = p.getProperty("MR.globalHost", "global.host.not.set");
+        centralCname = p.getProperty("MR.CentralCname");
+        createTopicRoles = "true".equalsIgnoreCase(p.getProperty("aaf.CreateTopicRoles", "true"));
+        String unit_test = p.getProperty("UnitTest", "No");
+        if ("Yes".equals(unit_test)) {
+            strictGraph = false;
+        }
+        mmPerMR = "true".equalsIgnoreCase(p.getProperty("MirrorMakerPerMR", "true"));
+        logger.info("TopicService properties: CentralCname=" + centralCname +
+                "   defaultGlobarlMrHost=" + defaultGlobalMrHost +
+                " createTopicRoles=" + createTopicRoles +
+                " mmPerMR=" + mmPerMR);
+        this.clusters = clusters;
+        this.locations = locations;
+        this.bridge = bridge;
+    }
+
+    public Map<String, Topic> getTopics() {
+        return mr_topics;
+    }
+
+    public List<Topic> getAllTopics() {
+        return getAllTopics(true);
+    }
+
+    public List<Topic> getAllTopicsWithoutClients() {
+        return getAllTopics(false);
+    }
+
+    private List<Topic> getAllTopics(Boolean withClients) {
+        ArrayList<Topic> topics = new ArrayList<>(mr_topics.values());
+        if (withClients) {
+            for (Topic topic : topics) {
+                topic.setClients(clientService.getAllMrClients(topic.getFqtn()));
+            }
+        }
+        return topics;
+    }
+
+
+    public Topic getTopic(String key, ApiError apiError) {
+        logger.info("getTopic: key=" + key);
+        Topic t = mr_topics.get(key);
+        if (t == null) {
+            apiError.setCode(Status.NOT_FOUND.getStatusCode());
+            apiError.setFields("fqtn");
+            apiError.setMessage("topic with fqtn " + key + " not found");
+            return null;
+        }
+        t.setClients(clientService.getAllMrClients(key));
+        apiError.setCode(Status.OK.getStatusCode());
+        return t;
+    }
+
+    private void aafTopicSetup(Topic topic, ApiError err) {
+
+        String nsr = dmaapSvc.getDmaap().getTopicNsRoot();
+        if (nsr == null) {
+            err.setCode(500);
+            err.setMessage("Unable to establish AAF namespace root: (check /dmaap object)");
+            err.setFields("topicNsRoot");
+            return;
+        }
+
+        // establish AAF Connection using TopicMgr identity
+        AafService aaf = new AafService(ServiceType.AAF_TopicMgr);
+
+        AafRole pubRole = null;
+        AafRole subRole = null;
+
+        // creating Topic Roles was not an original feature.
+        // For backwards compatibility, only do this if the feature is enabled.
+        // Also, if the namespace of the topic is a foreign namespace, (i.e. not the same as our root ns)
+        // then we likely don't have permission to create sub-ns and Roles so don't try.
+        if (createTopicRoles && topic.getFqtn().startsWith(nsr)) {
+            // create AAF namespace for this topic
+            AafNamespace ns = new AafNamespace(topic.getFqtn(), aaf.getIdentity());
+            {
+                int rc = aaf.addNamespace(ns);
+                if (rc != 201 && rc != 409) {
+                    err.setCode(500);
+                    err.setMessage("Unexpected response from AAF:" + rc);
+                    err.setFields("namespace:" + topic.getFqtn() + " identity=" + aaf.getIdentity());
+                    return;
+                }
+            }
+
+            // create AAF Roles for MR clients of this topic
+            String rn = "publisher";
+            pubRole = new AafRole(topic.getFqtn(), rn);
+            int rc = aaf.addRole(pubRole);
+            if (rc != 201 && rc != 409) {
+                err.setCode(500);
+                err.setMessage("Unexpected response from AAF:" + rc);
+                err.setFields("topic:" + topic.getFqtn() + " role=" + rn);
+                return;
+            }
+            topic.setPublisherRole(pubRole.getFullyQualifiedRole());
+
+            rn = "subscriber";
+            subRole = new AafRole(topic.getFqtn(), rn);
+            rc = aaf.addRole(subRole);
+            if (rc != 201 && rc != 409) {
+                err.setCode(500);
+                err.setMessage("Unexpected response from AAF:" + rc);
+                err.setFields("topic:" + topic.getFqtn() + " role=" + rn);
+                return;
+            }
+            topic.setSubscriberRole(subRole.getFullyQualifiedRole());
+        }
+
+        // create AAF perms checked by MR
+        String instance = ":topic." + topic.getFqtn();
+        String[] actions = {"pub", "sub", "view"};
+        String t = dmaapSvc.getTopicPerm();
+        for (String action : actions) {
+            DmaapPerm perm = new DmaapPerm(t, instance, action);
+            int rc = aaf.addPerm(perm);
+            if (rc != 201 && rc != 409) {
+                err.setCode(500);
+                err.setMessage("Unexpected response from AAF:" + rc);
+                err.setFields("t=" + t + " instance=" + instance + " action=" + action);
+                return;
+            }
+            if (createTopicRoles) {
+                // Grant perms to our default Roles
+                if (action.equals("pub") || action.equals("view")) {
+                    DmaapGrant g = new DmaapGrant(perm, pubRole.getFullyQualifiedRole());
+                    rc = aaf.addGrant(g);
+                    if (rc != 201 && rc != 409) {
+                        err.setCode(rc);
+                        err.setMessage("Grant of " + perm.toString() + " failed for " + pubRole.getFullyQualifiedRole());
+                        logger.warn(err.getMessage());
+                        return;
+                    }
+                }
+                if (action.equals("sub") || action.equals("view")) {
+                    DmaapGrant g = new DmaapGrant(perm, subRole.getFullyQualifiedRole());
+                    rc = aaf.addGrant(g);
+                    if (rc != 201 && rc != 409) {
+                        err.setCode(rc);
+                        err.setMessage("Grant of " + perm.toString() + " failed for " + subRole.getFullyQualifiedRole());
+                        logger.warn(err.getMessage());
+                        return;
+                    }
+                }
+            }
+
+        }
+    }
+
+    public Topic addTopic(Topic topic, ApiError err, Boolean useExisting) {
+        logger.info("Entry: addTopic");
+        logger.info("Topic name=" + topic.getTopicName() + " fqtnStyle=" + topic.getFqtnStyle());
+        String nFqtn = topic.genFqtn();
+        logger.info("FQTN=" + nFqtn);
+        Topic pTopic = getTopic(nFqtn, err);
+        if (pTopic != null) {
+            String t = "topic already exists: " + nFqtn;
+            logger.info(t);
+            if (useExisting) {
+                err.setCode(Status.OK.getStatusCode());
+                return pTopic;
+            }
+            err.setMessage(t);
+            err.setFields("fqtn");
+            err.setCode(Status.CONFLICT.getStatusCode());
+            return null;
+        }
+        err.reset();  // err filled with NOT_FOUND is expected case, but don't want to litter...
+
+        topic.setFqtn(nFqtn);
+
+        aafTopicSetup(topic, err);
+        if (err.getCode() >= 400) {
+            return null;
+        }
+
+        if (topic.getReplicationCase().involvesGlobal()) {
+            if (topic.getGlobalMrURL() == null) {
+                topic.setGlobalMrURL(defaultGlobalMrHost);
+            }
+            if (!Fqdn.isValid(topic.getGlobalMrURL())) {
+                logger.error("GlobalMR FQDN not valid: " + topic.getGlobalMrURL());
+                topic.setStatus(DmaapObject_Status.INVALID);
+                err.setCode(500);
+                err.setMessage("Value is not a valid FQDN:" + topic.getGlobalMrURL());
+                err.setFields("globalMrURL");
+
+                return null;
+            }
+        }
+
+
+        if (topic.getNumClients() > 0) {
+            ArrayList<MR_Client> clients = new ArrayList<MR_Client>(topic.getClients());
+
+
+            ArrayList<MR_Client> clients2 = new ArrayList<MR_Client>();
+            for (Iterator<MR_Client> it = clients.iterator(); it.hasNext(); ) {
+                MR_Client c = it.next();
+
+                logger.info("c fqtn=" + c.getFqtn() + " ID=" + c.getMrClientId() + " url=" + c.getTopicURL());
+                MR_Client nc = new MR_Client(c.getDcaeLocationName(), topic.getFqtn(), c.getClientRole(), c.getAction());
+                nc.setFqtn(topic.getFqtn());
+                nc.setClientIdentity(c.getClientIdentity());
+                logger.info("nc fqtn=" + nc.getFqtn() + " ID=" + nc.getMrClientId() + " url=" + nc.getTopicURL());
+                clients2.add(clientService.addMr_Client(nc, topic, err));
+                if (!err.is2xx()) {
+                    return null;
+                }
+            }
+
+            topic.setClients(clients2);
+        }
+
+        Topic ntopic = checkForBridge(topic, err);
+        if (ntopic == null) {
+            topic.setStatus(DmaapObject_Status.INVALID);
+            if (!err.is2xx()) {
+                return null;
+            }
+        }
+
+
+        mr_topics.put(nFqtn, ntopic);
+
+        err.setCode(Status.OK.getStatusCode());
+        return ntopic;
+    }
+
+
+    public Topic updateTopic(Topic topic, ApiError err) {
+        logger.info("updateTopic: entry");
+        logger.info("updateTopic: topic=" + topic);
+        logger.info("updateTopic: fqtn=" + topic.getFqtn());
+        if (topic.getFqtn().isEmpty()) {
+            return null;
+        }
+        logger.info("updateTopic: call checkForBridge");
+        Topic ntopic = checkForBridge(topic, err);
+        if (ntopic == null) {
+            topic.setStatus(DmaapObject_Status.INVALID);
+            if (!err.is2xx()) {
+                return null;
+            }
+        }
+        if (ntopic != null) {
+            logger.info("updateTopic: call put");
+            mr_topics.put(ntopic.getFqtn(), ntopic);
+        }
+        err.setCode(Status.OK.getStatusCode());
+        return ntopic;
+    }
+
+    public Topic removeTopic(String pubId, ApiError apiError) {
+        Topic topic = mr_topics.get(pubId);
+        if (topic == null) {
+            apiError.setCode(Status.NOT_FOUND.getStatusCode());
+            apiError.setMessage("Topic " + pubId + " does not exist");
+            apiError.setFields("fqtn");
+            return null;
+        }
+        ArrayList<MR_Client> clients = new ArrayList<MR_Client>(clientService.getAllMrClients(pubId));
+        for (Iterator<MR_Client> it = clients.iterator(); it.hasNext(); ) {
+            MR_Client c = it.next();
+
+
+            clientService.removeMr_Client(c.getMrClientId(), false, apiError);
+            if (!apiError.is2xx()) {
+                return null;
+            }
+        }
+        apiError.setCode(Status.OK.getStatusCode());
+        return mr_topics.remove(pubId);
+    }
+
+    public static ApiError setBridgeClientPerms(MR_Cluster node) {
+        DmaapConfig p = (DmaapConfig) DmaapConfig.getConfig();
+        String mmProvRole = p.getProperty("MM.ProvRole");
+        String mmAgentRole = p.getProperty("MM.AgentRole");
+        String[] Roles = {mmProvRole, mmAgentRole};
+        String[] actions = {"view", "pub", "sub"};
+        Topic bridgeAdminTopic = new Topic().init();
+        bridgeAdminTopic.setTopicName(dmaapSvc.getBridgeAdminFqtn());
+        bridgeAdminTopic.setTopicDescription("RESERVED topic for MirroMaker Provisioning");
+        bridgeAdminTopic.setOwner("DBC");
+
+        ArrayList<MR_Client> clients = new ArrayList<MR_Client>();
+        for (String role : Roles) {
+            MR_Client client = new MR_Client();
+            client.setAction(actions);
+            client.setClientRole(role);
+            client.setDcaeLocationName(node.getDcaeLocationName());
+            clients.add(client);
+        }
+        bridgeAdminTopic.setClients(clients);
+
+        TopicService ts = new TopicService();
+        ApiError err = new ApiError();
+        ts.addTopic(bridgeAdminTopic, err, true);
+
+        if (err.is2xx() || err.getCode() == 409) {
+            err.setCode(200);
+            return err;
+        }
+
+        errorLogger.error(DmaapbcLogMessageEnum.TOPIC_CREATE_ERROR, bridgeAdminTopic.getFqtn(), Integer.toString(err.getCode()), err.getFields(), err.getMessage());
+        return err;
+    }
+
+
+    public Topic checkForBridge(Topic topic, ApiError err) {
+        logger.info("checkForBridge: entry");
+        logger.info("fqtn=" + topic.getFqtn() + " replicatonType=" + topic.getReplicationCase());
+        if (topic.getReplicationCase() == ReplicationType.REPLICATION_NONE) {
+            topic.setStatus(DmaapObject_Status.VALID);
+            return topic;
+        }
+
+        boolean anythingWrong = false;
+
+        Set<String> groups = clusters.getGroups();
+        for (String g : groups) {
+            logger.info("buildBridge for " + topic.getFqtn() + " on group" + g);
+            anythingWrong |= buildBridge(topic, err, g);
+        }
+        if (anythingWrong) {
+            topic.setStatus(DmaapObject_Status.INVALID);
+            if (!err.is2xx()) {
+                return null;
+            }
+        } else {
+            topic.setStatus(DmaapObject_Status.VALID);
+        }
+        return topic;
+    }
+
+    private boolean buildBridge(Topic topic, ApiError err, String group) {
+        logger.info("buildBridge: entry");
+        boolean anythingWrong = false;
+        Graph graph;
+        logger.info("buildBridge: strictGraph=" + strictGraph);
+        if (group == null || group.isEmpty()) {
+            graph = new Graph(topic.getClients(), strictGraph);
+        } else {
+            graph = new Graph(topic.getClients(), strictGraph, group);
+        }
+        logger.info("buildBridge: graph=" + graph);
+        MR_Cluster groupCentralCluster = null;
+
+
+        if (graph.isEmpty()) {
+            logger.info("buildBridge: graph is empty.  return false");
+            return false;
+        } else if (group == null && topic.getReplicationCase().involvesFQDN()) {
+            logger.info("buildBridge: group is null and replicationCaseInvolvesFQDN. return false");
+            return false;
+        } else if (!graph.hasCentral()) {
+            logger.warn("Topic " + topic.getFqtn() + " wants to be " + topic.getReplicationCase() + " but has no central clients");
+            return true;
+        } else {
+            groupCentralCluster = clusters.getMr_ClusterByLoc(graph.getCentralLoc());
+        }
+        Collection<String> clientLocations = graph.getKeys();
+        for (String loc : clientLocations) {
+            logger.info("loc=" + loc);
+            DcaeLocation location = locations.getDcaeLocation(loc);
+            MR_Cluster cluster = clusters.getMr_ClusterByLoc(loc);
+            logger.info("cluster=" + cluster + " at " + cluster.getDcaeLocationName());
+            logger.info("location.isCentral()=" + location.isCentral() + " getCentralLoc()=" + graph.getCentralLoc());
+
+
+            String source = null;
+            String target = null;
+
+            /*
+             * Provision Edge to Central bridges...
+             */
+            if (!location.isCentral() && !graph.getCentralLoc().equals(cluster.getDcaeLocationName())) {
+                switch (topic.getReplicationCase()) {
+                    case REPLICATION_EDGE_TO_CENTRAL:
+                    case REPLICATION_EDGE_TO_CENTRAL_TO_GLOBAL:  // NOTE: this is for E2C portion only
+                        source = cluster.getFqdn();
+                        target = (mmPerMR) ? groupCentralCluster.getFqdn() : centralCname;
+                        logger.info("REPLICATION_EDGE_TO_CENTRAL: source=" + source + " target=" + target);
+                        break;
+                    case REPLICATION_CENTRAL_TO_EDGE:
+                    case REPLICATION_GLOBAL_TO_CENTRAL_TO_EDGE:  // NOTE: this is for C2E portion only
+                        source = (mmPerMR) ? groupCentralCluster.getFqdn() : centralCname;
+                        target = cluster.getFqdn();
+                        break;
+                    case REPLICATION_CENTRAL_TO_GLOBAL:
+                    case REPLICATION_GLOBAL_TO_CENTRAL:
+                    case REPLICATION_FQDN_TO_GLOBAL:
+                    case REPLICATION_GLOBAL_TO_FQDN:
+                        break;
+
+                    case REPLICATION_EDGE_TO_FQDN:
+                    case REPLICATION_EDGE_TO_FQDN_TO_GLOBAL:  // NOTE: this is for E2C portion only
+                        source = cluster.getFqdn();
+                        target = groupCentralCluster.getFqdn();
+                        break;
+                    case REPLICATION_FQDN_TO_EDGE:
+                    case REPLICATION_GLOBAL_TO_FQDN_TO_EDGE:  // NOTE: this is for F2E portion only
+                        source = groupCentralCluster.getFqdn();
+                        target = cluster.getFqdn();
+                        break;
+
+                    default:
+                        logger.error("Unexpected value for ReplicationType (" + topic.getReplicationCase() + ") for topic " + topic.getFqtn());
+                        anythingWrong = true;
+                        err.setCode(400);
+                        err.setFields("topic=" + topic.genFqtn() + " replicationCase="
+                                + topic.getReplicationCase());
+                        err.setMessage("Unexpected value for ReplicationType");
+                        continue;
+                }
+
+            } else if (location.isCentral() && graph.getCentralLoc().equals(cluster.getDcaeLocationName())) {
+                /*
+                 * Provision Central to Global bridges
+                 */
+                switch (topic.getReplicationCase()) {
+
+                    case REPLICATION_CENTRAL_TO_GLOBAL:
+                    case REPLICATION_EDGE_TO_CENTRAL_TO_GLOBAL:
+                        source = centralCname;
+                        target = topic.getGlobalMrURL();
+                        break;
+                    case REPLICATION_GLOBAL_TO_CENTRAL:
+                    case REPLICATION_GLOBAL_TO_CENTRAL_TO_EDGE:  // NOTE: this is for G2C portion only
+                        source = topic.getGlobalMrURL();
+                        target = centralCname;
+                        break;
+
+                    case REPLICATION_EDGE_TO_FQDN_TO_GLOBAL:  // NOTE: this is for E2F portion only
+                        source = groupCentralCluster.getFqdn();
+                        target = topic.getGlobalMrURL();
+                        break;
+
+                    case REPLICATION_FQDN_TO_GLOBAL:
+                        source = groupCentralCluster.getFqdn();
+                        target = topic.getGlobalMrURL();
+                        break;
+
+                    case REPLICATION_GLOBAL_TO_FQDN:
+                    case REPLICATION_GLOBAL_TO_FQDN_TO_EDGE:  // NOTE: this is for G2F portion only
+                        source = topic.getGlobalMrURL();
+                        target = groupCentralCluster.getFqdn();
+                        break;
+
+                    case REPLICATION_FQDN_TO_EDGE:
+                    case REPLICATION_EDGE_TO_FQDN:
+                    case REPLICATION_EDGE_TO_CENTRAL:
+                    case REPLICATION_CENTRAL_TO_EDGE:
+                        break;
+                    default:
+                        logger.error("Unexpected value for ReplicationType (" + topic.getReplicationCase() + ") for topic " + topic.getFqtn());
+                        anythingWrong = true;
+                        err.setCode(400);
+                        err.setFields("topic=" + topic.genFqtn() + " replicationCase="
+                                + topic.getReplicationCase());
+                        err.setMessage("Unexpected value for ReplicationType");
+                        continue;
+                }
+            } else {
+                logger.warn("dcaeLocation " + loc + " is neither Edge nor Central so no mmagent provisioning was done");
+                anythingWrong = true;
+                continue;
+            }
+            if (source != null && target != null) {
+                try {
+                    logger.info("Create a MM from " + source + " to " + target);
+                    MirrorMaker mm = bridge.findNextMM(source, target, topic.getFqtn());
+                    mm.addTopic(topic.getFqtn());
+                    bridge.updateMirrorMaker(mm);
+                } catch (Exception ex) {
+                    err.setCode(500);
+                    err.setFields("mirror_maker.topic");
+                    err.setMessage("Unexpected condition: " + ex);
+                    anythingWrong = true;
+                    break;
+                }
+            }
+
+
+        }
+        return anythingWrong;
+
+    }
+
+
+    /*
+     * Prior to 1707, we only supported EDGE_TO_CENTRAL replication.
+     * This was determined automatically based on presence of edge publishers and central subscribers.
+     * The following method is a modification of that original logic, to preserve some backwards compatibility,
+     * i.e. to be used when no ReplicationType is specified.
+     */
+    public ReplicationType reviewTopic(Topic topic) {
+
+
+        if (topic.getNumClients() > 1) {
+            Graph graph = new Graph(topic.getClients(), false);
+
+            String centralFqdn = new String();
+            if (graph.hasCentral()) {
+                DmaapConfig p = (DmaapConfig) DmaapConfig.getConfig();
+                centralFqdn = p.getProperty("MR.CentralCname");
+            }
+
+            Collection<String> locations = graph.getKeys();
+            for (String loc : locations) {
+                logger.info("loc=" + loc);
+                MR_Cluster cluster = clusters.getMr_ClusterByLoc(loc);
+                if (cluster == null) {
+                    logger.info("No MR cluster for location " + loc);
+                    continue;
+                }
+                if (graph.hasCentral() && !graph.getCentralLoc().equals(cluster.getDcaeLocationName())) {
+                    logger.info("Detected case for EDGE_TO_CENTRAL from " + cluster.getFqdn() + " to " + centralFqdn);
+                    return ReplicationType.REPLICATION_EDGE_TO_CENTRAL;
+
+                }
+
+            }
+        }
+
+        return ReplicationType.REPLICATION_NONE;
+    }
 
 }