/*
 * Copyright (c) 2005-2014, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * WSO2 Inc. licenses this file to you under the Apache License,
 * Version 2.0 (the "License"); you may not use this file except
 * in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.wso2.andes.server.queue;


import org.apache.log4j.Logger;
import org.wso2.andes.exchange.ExchangeDefaults;
import org.wso2.andes.framing.AMQShortString;
import org.wso2.andes.kernel.*;
import org.wso2.andes.server.ClusterResourceHolder;
import org.wso2.andes.server.cluster.coordination.ClusterCoordinationHandler;
import org.wso2.andes.server.cluster.coordination.hazelcast.HazelcastAgent;
import org.wso2.andes.server.util.AndesConstants;
import org.wso2.andes.server.virtualhost.VirtualHost;
import org.wso2.andes.subscription.AMQPLocalSubscription;

/**
 * This class centralises the management of Dead Letter Queues by creating Dead Letter Queues when requested and
 * deciding on whether a queue is a Dead Letter Queue or not and generating the 'Dead Letter Queue' queue name for
 * the tenant.
 */
public class DLCQueueUtils {

    private static final Logger log = Logger.getLogger(DLCQueueUtils.class);

    /**
     * Derive the Dead Letter Queue name of the tenant with respect to a given queue of the same tenant.
     *
     * @param queueName  A queue name in the same tenant.
     * @param dlc_string The Dead Letter Queue suffix.
     * @return The Dead Letter Queue name for the tenant.
     */
    public static String identifyTenantInformationAndGenerateDLCString(String queueName, String dlc_string) {
        String destination_string;
        if (queueName.contains("/")) {
            //The Queue is in the tenant realm
            destination_string = queueName.split("/", 2)[0] + "/" + dlc_string;
        } else {
            destination_string = dlc_string;
        }
        return destination_string;
    }

    /**
     * Decides on whether a given queue name is a Dead Letter Queue or not.
     *
     * @param queueName The Queue name to test.
     * @return True if a Dead Letter Queue, False if not a Dead Letter Queue.
     */
    public static boolean isDeadLetterQueue(String queueName) {
        boolean isDeadLetterQueue = false;
        if (queueName.contains("/")) {
            //The Queue is in the tenant realm
            if (queueName.split("/", 2)[1].contains(AndesConstants.DEAD_LETTER_QUEUE_NAME)) {
                isDeadLetterQueue = true;
            }
        } else {
            if (queueName.equals(AndesConstants.DEAD_LETTER_QUEUE_NAME)) {
                isDeadLetterQueue = true;
            }
        }

        return isDeadLetterQueue;
    }

    /**
     * Creates a Dead Letter Queue for the tenant in a given queue name.
     *
     * @param queueName A queue name in the same tenant.
     * @param host      The Virtual Host.
     * @param owner     The tenant owner.
     * @throws AndesException
     */
    public static synchronized void createDLCQueue(String queueName, VirtualHost host,
                                                   String owner) throws AndesException {
        String dlcQueueName = identifyTenantInformationAndGenerateDLCString(queueName,
                AndesConstants.DEAD_LETTER_QUEUE_NAME);
        QueueRegistry queueRegistry = host.getQueueRegistry();
        AMQQueue queue = queueRegistry.getQueue(new AMQShortString(dlcQueueName));
        if (queue == null && !isDeadLetterQueue(queueName)) {
            AndesQueue andesQueue = new AndesQueue(dlcQueueName, owner, false, true);

            AndesContext.getInstance().getAMQPConstructStore().addQueue(andesQueue, true);
            ClusterResourceHolder.getInstance().getVirtualHostConfigSynchronizer().queue(dlcQueueName, owner, false,
                    null);

            QueueListener queueListener = new ClusterCoordinationHandler(HazelcastAgent
                    .getInstance());
            queueListener.handleLocalQueuesChanged(andesQueue, QueueListener.QueueChange.Added);

            LocalSubscription mockSubscription = new AMQPLocalSubscription(queueRegistry.getQueue(new
                    AMQShortString(dlcQueueName)), null, "0", dlcQueueName, false, false, false, null,
                    dlcQueueName, owner, ExchangeDefaults.DIRECT_EXCHANGE_NAME.toString(), "DIRECT", null, false);
            AndesContext.getInstance().getSubscriptionStore().createDisconnectOrRemoveClusterSubscription
                    (mockSubscription, SubscriptionListener.SubscriptionChange.Added);

            log.info(dlcQueueName + " Queue Created as Dead Letter Channel");
        }
    }
}

@@ -1342,16 +1342,19 @@
 ues when
+%0A *
  request
@@ -1359,19 +1359,16 @@
 sted and
-%0A *
  decidin
@@ -1438,24 +1438,27 @@
 ng the 'Dead
+%0A *
  Letter Queu
@@ -1474,19 +1474,16 @@
 name for
-%0A *
  the ten
@@ -1695,32 +1695,39 @@
 ueue of the same
+%0A     *
  tenant.%0A     *%0A
@@ -1749,16 +1749,30 @@
 ueueName
+%0A     *       
   A queu
@@ -1815,23 +1815,37 @@
 aram dlc
-_s
+S
 tring
+%0A     *        
  The Dea
@@ -2021,16 +2021,87 @@
 eueName,
+%0A                                                                      
  String 
@@ -2099,26 +2099,25 @@
   String dlc
-_s
+S
 tring) %7B%0A   
@@ -2135,33 +2135,33 @@
  destination
-_s
+S
 tring;%0A
+%0A
         if (
@@ -2156,33 +2156,27 @@
         if (
-queueName
+%22/%22
 .contains(%22/
@@ -2165,35 +2165,41 @@
 f (%22/%22.contains(
-%22/%22
+queueName
 )) %7B%0A           
@@ -2249,34 +2249,33 @@
      destination
-_s
+S
 tring = queueNam
@@ -2300,26 +2300,25 @@
  + %22/%22 + dlc
-_s
+S
 tring;%0A     
@@ -2348,26 +2348,25 @@
  destination
-_s
+S
 tring = dlc_
@@ -2364,18 +2364,17 @@
 ng = dlc
-_s
+S
 tring;%0A 
@@ -2374,32 +2374,33 @@
 ring;%0A        %7D%0A
+%0A
         return d
@@ -2413,10 +2413,9 @@
 tion
-_s
+S
 trin
@@ -2538,16 +2538,31 @@
 ueueName
+%0A     *        
  The Que
@@ -3318,16 +3318,31 @@
 ueueName
+%0A     *        
  A queue
@@ -3385,16 +3385,26 @@
 ram host
+%0A     *   
       Th
@@ -3438,16 +3438,27 @@
 am owner
+%0A     *    
      The

Remove K&R style variable names and fix a possible NPE