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