From 3275d36930811d98baa55954f066514992267fe4 Mon Sep 17 00:00:00 2001
From: Martin Harris <github@nakomis.com>
Date: Fri, 14 Nov 2014 16:39:13 +0000
Subject: [PATCH] Minor changes to help downstream Riak project

---
 .../brooklyn/location/jclouds/JcloudsLocation.java | 14 +++++++++++++-
 .../entity/nosql/riak/RiakClusterImpl.java         |  2 +-
 .../entity/nosql/riak/RiakNodeSshDriver.java       | 12 ++++++++----
 3 files changed, 22 insertions(+), 6 deletions(-)

diff --git a/locations/jclouds/src/main/java/brooklyn/location/jclouds/JcloudsLocation.java b/locations/jclouds/src/main/java/brooklyn/location/jclouds/JcloudsLocation.java
index e77fb81de0..98daca08c1 100644
--- a/locations/jclouds/src/main/java/brooklyn/location/jclouds/JcloudsLocation.java
+++ b/locations/jclouds/src/main/java/brooklyn/location/jclouds/JcloudsLocation.java
@@ -718,7 +718,19 @@ protected JcloudsSshMachineLocation obtainOnce(ConfigBag setup) throws NoMachine
                     
                     List<String> iptablesRules = createIptablesRulesForNetworkInterface((Iterable<Integer>) setup.get(INBOUND_PORTS));
                     iptablesRules.add(IptablesCommands.saveIptablesRules());
-                    sshMachineLocation.execCommands("Inserting iptables rules", iptablesRules);
+                    List<String> batch = Lists.newArrayList();
+                    // Some entities, such as Riak (erlang based) have a huge range of ports, which leads to a script that
+                    // is too large to run (fails with a broken pipe). Batch the rules into batches of 50
+                    for (String rule : iptablesRules) {
+                        batch.add(rule);
+                        if (batch.size() == 50) {
+                            sshMachineLocation.execCommands("Inserting iptables rules, 50 command batch", batch);
+                            batch.clear();
+                        }
+                    }
+                    if (batch.size() > 0) {
+                        sshMachineLocation.execCommands("Inserting iptables rules", batch);
+                    }
                     sshMachineLocation.execCommands("List iptables rules", ImmutableList.of(IptablesCommands.listIptablesRule()));
                 }
                 
diff --git a/software/nosql/src/main/java/brooklyn/entity/nosql/riak/RiakClusterImpl.java b/software/nosql/src/main/java/brooklyn/entity/nosql/riak/RiakClusterImpl.java
index 4ffec3fc17..f16932bf9c 100644
--- a/software/nosql/src/main/java/brooklyn/entity/nosql/riak/RiakClusterImpl.java
+++ b/software/nosql/src/main/java/brooklyn/entity/nosql/riak/RiakClusterImpl.java
@@ -79,7 +79,7 @@ public boolean apply(@Nullable Entity entity) {
 
         if (anyNode.isPresent()) {
             log.info("Planning and Committing cluster changes on node: {}, cluster: {}", anyNode.get().getId(), getId());
-            Entities.invokeEffector(this, anyNode.get(), RiakNode.COMMIT_RIAK_CLUSTER);
+            Entities.invokeEffector(this, anyNode.get(), RiakNode.COMMIT_RIAK_CLUSTER).blockUntilEnded();
             setAttribute(IS_CLUSTER_INIT, true);
         } else {
             log.warn("No Riak Nodes are found on the cluster: {}. Initialization Failed", getId());
diff --git a/software/nosql/src/main/java/brooklyn/entity/nosql/riak/RiakNodeSshDriver.java b/software/nosql/src/main/java/brooklyn/entity/nosql/riak/RiakNodeSshDriver.java
index 8e45424ce6..090a85a847 100644
--- a/software/nosql/src/main/java/brooklyn/entity/nosql/riak/RiakNodeSshDriver.java
+++ b/software/nosql/src/main/java/brooklyn/entity/nosql/riak/RiakNodeSshDriver.java
@@ -287,11 +287,11 @@ public String getRiakEtcDir() {
         return isPackageInstall ? "/etc/riak" : Urls.mergePaths(getExpandedInstallDir(), "etc");
     }
 
-    private String getRiakCmd() {
+    protected String getRiakCmd() {
         return isPackageInstall ? "riak" : Urls.mergePaths(getExpandedInstallDir(), "bin/riak");
     }
 
-    private String getRiakAdminCmd() {
+    protected String getRiakAdminCmd() {
         return isPackageInstall ? "riak-admin" : Urls.mergePaths(getExpandedInstallDir(), "bin/riak-admin");
     }
 
@@ -414,12 +414,16 @@ private Boolean hasJoinedCluster() {
         return ((RiakNode) entity).hasJoinedCluster();
     }
 
-    private boolean isRiakOnPath() {
+    protected boolean isRiakOnPath() {
         return (newScript("riakOnPath")
                 .body.append("which riak")
                 .execute() == 0);
     }
 
+    protected boolean isPackageInstall() {
+        return isPackageInstall;
+    }
+
     private String getRiakName() {
         return entity.getAttribute(RiakNode.RIAK_NODE_NAME);
     }
@@ -429,7 +433,7 @@ private String getRingStateDir() {
         return isPackageInstall ? "/var/lib/riak/ring" : Urls.mergePaths(getExpandedInstallDir(), "lib/ring");
     }
 
-    private boolean isVersion1() {
+    protected boolean isVersion1() {
         return getVersion().startsWith("1.");
     }