Flush API: Add refresh flag (refresh after flush). Closes #14.

Author: kimchy

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/flush/FlushRequest.java

@@ -22,11 +22,17 @@
 import org.elasticsearch.action.support.broadcast.BroadcastOperationRequest;
 import org.elasticsearch.action.support.broadcast.BroadcastOperationThreading;
 
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
 /**
  * @author kimchy (Shay Banon)
  */
 public class FlushRequest extends BroadcastOperationRequest {
 
+    private boolean refresh = false;
+
     FlushRequest() {
 
     }
@@ -37,6 +43,15 @@ public FlushRequest(String... indices) {
         operationThreading(BroadcastOperationThreading.THREAD_PER_SHARD);
     }
 
+    public boolean refresh() {
+        return this.refresh;
+    }
+
+    public FlushRequest refresh(boolean refresh) {
+        this.refresh = refresh;
+        return this;
+    }
+
     @Override public FlushRequest listenerThreaded(boolean threadedListener) {
         super.listenerThreaded(threadedListener);
         return this;
@@ -46,4 +61,14 @@ public FlushRequest(String... indices) {
         super.operationThreading(operationThreading);
         return this;
     }
+
+    @Override public void writeTo(DataOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeBoolean(refresh);
+    }
+
+    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+        super.readFrom(in);
+        refresh = in.readBoolean();
+    }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/flush/ShardFlushRequest.java

@@ -30,18 +30,27 @@
  */
 public class ShardFlushRequest extends BroadcastShardOperationRequest {
 
+    private boolean refresh;
+
     ShardFlushRequest() {
     }
 
-    public ShardFlushRequest(String index, int shardId) {
+    public ShardFlushRequest(String index, int shardId, FlushRequest request) {
         super(index, shardId);
+        this.refresh = request.refresh();
+    }
+
+    public boolean refresh() {
+        return this.refresh;
     }
 
     @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
         super.readFrom(in);
+        refresh = in.readBoolean();
     }
 
     @Override public void writeTo(DataOutput out) throws IOException {
         super.writeTo(out);
+        out.writeBoolean(refresh);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/flush/TransportFlushAction.java

@@ -76,7 +76,7 @@ public class TransportFlushAction extends TransportBroadcastOperationAction<Flus
     }
 
     @Override protected ShardFlushRequest newShardRequest(ShardRouting shard, FlushRequest request) {
-        return new ShardFlushRequest(shard.index(), shard.id());
+        return new ShardFlushRequest(shard.index(), shard.id(), request);
     }
 
     @Override protected ShardFlushResponse newShardResponse() {
@@ -85,7 +85,7 @@ public class TransportFlushAction extends TransportBroadcastOperationAction<Flus
 
     @Override protected ShardFlushResponse shardOperation(ShardFlushRequest request) throws ElasticSearchException {
         IndexShard indexShard = indicesService.indexServiceSafe(request.index()).shardSafe(request.shardId());
-        indexShard.flush(new Engine.Flush());
+        indexShard.flush(new Engine.Flush().refresh(request.refresh()));
         return new ShardFlushResponse(request.index(), request.shardId());
     }
 

modules/elasticsearch/src/main/java/org/elasticsearch/http/action/admin/indices/flush/HttpFlushAction.java

@@ -56,6 +56,7 @@ public class HttpFlushAction extends BaseHttpServerHandler {
             operationThreading = BroadcastOperationThreading.THREAD_PER_SHARD;
         }
         flushRequest.operationThreading(operationThreading);
+        flushRequest.refresh(HttpActions.paramAsBoolean("refresh", false));
         client.admin().indices().execFlush(flushRequest, new ActionListener<FlushResponse>() {
             @Override public void onResponse(FlushResponse response) {
                 try {

modules/elasticsearch/src/main/java/org/elasticsearch/index/engine/Engine.java

@@ -139,8 +139,25 @@ public boolean waitForOperations() {
 
     static class Flush {
 
+        private boolean refresh = false;
+
+        /**
+         * Should a refresh be performed after flushing. Defaults to <tt>false</tt>.
+         */
+        public boolean refresh() {
+            return this.refresh;
+        }
+
+        /**
+         * Should a refresh be performed after flushing. Defaults to <tt>false</tt>.
+         */
+        public Flush refresh(boolean refresh) {
+            this.refresh = refresh;
+            return this;
+        }
+
         @Override public String toString() {
-            return "";
+            return "refresh[" + refresh + "]";
         }
     }
 

modules/elasticsearch/src/main/java/org/elasticsearch/index/engine/robin/RobinEngine.java

@@ -290,6 +290,9 @@ public class RobinEngine extends AbstractIndexShardComponent implements Engine,
         } finally {
             rwl.writeLock().unlock();
         }
+        if (flush.refresh()) {
+            refresh(new Refresh(false));
+        }
     }
 
     @Override public void optimize(Optimize optimize) throws EngineException {