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