src/main/java/com/att/nsa/dmaap/service/MMRestService.java

   1 /*******************************************************************************
   2  *  ============LICENSE_START=======================================================
   3  *  org.onap.dmaap
   4  *  ================================================================================
   5  *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
   6  *  ================================================================================
   7  *  Licensed under the Apache License, Version 2.0 (the "License");
   8  *  you may not use this file except in compliance with the License.
   9  *  You may obtain a copy of the License at
  10  *        http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  *  Unless required by applicable law or agreed to in writing, software
  13  *  distributed under the License is distributed on an "AS IS" BASIS,
  14  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  *  See the License for the specific language governing permissions and
  16  *  limitations under the License.
  17  *  ============LICENSE_END=========================================================
  18  *
  19  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  20  *
  21  *******************************************************************************/
  22 package com.att.nsa.dmaap.service;
  23
  24 import java.io.IOException;
  25 import java.io.InputStream;
  26 import java.util.ArrayList;
  27 import java.util.Date;
  28 import java.util.List;
  29
  30
  31 import javax.servlet.http.HttpServletRequest;
  32 import javax.servlet.http.HttpServletResponse;
  33 import javax.ws.rs.POST;
  34 import javax.ws.rs.Path;
  35 import javax.ws.rs.Produces;
  36 import javax.ws.rs.core.Context;
  37 import org.json.JSONObject;
  38 import org.apache.commons.io.IOUtils;
  39 import org.apache.commons.lang.StringUtils;
  40 import com.att.eelf.configuration.EELFLogger;
  41 import com.att.eelf.configuration.EELFManager;
  42 import org.springframework.beans.factory.annotation.Autowired;
  43 import org.springframework.beans.factory.annotation.Qualifier;
  44 import org.springframework.stereotype.Component;
  45
  46 import com.att.nsa.cambria.utils.ConfigurationReader;
  47 import com.att.nsa.cambria.utils.DMaaPResponseBuilder;
  48 import com.att.nsa.cambria.utils.Utils;
  49 import com.att.nsa.configs.ConfigDbException;
  50 import com.att.nsa.dmaap.mmagent.*;
  51 import com.att.nsa.drumlin.till.nv.rrNvReadable.missingReqdSetting;
  52 import com.att.nsa.security.ReadWriteSecuredResource.AccessDeniedException;
  53 import com.google.gson.Gson;
  54 import com.google.gson.JsonSyntaxException;
  55
  56 import edu.emory.mathcs.backport.java.util.Arrays;
  57
  58 import com.att.ajsc.filemonitor.AJSCPropertiesMap;
  59 import com.att.nsa.cambria.CambriaApiException;
  60 import com.att.nsa.cambria.backends.ConsumerFactory.UnavailableException;
  61
  62 import org.json.JSONArray;
  63 import org.json.JSONException;
  64 import com.att.nsa.cambria.beans.DMaaPContext;
  65 import com.att.nsa.cambria.constants.CambriaConstants;
  66 import com.att.nsa.cambria.exception.DMaaPErrorMessages;
  67 import com.att.nsa.cambria.metabroker.Broker.TopicExistsException;
  68 import com.att.nsa.cambria.security.DMaaPAAFAuthenticator;
  69 import com.att.nsa.cambria.security.DMaaPAAFAuthenticatorImpl;
  70 import com.att.nsa.cambria.service.MMService;
  71
  72 /**
  73  * Rest Service class for Mirror Maker proxy Rest Services
  74  *
  75  * @author <a href="mailto:"></a>
  76  *
  77  * @since May 25, 2016
  78  */
  79
  80 @Component
  81 public class MMRestService {
  82
  83         private static final EELFLogger LOGGER = EELFManager.getInstance().getLogger(MMRestService.class);
  84         private static final String NO_ADMIN_PERMISSION = "No Mirror Maker Admin permission.";
  85         private static final String NO_USER_PERMISSION = "No Mirror Maker User permission.";
  86         private static final String NO_USER_CREATE_PERMISSION = "No Mirror Maker User Create permission.";
  87         private static final String NAME_DOES_NOT_MEET_REQUIREMENT = "Mirror Maker name can only contain alpha numeric";
  88         private static final String INVALID_IPPORT = "This is not a valid IP:Port";
  89         private static final String MIRROR_MAKERADMIN = "msgRtr.mirrormakeradmin.aaf";
  90         private static final String MIRROR_MAKERUSER = "msgRtr.mirrormakeruser.aaf";
  91         private static final String UTF_8 = "UTF-8";
  92         private static final String MESSAGE = "message";
  93         private static final String LISTMIRRORMAKER = "listMirrorMaker";
  94         private static final String ERROR = "error";
  95         private static final String NAMESPACE = "namespace";
  96
  97         private String topic;
  98         private int timeout;
  99         private String consumergroup;
 100         private String consumerid;
 101
 102         @Autowired
 103         @Qualifier("configurationReader")
 104         private ConfigurationReader configReader;
 105
 106         @Context
 107         private HttpServletRequest request;
 108
 109         @Context
 110         private HttpServletResponse response;
 111
 112         @Autowired
 113         private MMService mirrorService;
 114
 115         @Autowired
 116         private DMaaPErrorMessages errorMessages;
 117
 118         /**
 119          * This method is used for taking Configuration Object,HttpServletRequest
 120          * Object,HttpServletRequest HttpServletResponse Object,HttpServletSession
 121          * Object.
 122          *
 123          * @return DMaaPContext object from where user can get Configuration
 124          *         Object,HttpServlet Object
 125          *
 126          */
 127         private DMaaPContext getDmaapContext() {
 128                 DMaaPContext dmaapContext = new DMaaPContext();
 129                 dmaapContext.setRequest(request);
 130                 dmaapContext.setResponse(response);
 131                 dmaapContext.setConfigReader(configReader);
 132                 dmaapContext.setConsumerRequestTime(Utils.getFormattedDate(new Date()));
 133
 134                 return dmaapContext;
 135         }
 136
 137         @POST
 138         @Produces("application/json")
 139         @Path("/create")
 140         public void callCreateMirrorMaker(InputStream msg) {
 141
 142                 DMaaPContext ctx = getDmaapContext();
 143                 if (checkMirrorMakerPermission(ctx,
 144                                 AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, MIRROR_MAKERADMIN))) {
 145
 146                         loadProperty();
 147                         String input = null;
 148                         String randomStr = getRandomNum();
 149
 150                         InputStream inStream = null;
 151                         Gson gson = new Gson();
 152                         CreateMirrorMaker createMirrorMaker = new CreateMirrorMaker();
 153
 154                         try {
 155                                 input = IOUtils.toString(msg, UTF_8);
 156
 157                                 if (input != null && input.length() > 0) {
 158                                         input = removeExtraChar(input);
 159                                 }
 160
 161                                 // Check if the request has CreateMirrorMaker
 162                                 try {
 163                                         createMirrorMaker = gson.fromJson(input, CreateMirrorMaker.class);
 164
 165                                 } catch (JsonSyntaxException ex) {
 166
 167                                         sendErrResponse(ctx, errorMessages.getIncorrectJson());
 168                                         LOGGER.error("JsonSyntaxException: ", ex);
 169                                 }
 170                                 String name = createMirrorMaker.getCreateMirrorMaker().getName();
 171                                 // send error message if it is not a CreateMirrorMaker request.
 172                                 if (createMirrorMaker.getCreateMirrorMaker() == null) {
 173                                         sendErrResponse(ctx, "This is not a CreateMirrorMaker request. Please try again.");
 174                                 }
 175
 176                                 // MirrorMaker whitelist and status should not be passed
 177                                 else if (createMirrorMaker.getCreateMirrorMaker().getWhitelist() != null
 178                                                 || createMirrorMaker.getCreateMirrorMaker().getStatus() != null) {
 179                                         sendErrResponse(ctx, "This is not a CreateMirrorMaker request. Please try again.");
 180                                 }
 181
 182                                 // if empty, blank name is entered
 183                                 else if (StringUtils.isBlank(name)) {
 184                                         sendErrResponse(ctx, "Name can not be empty or blank.");
 185                                 }
 186
 187                                 // Check if the name contains only Alpha Numeric
 188                                 else if (!isAlphaNumeric(name)) {
 189                                         sendErrResponse(ctx, NAME_DOES_NOT_MEET_REQUIREMENT);
 190
 191                                 }
 192
 193                                 // Validate the IP and Port
 194                                 else if (!StringUtils.isBlank(createMirrorMaker.getCreateMirrorMaker().getConsumer())
 195                                                 && !StringUtils.isBlank(createMirrorMaker.getCreateMirrorMaker().getProducer())
 196                                                 && !validateIPPort(createMirrorMaker.getCreateMirrorMaker().getConsumer())
 197                                                 || !validateIPPort(createMirrorMaker.getCreateMirrorMaker().getProducer())) {
 198                                         sendErrResponse(ctx, INVALID_IPPORT);
 199
 200                                 }
 201                                 // Set a random number as messageID, convert Json Object to
 202                                 // InputStream and finally call publisher and subscriber
 203                                 else if (isAlphaNumeric(name) && validateIPPort(createMirrorMaker.getCreateMirrorMaker().getConsumer())
 204                                                 && validateIPPort(createMirrorMaker.getCreateMirrorMaker().getProducer())) {
 205
 206                                         createMirrorMaker.setMessageID(randomStr);
 207                                         inStream = IOUtils.toInputStream(gson.toJson(createMirrorMaker), UTF_8);
 208                                         callPubSub(randomStr, ctx, inStream);
 209                                 }
 210
 211                         } catch (IOException e) {
 212
 213                                 LOGGER.error("IOException: ", e);
 214                         }
 215                 }
 216                 // Send error response if user does not provide Authorization
 217                 else {
 218                         sendErrResponse(ctx, NO_ADMIN_PERMISSION);
 219                 }
 220         }
 221
 222         @POST
 223         @Produces("application/json")
 224         @Path("/listall")
 225         public void callListAllMirrorMaker(InputStream msg) throws CambriaApiException {
 226                 DMaaPContext ctx = getDmaapContext();
 227
 228                 if (checkMirrorMakerPermission(ctx,
 229                                 AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, MIRROR_MAKERADMIN))) {
 230
 231                         loadProperty();
 232
 233                         String input = null;
 234
 235                         try {
 236                                 input = IOUtils.toString(msg, UTF_8);
 237
 238                                 if (input != null && input.length() > 0) {
 239                                         input = removeExtraChar(input);
 240                                 }
 241
 242                                 String randomStr = getRandomNum();
 243                                 JSONObject jsonOb = null;
 244
 245                                 try {
 246                                         jsonOb = new JSONObject(input);
 247
 248                                 } catch (JSONException ex) {
 249
 250                                         sendErrResponse(ctx, errorMessages.getIncorrectJson());
 251                                         LOGGER.error("JSONException: ", ex);
 252                                 }
 253
 254                                 // Check if request has listAllMirrorMaker and
 255                                 // listAllMirrorMaker is empty
 256                                 if ((jsonOb != null) && (jsonOb.has("listAllMirrorMaker") &&
 257                                     jsonOb.getJSONObject("listAllMirrorMaker").length() == 0)) {
 258                                         jsonOb.put("messageID", randomStr);
 259                                         InputStream inStream = null;
 260
 261                                         try {
 262                                                 inStream = IOUtils.toInputStream(jsonOb.toString(), UTF_8);
 263
 264                                         } catch (IOException ioe) {
 265                                                 ioe.printStackTrace();
 266                                                 LOGGER.error("IOException: ", ioe);
 267                                         }
 268
 269                                         callPubSub(randomStr, ctx, inStream);
 270
 271                                 } else {
 272
 273                                         sendErrResponse(ctx, "This is not a ListAllMirrorMaker request. Please try again.");
 274                                 }
 275
 276                         } catch (IOException ioe) {
 277
 278                                 ioe.printStackTrace();
 279                                 LOGGER.error("IOException: ", ioe);
 280                         }
 281
 282                 } else {
 283
 284                         sendErrResponse(getDmaapContext(), NO_ADMIN_PERMISSION);
 285                 }
 286         }
 287
 288         @POST
 289         @Produces("application/json")
 290         @Path("/update")
 291         public void callUpdateMirrorMaker(InputStream msg) throws CambriaApiException {
 292
 293                 DMaaPContext ctx = getDmaapContext();
 294                 if (checkMirrorMakerPermission(ctx,
 295                                 AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, MIRROR_MAKERADMIN))) {
 296
 297                         loadProperty();
 298                         String input = null;
 299                         String randomStr = getRandomNum();
 300
 301                         InputStream inStream = null;
 302                         Gson gson = new Gson();
 303                         UpdateMirrorMaker updateMirrorMaker = new UpdateMirrorMaker();
 304
 305                         try {
 306                                 input = IOUtils.toString(msg, UTF_8);
 307
 308                                 if (input != null && input.length() > 0) {
 309                                         input = removeExtraChar(input);
 310                                 }
 311
 312                                 // Check if the request has UpdateMirrorMaker
 313                                 try {
 314                                         updateMirrorMaker = gson.fromJson(input, UpdateMirrorMaker.class);
 315
 316                                 } catch (JsonSyntaxException ex) {
 317
 318                                         sendErrResponse(ctx, errorMessages.getIncorrectJson());
 319                                         LOGGER.error("JsonSyntaxException: ", ex);
 320
 321                                 }
 322                                 String name = updateMirrorMaker.getUpdateMirrorMaker().getName();
 323
 324                                 // send error message if it is not a UpdateMirrorMaker request.
 325                                 if (updateMirrorMaker.getUpdateMirrorMaker() == null) {
 326                                         sendErrResponse(ctx, "This is not a UpdateMirrorMaker request. Please try again.");
 327                                 }
 328
 329                                 // MirrorMaker whitelist and status should not be passed
 330                                 else if (updateMirrorMaker.getUpdateMirrorMaker().getWhitelist() != null
 331                                                 || updateMirrorMaker.getUpdateMirrorMaker().getStatus() != null) {
 332                                         sendErrResponse(ctx, "This is not a UpdateMirrorMaker request. Please try again.");
 333                                 }
 334
 335                                 // if empty, blank name is entered
 336                                 else if (StringUtils.isBlank(name)) {
 337                                         sendErrResponse(ctx, "Name can not be empty or blank.");
 338                                 }
 339
 340                                 // Check if the name contains only Alpha Numeric
 341                                 else if (!isAlphaNumeric(name)) {
 342                                         sendErrResponse(ctx, NAME_DOES_NOT_MEET_REQUIREMENT);
 343
 344                                 }
 345
 346                                 // Validate the IP and Port
 347                                 else if (!StringUtils.isBlank(updateMirrorMaker.getUpdateMirrorMaker().getConsumer())
 348                                                 && !StringUtils.isBlank(updateMirrorMaker.getUpdateMirrorMaker().getProducer())
 349                                                 && !validateIPPort(updateMirrorMaker.getUpdateMirrorMaker().getConsumer())
 350                                                 || !validateIPPort(updateMirrorMaker.getUpdateMirrorMaker().getProducer())) {
 351                                         sendErrResponse(ctx, INVALID_IPPORT);
 352
 353                                 }
 354                                 // Set a random number as messageID, convert Json Object to
 355                                 // InputStream and finally call publisher and subscriber
 356                                 else if (isAlphaNumeric(name) && validateIPPort(updateMirrorMaker.getUpdateMirrorMaker().getConsumer())
 357                                                 && validateIPPort(updateMirrorMaker.getUpdateMirrorMaker().getProducer())) {
 358
 359                                         updateMirrorMaker.setMessageID(randomStr);
 360                                         inStream = IOUtils.toInputStream(gson.toJson(updateMirrorMaker), UTF_8);
 361                                         callPubSub(randomStr, ctx, inStream);
 362                                 }
 363
 364                         } catch (IOException e) {
 365
 366                                 LOGGER.error("IOException: ", e);
 367                         }
 368                 }
 369                 // Send error response if user does not provide Authorization
 370                 else {
 371                         sendErrResponse(ctx, NO_ADMIN_PERMISSION);
 372                 }
 373         }
 374
 375         @POST
 376         @Produces("application/json")
 377         @Path("/delete")
 378         public void callDeleteMirrorMaker(InputStream msg) throws CambriaApiException {
 379                 DMaaPContext ctx = getDmaapContext();
 380
 381                 if (checkMirrorMakerPermission(ctx,
 382                                 AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, MIRROR_MAKERADMIN))) {
 383
 384                         loadProperty();
 385
 386                         String input = null;
 387
 388                         try {
 389                                 input = IOUtils.toString(msg, UTF_8);
 390
 391                                 if (input != null && input.length() > 0) {
 392                                         input = removeExtraChar(input);
 393                                 }
 394
 395                                 String randomStr = getRandomNum();
 396                                 JSONObject jsonOb = null;
 397
 398                                 try {
 399                                         jsonOb = new JSONObject(input);
 400
 401                                 } catch (JSONException ex) {
 402
 403                                         sendErrResponse(ctx, errorMessages.getIncorrectJson());
 404                                         LOGGER.error("JSONException: ", ex);
 405                                 }
 406
 407                                 // Check if request has DeleteMirrorMaker and
 408                                 // DeleteMirrorMaker has MirrorMaker object with name variable
 409                                 // and check if the name contain only alpha numeric
 410                                 if ((jsonOb != null) && (jsonOb.has("deleteMirrorMaker")
 411                                                 && jsonOb.getJSONObject("deleteMirrorMaker").length() == 1
 412                                                 && jsonOb.getJSONObject("deleteMirrorMaker").has("name")
 413                                                 && !StringUtils.isBlank(jsonOb.getJSONObject("deleteMirrorMaker").getString("name"))
 414                                                 && isAlphaNumeric(jsonOb.getJSONObject("deleteMirrorMaker").getString("name")))) {
 415
 416                                         jsonOb.put("messageID", randomStr);
 417                                         InputStream inStream = null;
 418
 419                                         try {
 420                                                 inStream = IOUtils.toInputStream(jsonOb.toString(), UTF_8);
 421
 422                                         } catch (IOException ioe) {
 423                                                 ioe.printStackTrace();
 424                                                 LOGGER.error("IOException: ", ioe);
 425                                         }
 426
 427                                         callPubSub(randomStr, ctx, inStream);
 428
 429                                 } else {
 430
 431                                         sendErrResponse(ctx, "This is not a DeleteMirrorMaker request. Please try again.");
 432                                 }
 433
 434                         } catch (IOException ioe) {
 435
 436                                 ioe.printStackTrace();
 437                                 LOGGER.error("IOException: ", ioe);
 438                         }
 439
 440                 } else {
 441
 442                         sendErrResponse(getDmaapContext(), NO_ADMIN_PERMISSION);
 443                 }
 444         }
 445
 446         private boolean isListMirrorMaker(String msg, String messageID) {
 447                 String topicmsg = msg;
 448                 topicmsg = removeExtraChar(topicmsg);
 449
 450                 JSONObject jObj;
 451                 JSONArray jArray;
 452                 boolean exist = false;
 453
 454                 if (!StringUtils.isBlank(topicmsg) && topicmsg.length() > 2) {
 455                         jArray = new JSONArray(topicmsg);
 456
 457                         for (int i = 0; i < jArray.length(); i++) {
 458                                 jObj = jArray.getJSONObject(i);
 459
 460                                 JSONObject obj = new JSONObject();
 461                                 if (jObj.has(MESSAGE)) {
 462                                         obj = jObj.getJSONObject(MESSAGE);
 463                                 }
 464                                 if (obj.has("messageID") && obj.get("messageID").equals(messageID) && obj.has(LISTMIRRORMAKER)) {
 465                                         exist = true;
 466                                         break;
 467                                 }
 468                         }
 469                 }
 470                 return exist;
 471         }
 472
 473         private void loadProperty() {
 474
 475                 this.timeout = Integer.parseInt(
 476                                 AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, "msgRtr.mirrormaker.timeout").trim());
 477                 this.topic = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, "msgRtr.mirrormaker.topic").trim();
 478                 this.consumergroup = AJSCPropertiesMap
 479                                 .getProperty(CambriaConstants.msgRtr_prop, "msgRtr.mirrormaker.consumergroup").trim();
 480                 this.consumerid = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, "msgRtr.mirrormaker.consumerid")
 481                                 .trim();
 482         }
 483
 484         private String removeExtraChar(String message) {
 485                 String str = message;
 486                 str = checkJsonFormate(str);
 487
 488                 if (str != null && str.length() > 0) {
 489                         str = str.replace("\\", "");
 490                         str = str.replace("\"{", "{");
 491                         str = str.replace("}\"", "}");
 492                 }
 493                 return str;
 494         }
 495
 496         private String getRandomNum() {
 497                 long random = Math.round(Math.random() * 89999) + 10000;
 498                 String strLong = Long.toString(random);
 499                 return strLong;
 500         }
 501
 502         private boolean isAlphaNumeric(String name) {
 503                 String pattern = "^[a-zA-Z0-9]*$";
 504                 if (name.matches(pattern)) {
 505                         return true;
 506                 }
 507                 return false;
 508         }
 509
 510         // This method validate IPv4
 511         private boolean validateIPPort(String ipPort) {
 512                 String pattern = "^([01]?\\d\\d?|2[0-4]\\d|25[0-5])\\.([01]?\\d\\d?|2[0-4]\\d|25[0-5])\\."
 513                                 + "([01]?\\d\\d?|2[0-4]\\d|25[0-5])\\.([01]?\\d\\d?|2[0-4]\\d|25[0-5]):"
 514                                 + "([1-9][0-9]{0,3}|[1-5][0-9]{4}|6[0-4][0-9]{3}|65[0-4][0-9]{2}|655[0-2][0-9]|6553[0-5])$";
 515                 if (ipPort.matches(pattern)) {
 516                         return true;
 517                 }
 518                 return false;
 519         }
 520
 521         private String checkJsonFormate(String jsonStr) {
 522
 523                 String json = jsonStr;
 524                 if (jsonStr != null && jsonStr.length() > 0 && jsonStr.startsWith("[") && !jsonStr.endsWith("]")) {
 525                         json = json + "]";
 526                 }
 527                 return json;
 528         }
 529
 530         private boolean checkMirrorMakerPermission(DMaaPContext ctx, String permission) {
 531
 532                 boolean hasPermission = false;
 533
 534                 DMaaPAAFAuthenticator aaf = new DMaaPAAFAuthenticatorImpl();
 535
 536                 if (aaf.aafAuthentication(ctx.getRequest(), permission)) {
 537                         hasPermission = true;
 538                 }
 539                 return hasPermission;
 540         }
 541
 542         private void callPubSub(String randomstr, DMaaPContext ctx, InputStream inStream) {
 543                 try {
 544                         mirrorService.pushEvents(ctx, topic, inStream, null, null);
 545                         long startTime = System.currentTimeMillis();
 546                         String msgFrmSubscribe = mirrorService.subscribe(ctx, topic, consumergroup, consumerid);
 547
 548                         while (!isListMirrorMaker(msgFrmSubscribe, randomstr)
 549                                         && (System.currentTimeMillis() - startTime) < timeout) {
 550                                 msgFrmSubscribe = mirrorService.subscribe(ctx, topic, consumergroup, consumerid);
 551                         }
 552
 553                         JSONObject jsonObj;
 554                         JSONObject finalJsonObj = new JSONObject();
 555                         JSONArray jsonArray;
 556
 557                         if (msgFrmSubscribe != null && msgFrmSubscribe.length() > 0
 558                                         && isListMirrorMaker(msgFrmSubscribe, randomstr)) {
 559                                 msgFrmSubscribe = removeExtraChar(msgFrmSubscribe);
 560                                 jsonArray = new JSONArray(msgFrmSubscribe);
 561
 562                                 for (int i = 0; i < jsonArray.length(); i++) {
 563                                         jsonObj = jsonArray.getJSONObject(i);
 564
 565                                         JSONObject obj = new JSONObject();
 566                                         if (jsonObj.has(MESSAGE)) {
 567                                                 obj = jsonObj.getJSONObject(MESSAGE);
 568                                         }
 569                                         if (obj.has("messageID") && obj.get("messageID").equals(randomstr) && obj.has(LISTMIRRORMAKER)) {
 570                                                 finalJsonObj.put(LISTMIRRORMAKER, obj.get(LISTMIRRORMAKER));
 571                                                 break;
 572                                         }
 573                                 }
 574
 575                                 DMaaPResponseBuilder.respondOk(ctx, finalJsonObj);
 576
 577                         } else {
 578
 579                                 JSONObject err = new JSONObject();
 580                                 err.append(ERROR, "listMirrorMaker is not available, please make sure MirrorMakerAgent is running");
 581                                 DMaaPResponseBuilder.respondOk(ctx, err);
 582                         }
 583
 584                 } catch (Exception e) {
 585                         e.printStackTrace();
 586                         LOGGER.error("Exception: ", e);
 587                 }
 588         }
 589
 590         private void sendErrResponse(DMaaPContext ctx, String errMsg) {
 591                 JSONObject err = new JSONObject();
 592                 err.append(ERROR, errMsg);
 593
 594                 try {
 595                         DMaaPResponseBuilder.respondOk(ctx, err);
 596                         LOGGER.error(errMsg);
 597
 598                 } catch (JSONException | IOException e) {
 599                         LOGGER.error(errMsg);
 600                 }
 601         }
 602
 603         @SuppressWarnings("unchecked")
 604         @POST
 605         @Produces("application/json")
 606         @Path("/listallwhitelist")
 607         public void listWhiteList(InputStream msg) {
 608
 609                 DMaaPContext ctx = getDmaapContext();
 610                 if (checkMirrorMakerPermission(ctx,
 611                                 AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, MIRROR_MAKERUSER))) {
 612
 613                         loadProperty();
 614                         String input = null;
 615
 616                         try {
 617                                 input = IOUtils.toString(msg, UTF_8);
 618
 619                                 if (input != null && input.length() > 0) {
 620                                         input = removeExtraChar(input);
 621                                 }
 622
 623                                 // Check if it is correct Json object
 624                                 JSONObject jsonOb = null;
 625
 626                                 try {
 627                                         jsonOb = new JSONObject(input);
 628
 629                                 } catch (JSONException ex) {
 630
 631                                         sendErrResponse(ctx, errorMessages.getIncorrectJson());
 632                                         LOGGER.error("JSONException: ", ex);
 633                                 }
 634
 635                                 // Check if the request has name and name contains only alpha
 636                                 // numeric
 637                                 // and check if the request has namespace and namespace contains
 638                                 // only alpha numeric
 639                                 if (jsonOb != null
 640                                                 && jsonOb.length() == 2 && jsonOb.has("name")
 641                                                 && !StringUtils.isBlank(jsonOb.getString("name"))
 642                                                 && isAlphaNumeric(jsonOb.getString("name")) && jsonOb.has(NAMESPACE)
 643                                                 && !StringUtils.isBlank(jsonOb.getString(NAMESPACE))) {
 644
 645                                         String permission = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,
 646                                                         "msgRtr.mirrormakeruser.aaf.create") + jsonOb.getString(NAMESPACE) + "|create";
 647
 648                                         // Check if the user have create permission for the
 649                                         // namespace
 650                                         if (checkMirrorMakerPermission(ctx, permission)) {
 651
 652                                                 JSONObject listAll = new JSONObject();
 653                                                 JSONObject emptyObject = new JSONObject();
 654
 655                                                 // Create a listAllMirrorMaker Json object
 656                                                 try {
 657                                                         listAll.put("listAllMirrorMaker", emptyObject);
 658
 659                                                 } catch (JSONException e) {
 660
 661                                                         LOGGER.error("JSONException: ", e);
 662                                                 }
 663
 664                                                 // set a random number as messageID
 665                                                 String randomStr = getRandomNum();
 666                                                 listAll.put("messageID", randomStr);
 667                                                 InputStream inStream = null;
 668
 669                                                 // convert listAll Json object to InputStream object
 670                                                 try {
 671                                                         inStream = IOUtils.toInputStream(listAll.toString(), UTF_8);
 672
 673                                                 } catch (IOException ioe) {
 674                                                         ioe.printStackTrace();
 675                                                         LOGGER.error("IOException: ", ioe);
 676                                                 }
 677                                                 // call listAllMirrorMaker
 678                                                 mirrorService.pushEvents(ctx, topic, inStream, null, null);
 679
 680                                                 // subscribe for listMirrorMaker
 681                                                 long startTime = System.currentTimeMillis();
 682                                                 String msgFrmSubscribe = mirrorService.subscribe(ctx, topic, consumergroup, consumerid);
 683
 684                                                 while (!isListMirrorMaker(msgFrmSubscribe, randomStr)
 685                                                                 && (System.currentTimeMillis() - startTime) < timeout) {
 686                                                         msgFrmSubscribe = mirrorService.subscribe(ctx, topic, consumergroup, consumerid);
 687                                                 }
 688
 689                                                 if (msgFrmSubscribe != null && msgFrmSubscribe.length() > 0
 690                                                                 && isListMirrorMaker(msgFrmSubscribe, randomStr)) {
 691
 692                                                         JSONArray listMirrorMaker;
 693                                                         listMirrorMaker = getListMirrorMaker(msgFrmSubscribe, randomStr);
 694
 695                                                         String whitelist = null;
 696                                                         for (int i = 0; i < listMirrorMaker.length(); i++) {
 697
 698                                                                 JSONObject mm;
 699                                                                 mm = listMirrorMaker.getJSONObject(i);
 700                                                                 String name = mm.getString("name");
 701
 702                                                                 if (name.equals(jsonOb.getString("name")) && mm.has("whitelist")) {
 703                                                                         whitelist = mm.getString("whitelist");
 704                                                                         break;
 705                                                                 }
 706                                                         }
 707
 708                                                         if (!StringUtils.isBlank(whitelist)) {
 709
 710                                                                 List<String> topicList = new ArrayList<>();
 711                                                                 List<String> finalTopicList = new ArrayList<>();
 712                                                                 topicList = Arrays.asList(whitelist.split(","));
 713
 714                                                                 for (String topic : topicList) {
 715                                                                         if (topic != null && !topic.equals("null")
 716                                                                                         && getNamespace(topic).equals(jsonOb.getString(NAMESPACE))) {
 717
 718                                                                                 finalTopicList.add(topic);
 719                                                                         }
 720                                                                 }
 721
 722                                                                 String topicNames = "";
 723
 724                                                                 if (!finalTopicList.isEmpty()) {
 725                                                                         topicNames = StringUtils.join(finalTopicList, ",");
 726                                                                 }
 727
 728                                                                 JSONObject listAllWhiteList = new JSONObject();
 729                                                                 listAllWhiteList.put("name", jsonOb.getString("name"));
 730                                                                 listAllWhiteList.put("whitelist", topicNames);
 731
 732                                                                 DMaaPResponseBuilder.respondOk(ctx, listAllWhiteList);
 733                                                         }
 734
 735                                                 } else {
 736
 737                                                         JSONObject err = new JSONObject();
 738                                                         err.append(ERROR,
 739                                                                         "listWhiteList is not available, please make sure MirrorMakerAgent is running");
 740                                                         DMaaPResponseBuilder.respondOk(ctx, err);
 741                                                 }
 742
 743                                         } else {
 744                                                 sendErrResponse(ctx, NO_USER_CREATE_PERMISSION);
 745                                         }
 746
 747                                 } else {
 748
 749                                         sendErrResponse(ctx, "This is not a ListAllWhitelist request. Please try again.");
 750                                 }
 751
 752                         } catch (IOException | CambriaApiException | ConfigDbException | AccessDeniedException
 753                                         | TopicExistsException | missingReqdSetting | UnavailableException e) {
 754
 755                                 LOGGER.error("IOException: ", e);
 756                         }
 757                 } else {
 758                         sendErrResponse(ctx, NO_USER_PERMISSION);
 759                 }
 760         }
 761
 762         @SuppressWarnings("unchecked")
 763         @POST
 764         @Produces("application/json")
 765         @Path("/createwhitelist")
 766         public void createWhiteList(InputStream msg) {
 767
 768                 DMaaPContext ctx = getDmaapContext();
 769                 if (checkMirrorMakerPermission(ctx,
 770                                 AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, MIRROR_MAKERUSER))) {
 771
 772                         loadProperty();
 773                         String input = null;
 774
 775                         try {
 776                                 input = IOUtils.toString(msg, UTF_8);
 777
 778                                 if (input != null && input.length() > 0) {
 779                                         input = removeExtraChar(input);
 780                                 }
 781
 782                                 // Check if it is correct Json object
 783                                 JSONObject jsonOb = null;
 784
 785                                 try {
 786                                         jsonOb = new JSONObject(input);
 787
 788                                 } catch (JSONException ex) {
 789
 790                                         sendErrResponse(ctx, errorMessages.getIncorrectJson());
 791                                         LOGGER.error("JSONException: ", ex);
 792                                 }
 793
 794                                 // Check if the request has name and name contains only alpha numeric,
 795                                 // check if the request has namespace and
 796                                 // check if the request has whitelistTopicName
 797                                 // check if the topic name contains only alpha numeric
 798                                 if (jsonOb != null && jsonOb.length() == 3 && jsonOb.has("name")
 799                                                 && !StringUtils.isBlank(jsonOb.getString("name"))
 800                                                 && isAlphaNumeric(jsonOb.getString("name"))
 801                                                 && jsonOb.has(NAMESPACE) && !StringUtils.isBlank(jsonOb.getString(NAMESPACE))
 802                                                 && jsonOb.has("whitelistTopicName") && !StringUtils.isBlank(jsonOb.getString("whitelistTopicName"))
 803                                                 && isAlphaNumeric(jsonOb.getString("whitelistTopicName").substring(jsonOb.getString("whitelistTopicName").lastIndexOf(".")+1,
 804                                                                 jsonOb.getString("whitelistTopicName").length()))) {
 805
 806                                         String permission = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,
 807                                                         "msgRtr.mirrormakeruser.aaf.create") + jsonOb.getString(NAMESPACE) + "|create";
 808
 809                                         // Check if the user have create permission for the
 810                                         // namespace
 811                                         if (checkMirrorMakerPermission(ctx, permission)) {
 812
 813                                                 JSONObject listAll = new JSONObject();
 814                                                 JSONObject emptyObject = new JSONObject();
 815
 816                                                 // Create a listAllMirrorMaker Json object
 817                                                 try {
 818                                                         listAll.put("listAllMirrorMaker", emptyObject);
 819
 820                                                 } catch (JSONException e) {
 821
 822                                                         LOGGER.error("JSONException: ", e);
 823                                                 }
 824
 825                                                 // set a random number as messageID
 826                                                 String randomStr = getRandomNum();
 827                                                 listAll.put("messageID", randomStr);
 828                                                 InputStream inStream = null;
 829
 830                                                 // convert listAll Json object to InputStream object
 831                                                 try {
 832                                                         inStream = IOUtils.toInputStream(listAll.toString(), UTF_8);
 833
 834                                                 } catch (IOException ioe) {
 835                                                         ioe.printStackTrace();
 836                                                         LOGGER.error("IOException: ", ioe);
 837                                                 }
 838                                                 // call listAllMirrorMaker
 839                                                 mirrorService.pushEvents(ctx, topic, inStream, null, null);
 840
 841                                                 // subscribe for listMirrorMaker
 842                                                 long startTime = System.currentTimeMillis();
 843                                                 String msgFrmSubscribe = mirrorService.subscribe(ctx, topic, consumergroup, consumerid);
 844
 845                                                 while (!isListMirrorMaker(msgFrmSubscribe, randomStr)
 846                                                                 && (System.currentTimeMillis() - startTime) < timeout) {
 847                                                         msgFrmSubscribe = mirrorService.subscribe(ctx, topic, consumergroup, consumerid);
 848                                                 }
 849
 850                                                 JSONArray listMirrorMaker;
 851
 852                                                 if (msgFrmSubscribe != null && msgFrmSubscribe.length() > 0
 853                                                                 && isListMirrorMaker(msgFrmSubscribe, randomStr)) {
 854
 855                                                         listMirrorMaker = getListMirrorMaker(msgFrmSubscribe, randomStr);
 856                                                         String whitelist = null;
 857
 858                                                         for (int i = 0; i < listMirrorMaker.length(); i++) {
 859                                                                 JSONObject mm;
 860                                                                 mm = listMirrorMaker.getJSONObject(i);
 861                                                                 String name = mm.getString("name");
 862
 863                                                                 if (name.equals(jsonOb.getString("name")) && mm.has("whitelist")) {
 864                                                                         whitelist = mm.getString("whitelist");
 865                                                                         break;
 866                                                                 }
 867                                                         }
 868
 869                                                         List<String> topicList = new ArrayList<>();
 870                                                         List<String> finalTopicList = new ArrayList<>();
 871
 872                                                         if (whitelist != null) {
 873                                                                 topicList = Arrays.asList(whitelist.split(","));
 874                                                         }
 875
 876                                                         for (String st : topicList) {
 877                                                                 if (!StringUtils.isBlank(st)) {
 878                                                                         finalTopicList.add(st);
 879                                                                 }
 880                                                         }
 881
 882                                                         String newTopic = jsonOb.getString("whitelistTopicName");
 883
 884                                                         if (!topicList.contains(newTopic)
 885                                                                         && getNamespace(newTopic).equals(jsonOb.getString(NAMESPACE))) {
 886
 887                                                                 UpdateWhiteList updateWhiteList = new UpdateWhiteList();
 888                                                                 MirrorMaker mirrorMaker = new MirrorMaker();
 889                                                                 mirrorMaker.setName(jsonOb.getString("name"));
 890                                                                 finalTopicList.add(newTopic);
 891                                                                 String newWhitelist = "";
 892
 893                                                                 if (!finalTopicList.isEmpty()) {
 894                                                                         newWhitelist = StringUtils.join(finalTopicList, ",");
 895                                                                 }
 896
 897                                                                 mirrorMaker.setWhitelist(newWhitelist);
 898
 899                                                                 String newRandom = getRandomNum();
 900                                                                 updateWhiteList.setMessageID(newRandom);
 901                                                                 updateWhiteList.setUpdateWhiteList(mirrorMaker);
 902
 903                                                                 Gson g = new Gson();
 904                                                                 g.toJson(updateWhiteList);
 905                                                                 InputStream inputStream;
 906                                                                 inputStream = IOUtils.toInputStream(g.toJson(updateWhiteList), UTF_8);
 907                                                                 // callPubSub(newRandom, ctx, inputStream);
 908                                                                 callPubSubForWhitelist(newRandom, ctx, inputStream, jsonOb.getString(NAMESPACE));
 909
 910                                                         } else if (topicList.contains(newTopic)) {
 911                                                                 sendErrResponse(ctx, "The topic already exist.");
 912
 913                                                         } else if (!getNamespace(newTopic).equals(jsonOb.getString(NAMESPACE))) {
 914                                                                 sendErrResponse(ctx,
 915                                                                                 "The namespace of the topic does not match with the namespace you provided.");
 916                                                         }
 917                                                 } else {
 918
 919                                                         JSONObject err = new JSONObject();
 920                                                         err.append(ERROR,
 921                                                                         "listWhiteList is not available, please make sure MirrorMakerAgent is running");
 922                                                         DMaaPResponseBuilder.respondOk(ctx, err);
 923                                                 }
 924
 925                                         } else {
 926                                                 sendErrResponse(ctx, NO_USER_CREATE_PERMISSION);
 927                                         }
 928
 929                                 } else {
 930
 931                                         sendErrResponse(ctx, "This is not a createWhitelist request. Please try again.");
 932                                 }
 933
 934                         } catch (IOException | CambriaApiException | ConfigDbException | AccessDeniedException
 935                                         | TopicExistsException | missingReqdSetting | UnavailableException e) {
 936
 937                                 LOGGER.error("IOException: ", e);
 938                         }
 939                 }
 940                 // Send error response if user does not provide Authorization
 941                 else {
 942                         sendErrResponse(ctx, NO_USER_PERMISSION);
 943                 }
 944         }
 945
 946         @SuppressWarnings("unchecked")
 947         @POST
 948         @Produces("application/json")
 949         @Path("/deletewhitelist")
 950         public void deleteWhiteList(InputStream msg) {
 951
 952                 DMaaPContext ctx = getDmaapContext();
 953                 if (checkMirrorMakerPermission(ctx,
 954                                 AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, MIRROR_MAKERUSER))) {
 955
 956                         loadProperty();
 957                         String input = null;
 958
 959                         try {
 960                                 input = IOUtils.toString(msg, UTF_8);
 961
 962                                 if (input != null && input.length() > 0) {
 963                                         input = removeExtraChar(input);
 964                                 }
 965
 966                                 // Check if it is correct Json object
 967                                 JSONObject jsonOb = null;
 968
 969                                 try {
 970                                         jsonOb = new JSONObject(input);
 971
 972                                 } catch (JSONException ex) {
 973
 974                                         sendErrResponse(ctx, errorMessages.getIncorrectJson());
 975                                         LOGGER.error("JSONException: ", ex);
 976                                 }
 977
 978                                 // Check if the request has name and name contains only alpha numeric,
 979                                 // check if the request has namespace and
 980                                 // check if the request has whitelistTopicName
 981                                 if (jsonOb != null && jsonOb.length() == 3 && jsonOb.has("name") && isAlphaNumeric(jsonOb.getString("name"))
 982                                                 && jsonOb.has(NAMESPACE) && jsonOb.has("whitelistTopicName")
 983                                                 && isAlphaNumeric(jsonOb.getString("whitelistTopicName").substring(jsonOb.getString("whitelistTopicName").lastIndexOf(".")+1,
 984                                                                 jsonOb.getString("whitelistTopicName").length()))) {
 985
 986                                         String permission = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,
 987                                                         "msgRtr.mirrormakeruser.aaf.create") + jsonOb.getString(NAMESPACE) + "|create";
 988
 989                                         // Check if the user have create permission for the
 990                                         // namespace
 991                                         if (checkMirrorMakerPermission(ctx, permission)) {
 992
 993                                                 JSONObject listAll = new JSONObject();
 994                                                 JSONObject emptyObject = new JSONObject();
 995
 996                                                 // Create a listAllMirrorMaker Json object
 997                                                 try {
 998                                                         listAll.put("listAllMirrorMaker", emptyObject);
 999
1000                                                 } catch (JSONException e) {
1001
1002                                                         LOGGER.error("JSONException: ", e);
1003                                                 }
1004
1005                                                 // set a random number as messageID
1006                                                 String randomStr = getRandomNum();
1007                                                 listAll.put("messageID", randomStr);
1008                                                 InputStream inStream = null;
1009
1010                                                 // convert listAll Json object to InputStream object
1011                                                 try {
1012                                                         inStream = IOUtils.toInputStream(listAll.toString(), UTF_8);
1013
1014                                                 } catch (IOException ioe) {
1015                                                         ioe.printStackTrace();
1016                                                         LOGGER.error("IOException: ", ioe);
1017                                                 }
1018                                                 // call listAllMirrorMaker
1019                                                 mirrorService.pushEvents(ctx, topic, inStream, null, null);
1020
1021                                                 // subscribe for listMirrorMaker
1022                                                 long startTime = System.currentTimeMillis();
1023                                                 String msgFrmSubscribe = mirrorService.subscribe(ctx, topic, consumergroup, consumerid);
1024
1025                                                 while (!isListMirrorMaker(msgFrmSubscribe, randomStr)
1026                                                                 && (System.currentTimeMillis() - startTime) < timeout) {
1027                                                         msgFrmSubscribe = mirrorService.subscribe(ctx, topic, consumergroup, consumerid);
1028                                                 }
1029
1030                                                 JSONObject jsonObj;
1031                                                 JSONArray jsonArray;
1032                                                 JSONArray listMirrorMaker = null;
1033
1034                                                 if (msgFrmSubscribe != null && msgFrmSubscribe.length() > 0
1035                                                                 && isListMirrorMaker(msgFrmSubscribe, randomStr)) {
1036                                                         msgFrmSubscribe = removeExtraChar(msgFrmSubscribe);
1037                                                         jsonArray = new JSONArray(msgFrmSubscribe);
1038
1039                                                         for (int i = 0; i < jsonArray.length(); i++) {
1040                                                                 jsonObj = jsonArray.getJSONObject(i);
1041
1042                                                                 JSONObject obj = new JSONObject();
1043                                                                 if (jsonObj.has(MESSAGE)) {
1044                                                                         obj = jsonObj.getJSONObject(MESSAGE);
1045                                                                 }
1046                                                                 if (obj.has("messageID") && obj.get("messageID").equals(randomStr) && obj.has(LISTMIRRORMAKER)) {
1047                                                                         listMirrorMaker = obj.getJSONArray(LISTMIRRORMAKER);
1048                                                                         break;
1049                                                                 }
1050                                                         }
1051                                                         String whitelist = null;
1052                                                         if (listMirrorMaker != null) {
1053                                                                 for (int i = 0; i < listMirrorMaker.length(); i++) {
1054
1055                                                                         JSONObject mm = new JSONObject();
1056                                                                         mm = listMirrorMaker.getJSONObject(i);
1057                                                                         String name = mm.getString("name");
1058
1059                                                                         if (name.equals(jsonOb.getString("name")) && mm.has("whitelist")) {
1060                                                                                 whitelist = mm.getString("whitelist");
1061                                                                                 break;
1062                                                                         }
1063                                                                 }
1064                                                         }
1065
1066                                                         List<String> topicList = new ArrayList<>();
1067
1068                                                         if (whitelist != null) {
1069                                                                 topicList = Arrays.asList(whitelist.split(","));
1070                                                         }
1071                                                         boolean removeTopic = false;
1072                                                         String topicToRemove = jsonOb.getString("whitelistTopicName");
1073
1074                                                         if (topicList.contains(topicToRemove)) {
1075                                                                 removeTopic = true;
1076                                                         } else {
1077                                                                 sendErrResponse(ctx, "The topic does not exist.");
1078                                                         }
1079
1080
1081                                                         if (removeTopic) {
1082                                                                 UpdateWhiteList updateWhiteList = new UpdateWhiteList();
1083                                                                 MirrorMaker mirrorMaker = new MirrorMaker();
1084
1085                                                                 mirrorMaker.setName(jsonOb.getString("name"));
1086                                                                 mirrorMaker.setWhitelist(removeTopic(whitelist, topicToRemove));
1087
1088                                                                 String newRandom = getRandomNum();
1089
1090                                                                 updateWhiteList.setMessageID(newRandom);
1091                                                                 updateWhiteList.setUpdateWhiteList(mirrorMaker);
1092
1093                                                                 Gson g = new Gson();
1094                                                                 g.toJson(updateWhiteList);
1095
1096                                                                 InputStream inputStream;
1097                                                                 inputStream = IOUtils.toInputStream(g.toJson(updateWhiteList), UTF_8);
1098                                                                 callPubSubForWhitelist(newRandom, ctx, inputStream, getNamespace(topicToRemove));
1099                                                         }
1100
1101                                                 } else {
1102
1103                                                         JSONObject err = new JSONObject();
1104                                                         err.append(ERROR,
1105                                                                         "listWhiteList is not available, please make sure MirrorMakerAgent is running");
1106                                                         DMaaPResponseBuilder.respondOk(ctx, err);
1107                                                 }
1108
1109                                         } else {
1110                                                 sendErrResponse(ctx, NO_USER_CREATE_PERMISSION);
1111                                         }
1112
1113                                 } else {
1114
1115                                         sendErrResponse(ctx, "This is not a DeleteAllWhitelist request. Please try again.");
1116                                 }
1117
1118                         } catch (IOException | CambriaApiException | ConfigDbException | AccessDeniedException
1119                                         | TopicExistsException | missingReqdSetting | UnavailableException e) {
1120
1121                                 LOGGER.error("IOException: ", e);
1122                         }
1123                 }
1124                 // Send error response if user does not provide Authorization
1125                 else {
1126                         sendErrResponse(ctx, NO_USER_PERMISSION);
1127                 }
1128         }
1129
1130         private String getNamespace(String topic) {
1131                 return topic.substring(0, topic.lastIndexOf("."));
1132         }
1133
1134         private String removeTopic(String whitelist, String topicToRemove) {
1135                 List<String> topicList = new ArrayList<>();
1136                 List<String> newTopicList = new ArrayList<>();
1137
1138                 if (whitelist.contains(",")) {
1139                         topicList = Arrays.asList(whitelist.split(","));
1140
1141                 }
1142
1143                 if (topicList.contains(topicToRemove)) {
1144                         for (String topic: topicList) {
1145                                 if (!topic.equals(topicToRemove)) {
1146                                         newTopicList.add(topic);
1147                                 }
1148                         }
1149                 }
1150
1151                 String newWhitelist = StringUtils.join(newTopicList, ",");
1152
1153                 return newWhitelist;
1154         }
1155
1156         private void callPubSubForWhitelist(String randomStr, DMaaPContext ctx, InputStream inStream, String namespace) {
1157
1158                 try {
1159                         mirrorService.pushEvents(ctx, topic, inStream, null, null);
1160                         long startTime = System.currentTimeMillis();
1161                         String msgFrmSubscribe = mirrorService.subscribe(ctx, topic, consumergroup, consumerid);
1162
1163                         while (!isListMirrorMaker(msgFrmSubscribe, randomStr)
1164                                         && (System.currentTimeMillis() - startTime) < timeout) {
1165                                 msgFrmSubscribe = mirrorService.subscribe(ctx, topic, consumergroup, consumerid);
1166                         }
1167
1168                         JSONObject jsonObj;
1169                         JSONArray jsonArray;
1170                         JSONArray jsonArrayNamespace = null;
1171
1172                         if (msgFrmSubscribe != null && msgFrmSubscribe.length() > 0
1173                                         && isListMirrorMaker(msgFrmSubscribe, randomStr)) {
1174                                 msgFrmSubscribe = removeExtraChar(msgFrmSubscribe);
1175                                 jsonArray = new JSONArray(msgFrmSubscribe);
1176
1177                                 for (int i = 0; i < jsonArray.length(); i++) {
1178                                         jsonObj = jsonArray.getJSONObject(i);
1179
1180                                         JSONObject obj = new JSONObject();
1181                                         if (jsonObj.has(MESSAGE)) {
1182                                                 obj = jsonObj.getJSONObject(MESSAGE);
1183                                         }
1184                                         if (obj.has("messageID") && obj.get("messageID").equals(randomStr) && obj.has(LISTMIRRORMAKER)) {
1185                                                 jsonArrayNamespace = obj.getJSONArray(LISTMIRRORMAKER);
1186                                         }
1187                                 }
1188                                 JSONObject finalJasonObj = new JSONObject();
1189                                 JSONArray finalJsonArray = new JSONArray();
1190
1191                                 for (int i = 0; i < jsonArrayNamespace.length(); i++) {
1192
1193                                         JSONObject mmObj;
1194                                         mmObj = jsonArrayNamespace.getJSONObject(i);
1195                                         String whitelist;
1196
1197                                         if (mmObj.has("whitelist")) {
1198                                                 whitelist = getWhitelistByNamespace(mmObj.getString("whitelist"), namespace);
1199
1200                                                 if (whitelist != null) {
1201                                                         mmObj.remove("whitelist");
1202                                                         mmObj.put("whitelist", whitelist);
1203                                                 } else {
1204                                                         mmObj.remove("whitelist");
1205                                                 }
1206                                         }
1207                                         finalJsonArray.put(mmObj);
1208                                 }
1209                                 finalJasonObj.put(LISTMIRRORMAKER, finalJsonArray);
1210
1211                                 DMaaPResponseBuilder.respondOk(ctx, finalJasonObj);
1212
1213                         } else {
1214
1215                                 JSONObject err = new JSONObject();
1216                                 err.append(ERROR, "listMirrorMaker is not available, please make sure MirrorMakerAgent is running");
1217                                 DMaaPResponseBuilder.respondOk(ctx, err);
1218                         }
1219
1220                 } catch (Exception e) {
1221                         LOGGER.error("Exception: ", e);
1222                 }
1223         }
1224
1225         private String getWhitelistByNamespace(String originalWhitelist, String namespace) {
1226
1227                 String whitelist = null;
1228                 List<String> resultList = new ArrayList<>();
1229                 List<String> whitelistList = new ArrayList<>();
1230                 whitelistList = Arrays.asList(originalWhitelist.split(","));
1231
1232                 for (String topic : whitelistList) {
1233                         if (StringUtils.isNotBlank(originalWhitelist) && getNamespace(topic).equals(namespace)) {
1234                                 resultList.add(topic);
1235                         }
1236                 }
1237                 if (!resultList.isEmpty()) {
1238                         whitelist = StringUtils.join(resultList, ",");
1239                 }
1240
1241                 return whitelist;
1242         }
1243
1244         private JSONArray getListMirrorMaker(String msgFrmSubscribe, String randomStr) {
1245                 JSONObject jsonObj;
1246                 JSONArray jsonArray;
1247                 JSONArray listMirrorMaker = new JSONArray();
1248
1249                 msgFrmSubscribe = removeExtraChar(msgFrmSubscribe);
1250                 jsonArray = new JSONArray(msgFrmSubscribe);
1251
1252                 for (int i = 0; i < jsonArray.length(); i++) {
1253                         jsonObj = jsonArray.getJSONObject(i);
1254
1255                         JSONObject obj = new JSONObject();
1256                         if (jsonObj.has(MESSAGE)) {
1257                                 obj = jsonObj.getJSONObject(MESSAGE);
1258                         }
1259                         if (obj.has("messageID") && obj.get("messageID").equals(randomStr) && obj.has(LISTMIRRORMAKER)) {
1260                                 listMirrorMaker = obj.getJSONArray(LISTMIRRORMAKER);
1261                                 break;
1262                         }
1263                 }
1264                 return listMirrorMaker;
1265         }
1266 }