Add optional timeout to YZ create index command

riak/client/operations.py

@@ -685,7 +685,8 @@ def stream_mapred(self, inputs, query, timeout):
             stream.close()
 
     @retryable
-    def create_search_index(self, transport, index, schema=None, n_val=None):
+    def create_search_index(self, transport, index, schema=None, n_val=None,
+                            timeout=None):
         """
         create_search_index(index, schema=None, n_val=None)
 
@@ -698,8 +699,10 @@ def create_search_index(self, transport, index, schema=None, n_val=None):
         :type schema: string, None
         :param n_val: this indexes N value
         :type n_val: integer, None
+        :param timeout: optional timeout (in ms)
+        :type timeout: integer, None
         """
-        return transport.create_search_index(index, schema, n_val)
+        return transport.create_search_index(index, schema, n_val, timeout)
 
     @retryable
     def get_search_index(self, transport, index):

riak/tests/test_all.py

@@ -87,7 +87,7 @@ def setUpModule():
                          'index': 'mrbucket'}
 
         for yz in (testrun_yz, testrun_yz_index, testrun_yz_mr):
-            c.create_search_index(yz['index'])
+            c.create_search_index(yz['index'], timeout=30000)
             if yz['btype'] is not None:
                 t = c.bucket_type(yz['btype'])
                 b = t.bucket(yz['bucket'])

riak/transports/http/transport.py

@@ -444,7 +444,8 @@ def stream_index(self, bucket, index, startkey, endkey=None,
         else:
             raise RiakError('Error streaming secondary index.')
 
-    def create_search_index(self, index, schema=None, n_val=None):
+    def create_search_index(self, index, schema=None, n_val=None,
+                            timeout=None):
         """
         Create a Solr search index for Yokozuna.
 
@@ -454,6 +455,8 @@ def create_search_index(self, index, schema=None, n_val=None):
         :type schema: string
         :param n_val: N value of the write
         :type n_val: int
+        :param timeout: optional timeout (in ms)
+        :type timeout: integer, None
 
         :rtype boolean
         """
@@ -468,6 +471,8 @@ def create_search_index(self, index, schema=None, n_val=None):
             content_dict['schema'] = schema
         if n_val:
             content_dict['n_val'] = n_val
+        if timeout:
+            content_dict['timeout'] = timeout
         content = json.dumps(content_dict)
 
         # Run the request...

riak/transports/pbc/transport.py

@@ -487,7 +487,8 @@ def stream_index(self, bucket, index, startkey, endkey=None,
 
         return RiakPbcIndexStream(self, index, return_terms)
 
-    def create_search_index(self, index, schema=None, n_val=None):
+    def create_search_index(self, index, schema=None, n_val=None,
+                            timeout=None):
         if not self.pb_search_admin():
             raise NotImplementedError("Search 2.0 administration is not "
                                       "supported for this version")
@@ -498,6 +499,8 @@ def create_search_index(self, index, schema=None, n_val=None):
         if n_val:
             idx.n_val = n_val
         req = riak_pb.RpbYokozunaIndexPutReq(index=idx)
+        if timeout is not None:
+            req.timeout = timeout
 
         self._request(MSG_CODE_YOKOZUNA_INDEX_PUT_REQ, req,
                       MSG_CODE_PUT_RESP)

riak/transports/transport.py

@@ -172,7 +172,8 @@ def get_client_id(self):
         """
         raise NotImplementedError
 
-    def create_search_index(self, index, schema=None, n_val=None):
+    def create_search_index(self, index, schema=None, n_val=None,
+                            timeout=None):
         """
         Creates a yokozuna search index.
         """