From d3994fb85f92d429b626013a0bd9f2ff69418c95 Mon Sep 17 00:00:00 2001 From: pkaras Date: Fri, 31 May 2019 10:38:08 +0200 Subject: code formatting by standard rules Change-Id: I24dc78e826abf1e0cfeccfbd20d61c6723a0c519 Issue-ID: DMAAP-1211 Signed-off-by: piotr.karas --- .../dmaap/dbcapi/service/MR_ClientService.java | 534 +++++---- .../onap/dmaap/dbcapi/service/TopicService.java | 1139 ++++++++++---------- 2 files changed, 831 insertions(+), 842 deletions(-) (limited to 'src/main/java/org') 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 mr_clients = DatabaseClass.getMr_clients(); + private Map clusters = DatabaseClass.getMr_clusters(); + private Map 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 getAllMr_Clients() { + return new ArrayList<>(mr_clients.values()); + } + List getAllMrClients(String fqtn) { + ArrayList results = new ArrayList<>(); + for (Map.Entry entry : mr_clients.entrySet()) { + MR_Client client = entry.getValue(); + if (fqtn.equals(client.getFqtn())) { + results.add(client); + } + } + return results; + } + List getClientsByLocation(String location) { + List results = new ArrayList<>(); + for (Map.Entry 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 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 mr_clients = DatabaseClass.getMr_clients(); - private Map clusters = DatabaseClass.getMr_clusters(); - private Map 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 getAllMr_Clients() { - return new ArrayList<>(mr_clients.values()); - } - - List getAllMrClients(String fqtn) { - ArrayList results = new ArrayList<>(); - for (Map.Entry entry : mr_clients.entrySet()) - { - MR_Client client = entry.getValue(); - if ( fqtn.equals(client.getFqtn() ) ) { - results.add( client ); - } - } - return results; - } - - List getClientsByLocation(String location) { - List results = new ArrayList<>(); - for (Map.Entry 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 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 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 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 getTopics() { - return mr_topics; - } - - public List getAllTopics() { - return getAllTopics( true ); - } - public List getAllTopicsWithoutClients() { - return getAllTopics(false); - } - - private List getAllTopics( Boolean withClients ) { - ArrayList 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 clients = new ArrayList(topic.getClients()); - - - ArrayList clients2 = new ArrayList(); - for ( Iterator 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 clients = new ArrayList(clientService.getAllMrClients( pubId )); - for ( Iterator 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 clients = new ArrayList(); - 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 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 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 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 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 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 getTopics() { + return mr_topics; + } + + public List getAllTopics() { + return getAllTopics(true); + } + + public List getAllTopicsWithoutClients() { + return getAllTopics(false); + } + + private List getAllTopics(Boolean withClients) { + ArrayList 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 clients = new ArrayList(topic.getClients()); + + + ArrayList clients2 = new ArrayList(); + for (Iterator 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 clients = new ArrayList(clientService.getAllMrClients(pubId)); + for (Iterator 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 clients = new ArrayList(); + 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 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 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 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; + } } -- cgit 1.2.3-korg