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

import org.apache.karaf.cellar.core.Node;
import org.apache.karaf.cellar.core.control.ConsumerSwitchCommand;
import org.apache.karaf.cellar.core.control.ConsumerSwitchResult;
import org.apache.karaf.cellar.core.control.SwitchStatus;
import org.apache.karaf.cellar.shell.ClusterCommandSupport;

import java.util.List;
import java.util.Map;
import java.util.Set;

public abstract class ConsumerSupport extends ClusterCommandSupport {

    protected static final String OUTPUT_FORMAT = "%-20s %s";

    protected Object doExecute(List<String> nodes, SwitchStatus status) throws Exception {
        ConsumerSwitchCommand command = new ConsumerSwitchCommand(clusterManager.generateId());
        Set<Node> recipientList = clusterManager.listNodes(nodes);

        if (recipientList != null && !recipientList.isEmpty()) {
            command.setDestination(recipientList);
        }

        command.setStatus(status);

        Map<Node, ConsumerSwitchResult> results = executionContext.execute(command);
        if (results == null || results.isEmpty()) {
            System.out.println("No result received within given timeout");
        } else {
            System.out.println(String.format(OUTPUT_FORMAT, "Node", "Status"));
            for (Node node : results.keySet()) {
                ConsumerSwitchResult result = results.get(node);
                System.out.println(String.format(OUTPUT_FORMAT, node.getId(), result.getStatus()));
            }
        }
        return null;
    }

}

@@ -900,16 +900,42 @@
 pport;%0A%0A
+import java.util.HashSet;%0A
 import j
@@ -1065,16 +1065,83 @@
 port %7B%0A%0A
+    protected static final String HEADER_FORMAT = %22 %25-30s   %25-5s%22;%0A
     prot
@@ -1187,16 +1187,22 @@
  = %22
+%5B
 %25-
-20s %25s
+30s%5D %5B%25-5s%5D
 %22;%0A%0A
@@ -1249,16 +1249,18 @@
 ng%3E node
+Id
 s, Switc
@@ -1294,16 +1294,17 @@
 ption %7B%0A
+%0A
         
@@ -1395,88 +1395,553 @@
 ));%0A
-        Set%3CNode%3E recipientList = clusterManager.listNodes
+%0A        // looking for nodes and check if exist%0A        Set%3CNode%3E recipientList;%0A        if (nodeIds != null && !nodeIds.isEmpty()) %7B%0A            recipientList = new HashSet%3CNode%3E();%0A            for (String nodeId : nodeIds) %7B%0A                Node node = clusterManager.findNodeById(nodeId);%0A                if (node == null) %7B%0A                    System.err.println(%22Cluster node %22 + nodeId + %22 doesn't exist%22);%0A                %7D else %7B%0A                    recipientList.add
 (node
-s
 );%0A
-%0A        if (
+                %7D%0A            %7D%0A        %7D else %7B%0A            
 reci
@@ -1950,28 +1950,69 @@
 entList 
-!
 = 
-null && !
+clusterManager.listNodes();%0A        %7D%0A%0A        if (
 recipien
@@ -2017,33 +2017,34 @@
 entList.
-isEmpty()
+size() %3C 1
 ) %7B%0A    
         
@@ -2027,32 +2027,65 @@
 ze() %3C 1) %7B%0A    
+        return null;%0A        %7D%0A%0A%0A
         command.
@@ -2115,27 +2115,16 @@
 tList);%0A
-        %7D%0A%0A
         
@@ -2417,38 +2417,38 @@
 n(String.format(
-OUTPUT
+HEADER
 _FORMAT, %22Node%22,

Check if nodes exist in the cluster:consumer-* commands