Add support for creating river indexes.

lib/elasticsearch/client/admin_index.rb

@@ -122,6 +122,22 @@ def optimize(*args)
           indices.collect! { |i| PSEUDO_INDICES.include?(i) ? "_#{i}" : i }
           execute(:optimize, indices, options)
         end
+
+        def create_river(type, create_options, options={})
+          execute(:create_river, type, create_options, options)
+        end
+
+        def get_river(type, options={})
+          execute(:get_river, type, options)
+        end
+
+        def river_status(type, options={})
+          execute(:river_status, type, options)
+        end
+
+        def delete_river(type=nil, options={})
+          execute(:delete_river, type, options)
+        end
       end
     end
   end

lib/elasticsearch/transport/base_protocol.rb

@@ -141,6 +141,24 @@ def snapshot(index_list, options={})
       def optimize(index_list, options={})
         standard_request(:post, {:index => index_list, :op => "_optimize"}, options, {})
       end
+
+      def create_river(type, create_options={}, options={})
+        standard_request(:put, {:index => "_river", :type => type, :op => "_meta"}, options, encoder.encode(create_options))
+      end
+
+      def get_river(type, options={})
+        standard_request(:get, {:index => "_river", :type => type, :op => "_meta"})
+      end
+
+      def river_status(type, options={})
+        standard_request(:get, {:index => "_river", :type => type, :op => "_status"})
+      end
+
+      def delete_river(type, options={})
+        params = {:index => "_river"}
+        params[:type] = type unless type.nil?
+        standard_request(:delete, params)
+      end
     end
 
     module ClusterAdminProtocol