zorba-coders team mailing list archive

[Juan Zacarias <juan457@xxxxxxxxx>]

Juan Zacarias has proposed merging lp:~zorba-coders/zorba/couchbase-module-new-options into lp:zorba/couchbase-module.

Commit message:
*Added new options to the cb:view Function:
-"limit" option, allows the user to limit the numbers of results shown by the view
-"stale" option, allows the user to specify when to update the view: update after the view is returned(default by couchbase), update before the view is returned, and don't update if the view function is called.

*Added a new option to the cb:put-text and cb:put-binary Functions: 
-"wait" option, allows the user to wait for confirmation from couchbase that the key/value pair was stored in disk.

Modified test view.xq using the new options so it doesn't fail when ran for the first time.

Requested reviews:
  Sorin Marian Nasoi (sorin.marian.nasoi)
  Chris Hillery (ceejatec)

For more details, see:
https://code.launchpad.net/~zorba-coders/zorba/couchbase-module-new-options/+merge/143187
-- 
https://code.launchpad.net/~zorba-coders/zorba/couchbase-module-new-options/+merge/143187
Your team Zorba Coders is subscribed to branch lp:zorba/couchbase-module.

=== modified file 'examples/example.xq'
--- examples/example.xq	2013-01-10 07:59:35 +0000
+++ examples/example.xq	2013-01-14 21:43:21 +0000
@@ -7,7 +7,7 @@
   "password" : null,
   "bucket" : "default"});
 variable $view-name := cb:create-view($instance, "zip", "zip", {"key" : "doc.state", "values" : "doc.pop"});
-variable $data := cb:view($instance, $view-name);
+variable $data := cb:view($instance, $view-name, {"stale" : "false"});
 for $d in jn:members($data("rows"))
 let $state := $d("key")
 group by $state

=== modified file 'src/couchbase.xq'
--- src/couchbase.xq	2013-01-10 08:07:25 +0000
+++ src/couchbase.xq	2013-01-14 21:43:21 +0000
@@ -232,7 +232,10 @@
  : @option "operation" type of operation, possible values are 
  :         "add", "replace", "set", "append" and "prepend".
  : @option "encoding" the encoding that should be used for the
- :   value (default is UTF-8).
+ :         value (default is UTF-8).
+ : @option "wait" variable for setting if a wait for persistancy in 
+ :         the storing key is needed, possible values are "persist" 
+ :         and "false".
  : 
  : @error cb:LCB0002 if any error occurs in the communication with
  :   the server.
@@ -241,6 +244,7 @@
  : @error cb:CB0006 if the given encoding is not supported.
  : @error cb:CB0007 if any of the options is not supported.
  : @error cb:CB0009 if the given expiration time is not an xs:integer. 
+ : @error cb:CB0011 if the stored Variable was not stored
  :
  : @return a empty sequence.
  :)  
@@ -286,6 +290,9 @@
  :         expiration time in seconds.
  : @option "operation" type of operation, possible values are 
  :         "add", "replace", "set", "append" and "prepend".
+ : @option "wait" variable for setting if a wait for persistancy in 
+ :         the storing key is needed, possible values are "persist" 
+ :         and "false".
  :
  : @error cb:LCB0002 if any error occurs in the communication with
  :   the server.
@@ -293,6 +300,7 @@
  :   of values.
  : @error cb:CB0007 if any of the options is not supported.
  : @error cb:CB0009 if the given expiration time is not an xs:integer.
+ : @error cb:CB0011 if the stored Variable was not stored
  :
  : @return a empty sequence.
  :)  
@@ -365,8 +373,18 @@
  :        (e.g. "_design/test/_view/vies").
  : @param $options JSONiq object with additional options
  :
- : @option "encoding" string with the name of the encoding of the returned
- :   strings (if not UTF-8).
+ : @option Json object whith options for the querying the view. available options:
+ :         "encoding" string with the name of the encoding of the returned
+ :         strings (if not UTF-8).
+ :         "stale" option's value is a string with the type of stale to be used. 
+ :         Valid values:
+ :           "ok" : the view is not updated
+ :           "false" : the view is updated before the function view is executed, 
+ :            this options needs the key to be on disk before the call of the 
+ :            view function.
+ :           "update_after" : the view is updated after the call of view
+ :         "limit" option's value is an integer which sets a number of how many 
+ :         rows the view will show.  
  :
  : @error cb:LCB0002 if any error occurs in the communication with
  :   the server.

=== modified file 'src/couchbase.xq.src/couchbase.cpp'
--- src/couchbase.xq.src/couchbase.cpp	2013-01-10 07:59:35 +0000
+++ src/couchbase.xq.src/couchbase.cpp	2013-01-14 21:43:21 +0000
@@ -190,6 +190,30 @@
   return NULL;
 }
 
+String
+  CouchbaseFunction::ViewOptions::getPathOptions()
+{
+  String lPathOptions("?");
+  bool lAmp = false;
+  if (theStaleOption != "")
+  {
+    lPathOptions.append(theStaleOption);
+    lAmp = true;
+  }
+  if (theLimitOption != "")
+  {
+    if(lAmp)
+      lPathOptions.append("&");
+    lPathOptions.append(theLimitOption);
+    lAmp = true;
+  }
+
+  if (lPathOptions == "?")
+    lPathOptions = "";
+
+  return lPathOptions;
+}
+
 void
   CouchbaseFunction::ViewOptions::setOptions(Item& aOptions)
 {
@@ -216,6 +240,42 @@
         throwError("CB0006", lMsg.str().c_str());
       }
     }
+    else if (lStrKey == "stale")
+    {
+      Item lValue = aOptions.getObjectValue(lStrKey);
+      String lString = lValue.getStringValue();
+      if (lString == "false")
+      {
+        theStaleOption ="stale=false";
+      }
+      else if (lString == "ok")
+      {
+        theStaleOption ="stale=ok";
+      }
+      else if (lString == "update_after")
+      {
+        theStaleOption ="stale=update_after";
+      }
+      else
+      {
+        std::ostringstream lMsg;
+        lMsg << lStrKey << "=" << lString << ": option not supported";
+        throwError("CB0007", lMsg.str().c_str());
+      }
+    }
+    else if (lStrKey == "limit")
+    {
+      Item lValue = aOptions.getObjectValue(lStrKey);
+      try
+      {
+        int lLimit = lValue.getIntValue();
+        theLimitOption = "limit=" + lLimit;  
+      }
+      catch (ZorbaException& e)
+      {
+        throwError("CB0009", " limit option must be an integer value");
+      } 
+    }
     else
     {
       std::ostringstream lMsg;
@@ -371,6 +431,28 @@
             throwError("CB0006", lMsg.str().c_str());
       }
     }
+    else if (lStrKey == "wait")
+    {
+      Item lValue = aOptions.getObjectValue(lStrKey);
+      String lStrValue = lValue.getStringValue();
+      std::transform(
+        lStrValue.begin(), lStrValue.end(),
+        lStrValue.begin(), tolower);
+      if (lStrValue == "persist")
+      {
+        theWaitType = CB_WAIT_PERSIST;
+      }
+      else if (lStrValue == "false")
+      {
+        theWaitType = CB_WAIT_FALSE;
+      }
+      else
+      {
+        std::ostringstream lMsg;
+        lMsg << lStrKey << "=" << lStrValue << " : option not supported";
+        throwError("CB0007", lMsg.str().c_str());
+      }
+    }
     else
     {
       std::ostringstream lMsg;
@@ -731,6 +813,26 @@
 /*******************************************************************************
  ******************************************************************************/
 
+void CouchbaseFunction::observe_callback(lcb_t instance, const void *cookie, lcb_error_t error, const lcb_observe_resp_t *resp)
+{
+  if (error != LCB_SUCCESS)
+  {
+    libCouchbaseError (instance, error);
+  }
+  
+  PutOptions* lWait = (PutOptions*) cookie;
+  //verify is coming from the master
+  if (resp->v.v0.from_master > 0)
+  {
+    lcb_observe_t lStatus = resp->v.v0.status;if (lStatus == LCB_OBSERVE_NOT_FOUND)
+    {
+      throwError("CB0011", "Variable stored not found.");
+    }
+    //check for flag of persisntancy
+    lWait->setWaiting(lStatus & LCB_OBSERVE_PERSISTED?false:true);
+  }
+}
+
 void CouchbaseFunction::put (lcb_t aInstance, Iterator_t aKeys, Iterator_t aValues, PutOptions aOptions)
 {
   lcb_error_t lError;
@@ -800,9 +902,23 @@
     {
       libCouchbaseError (aInstance, lError);
     } 
-
-
+    //Wait for store
     lcb_wait(aInstance);
+    //Check if wait for disk
+    if (aOptions.getWaitType() != CB_WAIT_FALSE)
+    {     
+      lcb_set_observe_callback(aInstance, observe_callback);
+      PutOptions* lOptions = &aOptions;
+      do {
+        lcb_observe_cmd_t lObserve;
+        lObserve.version = 0;
+        lObserve.v.v0.key = lStrKey.c_str();
+        lObserve.v.v0.nkey = lStrKey.size();
+        lcb_observe_cmd_t* lCommands[1] = { &lObserve };
+        lcb_observe(aInstance, lOptions, 1, lCommands);
+        lcb_wait(aInstance);
+      }while(lOptions->isWaiting());
+    }
   }
 
   if (aValues->next(lValue))
@@ -1005,19 +1121,23 @@
   if (thePaths->next(lPath))
   {
     ViewOptions* lOptions = &theOptions;
-    
+    String lPathOptions = lOptions->getPathOptions();
+    String lPathString = lPath.getStringValue();    
+    if (lPathOptions != "")
+    {
+      lPathString.append(lPathOptions);
+    }    
     lcb_http_request_t lReq;
     lcb_http_cmd_t lCmd;
     lCmd.version = 0;
-    lCmd.v.v0.path = lPath.getStringValue().c_str();
-    lCmd.v.v0.npath = lPath.getStringValue().size();
+    lCmd.v.v0.path = lPathString.c_str();
+    lCmd.v.v0.npath = lPathString.size();
     lCmd.v.v0.body = NULL;
     lCmd.v.v0.nbody = 0;
     lCmd.v.v0.method = LCB_HTTP_METHOD_GET;
     lCmd.v.v0.chunked = 1;
     lCmd.v.v0.content_type = "application/json";
     lcb_error_t err = lcb_make_http_request(theInstance, lOptions, LCB_HTTP_TYPE_VIEW, &lCmd, &lReq);
-  
     if (err != LCB_SUCCESS)
     {
       libCouchbaseError (theInstance, err);

=== modified file 'src/couchbase.xq.src/couchbase.h'
--- src/couchbase.xq.src/couchbase.h	2013-01-10 06:51:21 +0000
+++ src/couchbase.xq.src/couchbase.h	2013-01-14 21:43:21 +0000
@@ -86,16 +86,25 @@
 
     } lcb_storage_type_t;
 
+    typedef enum
+    {
+      CB_WAIT_FALSE = 0x00,
+      CB_WAIT_PERSIST = 0x01,
+      CB_WAIT_REPLICATE = 0x02
+    } cb_wait_type_t;
+
     class ViewOptions
     {
       protected:
         String theEncoding;
         String thePath;
+        String theStaleOption;
+        String theLimitOption;
 
       public:
         std::unique_ptr<std::stringstream>* theStream;
 
-        ViewOptions() : theEncoding("UTF-8"), thePath("") { theStream = NULL; }
+        ViewOptions() : theEncoding("UTF-8"), thePath(""), theStaleOption(""), theLimitOption("") { theStream = NULL; }
 
         ViewOptions(String& aPath) : theEncoding("UTF-8"), thePath(aPath) {}
 
@@ -106,6 +115,8 @@
         String getEncoding() { return theEncoding; }
 
         String getPath() { return thePath; }
+
+        String getPathOptions();
     };
 
     class GetOptions
@@ -137,17 +148,18 @@
     class PutOptions
     {
       protected:
-        
         lcb_storage_t theOperation;
         lcb_storage_type_t theType;
         unsigned int theExpTime;
         String theEncoding;
+        cb_wait_type_t theWaitType;
+        bool theIsWaiting;
 
       public:
 
-        PutOptions() : theOperation(LCB_ADD), theType(LCB_JSON), theExpTime(0), theEncoding("") { }
+        PutOptions() : theOperation(LCB_ADD), theType(LCB_JSON), theExpTime(0), theEncoding(""), theWaitType(CB_WAIT_FALSE), theIsWaiting(false){ }
 
-        PutOptions(lcb_storage_type_t aType) : theOperation(LCB_SET), theType(aType), theExpTime(0) { }
+        PutOptions(lcb_storage_type_t aType) : theOperation(LCB_SET), theType(aType), theExpTime(0), theWaitType(CB_WAIT_FALSE), theIsWaiting(false) { }
 
         void setOptions(Item& aOptions);
 
@@ -160,6 +172,12 @@
         unsigned int getExmpTime() { return theExpTime; }
 
         String getEncoding() { return theEncoding; }
+
+        cb_wait_type_t getWaitType() { return theWaitType; }
+
+        bool isWaiting() { return theIsWaiting; }
+
+        void setWaiting(bool isWaiting) { theIsWaiting = isWaiting; }
     };
 
     class ViewItemSequence : public ItemSequence
@@ -299,6 +317,14 @@
     static void
       put (lcb_t aInstance, Iterator_t aKeys, Iterator_t aValues, PutOptions aOptions);
 
+    
+    static void
+      observe_callback(
+        lcb_t instance, 
+        const void *cookie,
+        lcb_error_t error,
+        const lcb_observe_resp_t *resp);
+
   public:
     
     CouchbaseFunction(const CouchbaseModule* module);

=== modified file 'test/Queries/couchbase_module/view.xq'
--- test/Queries/couchbase_module/view.xq	2013-01-10 07:59:35 +0000
+++ test/Queries/couchbase_module/view.xq	2013-01-14 21:43:21 +0000
@@ -6,12 +6,14 @@
   "password" : null,
   "bucket" : "default"});
 
-cb:put-text($instance, "view", '{ "view" : 1 }');
+cb:remove($instance, "view");
+cb:put-text($instance, "view", '{ "view" : 1 }', { "wait" : "persist" });
 
 variable $view-name := cb:create-view($instance, "test-view", "test", {"key":"doc.view"});
 
-variable $data := cb:view($instance, $view-name);
+
+variable $data := cb:view($instance, $view-name, {"stale" : "false"});
 for $d in jn:members($data("rows"))
-where $d("key") >0
+where $d("key") > 0
 return $d