Merge pull request #7796 from jamienoss/issue14109-Redis-plugin-change-timeout-functionality-alpha-rebase

HPCC-14109 Redis plugin - extend timeouts to duration of plugin calls

Reviewed-by: Gavin Halliday <ghalliday@hpccsystems.com>

plugins/redis/README.md

@@ -45,7 +45,7 @@ The Actual Plugin
 -----------------
 
 The bulk of this redis plugin for **ECL** is made up of the various `SET` and `GET` commands e.g. `GetString` or `SetReal`. They are accessible via the module `redis`
-from the redis plugin **ECL** library `lib-redis`. i.e.
+from the redis plugin **ECL** [library](https://github.com/hpcc-systems/HPCC-Platform/blob/master/plugins/redis/lib_redis.ecllib) `lib-redis`. i.e.
 ```
 IMPORT redis FROM lib_redis;
 ```
@@ -91,8 +91,7 @@ The core points to note here are:
    STRING of length 8, set with SetString, being successfully retrieved from the cache via GetInteger without an **ECL** exception being thrown.
    * `CONST VARSTRING options` passes the server **IP** and **port** to the plugin in the *strict* format - `--SERVER=<ip>:<port>`. If `options` is empty, the default
    127.0.0.1:6379 is used. *Note:* 6379 is the default port for **redis-server**.
-   * `UNSIGNED timeout` has units *ms* and has a default value of 1 second. This is not a timeout duration for an entire plugin call but rather that set for each
-   communication transaction with the redis server. *c.f.* 'Behaviour and Implementation Details' below.
+   * `UNSIGNED timeout` has units *ms* and has a default value of 1 second (0 := infinity).  *c.f.* 'Timeout Values' below for advice on choosing appropriate values.
    * `UNSIGNED expire` has units *ms* and a default of **0**, i.e. *forever*.
 
 ###The redisServer MODULE
@@ -106,7 +105,7 @@ myRedis.GetString('myKey');
 ```
 
 ###A Redis 'Database'
-The notion of a *database* within a redis cache is a that of an UNSIGNED INTEGER, effectively partitioning the cache such that it may contain an identical key per database e.g.
+The notion of a *database* within a redis cache is that of a partition, such that it may contain an identical key per database e.g.
 ```
 myRedis.SetString('myKey', 'foo', 0);
 myRedis.SetString('myKey', 'bar', 1);
@@ -114,7 +113,8 @@ myRedis.SetString('myKey', 'bar', 1);
 myRedis.GetString('myKey', 0);//returns 'foo'
 myRedis.GetString('myKey', 1);//returns 'bar'
 ```
-*Note:* that the default database is 0.
+
+*Note:* that the default database is 0. The maximum number of databases allowed by **Redis** is 2147483647 (int32).
 
 
 Race Retrieval and Locking Keys
@@ -152,12 +152,27 @@ SEQUENTIAL(
     );
 ```
 
+*Note:* further **ECL** examples can be found in the following files regarding the [locking](https://github.com/hpcc-systems/HPCC-Platform/blob/master/testing/regress/ecl/redislockingtest.ecl) and [non-locking](https://github.com/hpcc-systems/HPCC-Platform/blob/master/testing/regress/ecl/redissynctest.ecl) functions.
+
+###Timeout Values
+The timeout durations are effectively for the entire duration of a call to each of the functions exported by this plugin library. By 'effectively', it is meant that a timer is
+initiated at the start of each call and upon each internal communication with the redis server, any time remaining (at this point) is the timeout value passed to the redis API
+(hiredis) for that communication call. Since some plugin functions make more calls to the server than others (*c.f.* 'Behaviour and Implementation Details' below) it is
+possible for those functions with more server calls to timeout more regularly than those with less. To avoid this, it is advised to set the timeouts to a multiple of the
+anticipated latency of the client-server-IO, where such multiple should be at least the maximum expected number of internal redis calls made by these plugin functions, e.g. 12.
+
+When using the **ECL** pattern described in the above section *An ECL Example*, it is required to set the timeout and lock expiration to be equal to the timeout (if any)
+of `myFunc` **+** that passed to `SetAndPublish<type>`, such that both the lock and waiting subscribers live long enough for a value to be set/published.
+
+It should also be noted that, whilst it is possible to set different values for `timeout` and `expire` to the function `GetOrLock<type>`, it is advisable not to.
+This is such that the lock does not out live all waiting subscribers that collectively timeout and thus not blocking any subsequent retries of the locking pattern.
+
 Behaviour and Implementation Details
 ------------------------------------
 A few notes to point out here:
    * PUB-SUB channels are not disconnected from the keyspace as they are in their native redis usage. The key itself is used as the lock with its value being set as the channel to later
-   PUBLISH on or SUBSCRIBE to. This channel is a string, unique by only the *key* and *database*, prefixed with **'redis_ecl_lock'**.
-   * The lock itself is set to expire with a duration equal to the `timeout` value passed to the `GetOrLock<type>` function (default 1s).
+   PUBLISH on or SUBSCRIBE to. This channel is a string, unique by only the *key* and *database*, prefixed with **'redis_ecl_lock'**. E.g. the *key* 'myKey' designated for *database* 1,
+   will have the following lock id - 'redis_ecl_lock_myKey_1'.
    * It is possible to manually 'unlock' this lock (`DELETE` the key) via the `Unlock(<key>)` function. *Note:* this function will fail on any communication or reply error however,
    it will **silently fail**, leaving the lock to expire, if the server observes any change to the key during the function call duration.
    * When the *race-winner* publishes, it actually publishes the value itself and that any subscriber will then obtain the key-value in this fashion. Therefore, not requiring an
@@ -165,10 +180,8 @@ A few notes to point out here:
     have the key-value received on another, and yet, the key-value still does not exist on the cache.
    * At present the 'lock' is not as such an actual lock, as only the locking functions acknowledge it. By current implementation it is better thought as a flag for
    `GET` to wait and subscribe. I.e. the locked key can be deleted and re-set just as any other key can be.
-   * Since the timeout duration is not for an individual plugin call but instead that waiting for each reply from the server, the actual possible maximum timeout duration differs from
-     various functions within this plugin, i.e. some functions do more than others. Below is a table for each of the plugin functions (or categories of) including the maximum possible and
-     nominal expected, where the latter is due to using a cached connection, i.e. neither the server IP, port, nor password have changed from the function called prior to the one in
-     question. The values given are multiples of the given timeout.
+   * Below is a table of the number of calls to the redis server for each of the plugin functions (or categories of) including the maximum possible and nominal expected, where the
+   latter is due to using a cached connection, i.e. neither the server IP, port, nor password have changed from the function called prior to the one in question.
 
 | Operation/Function  | Nominal | Maximum | Diff due to...   |
 |:--------------------|:-------:|:-------:|-----------------:|

plugins/redis/lib_redis.ecllib

@@ -17,77 +17,115 @@
 
 
 EXPORT redis := SERVICE : plugin('redis'), namespace('RedisPlugin')
-  SetUnicode( CONST VARSTRING key, CONST UNICODE value, CONST VARSTRING options, UNSIGNED database = 0, UNSIGNED4 expire = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,action,context,entrypoint='SyncRSetUChar';
-  SetString(  CONST VARSTRING key, CONST STRING value,  CONST VARSTRING options, UNSIGNED database = 0, UNSIGNED4 expire = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,action,context,entrypoint='SyncRSetStr';
-  SetUtf8(    CONST VARSTRING key, CONST UTF8 value,    CONST VARSTRING options, UNSIGNED database = 0, UNSIGNED4 expire = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,action,context,entrypoint='SyncRSetUtf8';
-  SetBoolean( CONST VARSTRING key, BOOLEAN value,       CONST VARSTRING options, UNSIGNED database = 0, UNSIGNED4 expire = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,action,context,entrypoint='SyncRSetBool';
-  SetReal(    CONST VARSTRING key, REAL value,          CONST VARSTRING options, UNSIGNED database = 0, UNSIGNED4 expire = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,action,context,entrypoint='SyncRSetReal';
-  SetInteger( CONST VARSTRING key, INTEGER value,       CONST VARSTRING options, UNSIGNED database = 0, UNSIGNED4 expire = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,action,context,entrypoint='SyncRSetInt';
-  SetUnsigned(CONST VARSTRING key, UNSIGNED value,      CONST VARSTRING options, UNSIGNED database = 0, UNSIGNED4 expire = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,action,context,entrypoint='SyncRSetUInt';
-  SetData(    CONST VARSTRING key, CONST DATA value,    CONST VARSTRING options, UNSIGNED database = 0, UNSIGNED4 expire = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,action,context,entrypoint='SyncRSetData';
-
-  INTEGER8   GetInteger(CONST VARSTRING key, CONST VARSTRING options, UNSIGNED database = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,once,context,entrypoint='SyncRGetInt8';
-  UNSIGNED8 GetUnsigned(CONST VARSTRING key, CONST VARSTRING options, UNSIGNED database = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,once,context,entrypoint='SyncRGetUint8';
-  STRING      GetString(CONST VARSTRING key, CONST VARSTRING options, UNSIGNED database = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,once,context,entrypoint='SyncRGetStr';
-  UNICODE    GetUnicode(CONST VARSTRING key, CONST VARSTRING options, UNSIGNED database = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,once,context,entrypoint='SyncRGetUChar';
-  UTF8          GetUtf8(CONST VARSTRING key, CONST VARSTRING options, UNSIGNED database = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,once,context,entrypoint='SyncRGetUtf8';
-  BOOLEAN    GetBoolean(CONST VARSTRING key, CONST VARSTRING options, UNSIGNED database = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,once,context,entrypoint='SyncRGetBool';
-  REAL          GetReal(CONST VARSTRING key, CONST VARSTRING options, UNSIGNED database = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,once,context,entrypoint='SyncRGetDouble';
-  DATA          GetData(CONST VARSTRING key, CONST VARSTRING options, UNSIGNED database = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,once,context,entrypoint='SyncRGetData';
-
-  BOOLEAN Exists(CONST VARSTRING key, CONST VARSTRING options, UNSIGNED database = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,once,context,entrypoint='RExist';
-  FlushDB(CONST VARSTRING options, UNSIGNED database = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,action,context,entrypoint='RClear';
-  Delete(CONST VARSTRING key, CONST VARSTRING options, UNSIGNED database = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,action,context,entrypoint='RDel';
-  Del(CONST VARSTRING key, CONST VARSTRING options, UNSIGNED database = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,action,context,entrypoint='RDel';
-  Persist(CONST VARSTRING key, CONST VARSTRING options, UNSIGNED database = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,action,context,entrypoint='RPersist';
-  Expire(CONST VARSTRING key, CONST VARSTRING options, UNSIGNED database = 0, UNSIGNED4 expire, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,action,context,entrypoint='RExpire';
-  INTEGER DBSize(CONST VARSTRING options, UNSIGNED database = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,once,context,entrypoint='RDBSize';
-
-  STRING   SetAndPublishString(  CONST VARSTRING key, CONST STRING value,  CONST VARSTRING options, UNSIGNED database = 0, UNSIGNED4 expire = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,once,context,entrypoint='SyncLockRSetStr';
-  UNICODE  SetAndPublishUnicode( CONST VARSTRING key, CONST UNICODE value, CONST VARSTRING options, UNSIGNED database = 0, UNSIGNED4 expire = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,once,context,entrypoint='SyncLockRSetUChar';
-  UTF8     SetAndPublishUtf8(    CONST VARSTRING key, CONST UTF8 value,    CONST VARSTRING options, UNSIGNED database = 0, UNSIGNED4 expire = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,once,context,entrypoint='SyncLockRSetUtf8';
-
-  STRING      GetOrLockString(CONST VARSTRING key, CONST VARSTRING options, UNSIGNED database = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000, UNSIGNED4 expire = 1000) : cpp,once,context,entrypoint='SyncLockRGetStr';
-  UNICODE    GetOrLockUnicode(CONST VARSTRING key, CONST VARSTRING options, UNSIGNED database = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000, UNSIGNED4 expire = 1000) : cpp,once,context,entrypoint='SyncLockRGetUChar';
-  UTF8          GetOrLockUtf8(CONST VARSTRING key, CONST VARSTRING options, UNSIGNED database = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000, UNSIGNED4 expire = 1000) : cpp,once,context,entrypoint='SyncLockRGetUtf8';
-
-  Unlock(CONST VARSTRING key, CONST VARSTRING options, UNSIGNED database = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,action,context,entrypoint='SyncLockRUnlock';
+  SetUnicode( CONST VARSTRING key, CONST UNICODE value, CONST VARSTRING options, INTEGER4 database = 0, UNSIGNED4 expire = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000) : cpp,action,context,entrypoint='SyncRSetUChar';
+  SetString(  CONST VARSTRING key, CONST STRING value,  CONST VARSTRING options, INTEGER4 database = 0, UNSIGNED4 expire = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000) : cpp,action,context,entrypoint='SyncRSetStr';
+  SetUtf8(    CONST VARSTRING key, CONST UTF8 value,    CONST VARSTRING options, INTEGER4 database = 0, UNSIGNED4 expire = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000) : cpp,action,context,entrypoint='SyncRSetUtf8';
+  SetBoolean( CONST VARSTRING key, BOOLEAN value,       CONST VARSTRING options, INTEGER4 database = 0, UNSIGNED4 expire = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000) : cpp,action,context,entrypoint='SyncRSetBool';
+  SetReal(    CONST VARSTRING key, REAL value,          CONST VARSTRING options, INTEGER4 database = 0, UNSIGNED4 expire = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000) : cpp,action,context,entrypoint='SyncRSetReal';
+  SetInteger( CONST VARSTRING key, INTEGER value,       CONST VARSTRING options, INTEGER4 database = 0, UNSIGNED4 expire = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000) : cpp,action,context,entrypoint='SyncRSetInt';
+  SetUnsigned(CONST VARSTRING key, UNSIGNED value,      CONST VARSTRING options, INTEGER4 database = 0, UNSIGNED4 expire = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000) : cpp,action,context,entrypoint='SyncRSetUInt';
+  SetData(    CONST VARSTRING key, CONST DATA value,    CONST VARSTRING options, INTEGER4 database = 0, UNSIGNED4 expire = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000) : cpp,action,context,entrypoint='SyncRSetData';
+
+  INTEGER8   GetInteger(CONST VARSTRING key, CONST VARSTRING options, INTEGER4 database = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000) : cpp,once,context,entrypoint='SyncRGetInt8';
+  UNSIGNED8 GetUnsigned(CONST VARSTRING key, CONST VARSTRING options, INTEGER4 database = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000) : cpp,once,context,entrypoint='SyncRGetUint8';
+  STRING      GetString(CONST VARSTRING key, CONST VARSTRING options, INTEGER4 database = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000) : cpp,once,context,entrypoint='SyncRGetStr';
+  UNICODE    GetUnicode(CONST VARSTRING key, CONST VARSTRING options, INTEGER4 database = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000) : cpp,once,context,entrypoint='SyncRGetUChar';
+  UTF8          GetUtf8(CONST VARSTRING key, CONST VARSTRING options, INTEGER4 database = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000) : cpp,once,context,entrypoint='SyncRGetUtf8';
+  BOOLEAN    GetBoolean(CONST VARSTRING key, CONST VARSTRING options, INTEGER4 database = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000) : cpp,once,context,entrypoint='SyncRGetBool';
+  REAL          GetReal(CONST VARSTRING key, CONST VARSTRING options, INTEGER4 database = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000) : cpp,once,context,entrypoint='SyncRGetDouble';
+  DATA          GetData(CONST VARSTRING key, CONST VARSTRING options, INTEGER4 database = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000) : cpp,once,context,entrypoint='SyncRGetData';
+
+  BOOLEAN Exists(CONST VARSTRING key, CONST VARSTRING options, INTEGER4 database = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000) : cpp,once,context,entrypoint='RExist';
+  FlushDB(CONST VARSTRING options, INTEGER4 database = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000) : cpp,action,context,entrypoint='RClear';
+  Delete(CONST VARSTRING key, CONST VARSTRING options, INTEGER4 database = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000) : cpp,action,context,entrypoint='RDel';
+  Del(CONST VARSTRING key, CONST VARSTRING options, INTEGER4 database = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000) : cpp,action,context,entrypoint='RDel';
+  Persist(CONST VARSTRING key, CONST VARSTRING options, INTEGER4 database = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000) : cpp,action,context,entrypoint='RPersist';
+  Expire(CONST VARSTRING key, CONST VARSTRING options, INTEGER4 database = 0, UNSIGNED4 expire, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000) : cpp,action,context,entrypoint='RExpire';
+  UNSIGNED DBSize(CONST VARSTRING options, INTEGER4 database = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000) : cpp,once,context,entrypoint='RDBSize';
+
+  STRING   SetAndPublishString(  CONST VARSTRING key, CONST STRING value,  CONST VARSTRING options, INTEGER4 database = 0, UNSIGNED4 expire = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000) : cpp,once,context,entrypoint='SyncLockRSetStr';
+  UNICODE  SetAndPublishUnicode( CONST VARSTRING key, CONST UNICODE value, CONST VARSTRING options, INTEGER4 database = 0, UNSIGNED4 expire = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000) : cpp,once,context,entrypoint='SyncLockRSetUChar';
+  UTF8     SetAndPublishUtf8(    CONST VARSTRING key, CONST UTF8 value,    CONST VARSTRING options, INTEGER4 database = 0, UNSIGNED4 expire = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000) : cpp,once,context,entrypoint='SyncLockRSetUtf8';
+
+  STRING      GetOrLockString(CONST VARSTRING key, CONST VARSTRING options, INTEGER4 database = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000, UNSIGNED4 expire = 1000) : cpp,once,context,entrypoint='SyncLockRGetStr';
+  UNICODE    GetOrLockUnicode(CONST VARSTRING key, CONST VARSTRING options, INTEGER4 database = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000, UNSIGNED4 expire = 1000) : cpp,once,context,entrypoint='SyncLockRGetUChar';
+  UTF8          GetOrLockUtf8(CONST VARSTRING key, CONST VARSTRING options, INTEGER4 database = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000, UNSIGNED4 expire = 1000) : cpp,once,context,entrypoint='SyncLockRGetUtf8';
+
+  Unlock(CONST VARSTRING key, CONST VARSTRING options, INTEGER4 database = 0, CONST VARSTRING password = '', UNSIGNED4 timeout = 1000) : cpp,action,context,entrypoint='SyncLockRUnlock';
 END;
 
-EXPORT RedisServer(VARSTRING options, VARSTRING password = '', UNSIGNED timeout = 1000) := MODULE
-  EXPORT  SetUnicode(VARSTRING key, UNICODE value,  UNSIGNED database = 0, UNSIGNED4 expire = 0) := redis.SetUnicode (key, value, options, database, expire, password, timeout);
-  EXPORT   SetString(VARSTRING key, STRING value,   UNSIGNED database = 0, UNSIGNED4 expire = 0) := redis.SetString  (key, value, options, database, expire, password, timeout);
-  EXPORT     SetUtf8(VARSTRING key, UTF8 value,     UNSIGNED database = 0, UNSIGNED4 expire = 0) := redis.SetUtf8    (key, value, options, database, expire, password, timeout);
-  EXPORT  SetBoolean(VARSTRING key, BOOLEAN value,  UNSIGNED database = 0, UNSIGNED4 expire = 0) := redis.SetBoolean (key, value, options, database, expire, password, timeout);
-  EXPORT     SetReal(VARSTRING key, REAL value,     UNSIGNED database = 0, UNSIGNED4 expire = 0) := redis.SetReal    (key, value, options, database, expire, password, timeout);
-  EXPORT  SetInteger(VARSTRING key, INTEGER value,  UNSIGNED database = 0, UNSIGNED4 expire = 0) := redis.SetInteger (key, value, options, database, expire, password, timeout);
-  EXPORT SetUnsigned(VARSTRING key, UNSIGNED value, UNSIGNED database = 0, UNSIGNED4 expire = 0) := redis.SetUnsigned(key, value, options, database, expire, password, timeout);
-  EXPORT     SetData(VARSTRING key, DATA value,     UNSIGNED database = 0, UNSIGNED4 expire = 0) := redis.SetData    (key, value, options, database, expire, password, timeout);
-
-  EXPORT  GetUnicode(VARSTRING key, UNSIGNED database = 0) :=  redis.GetUnicode(key, options, database, password, timeout);
-  EXPORT   GetString(VARSTRING key, UNSIGNED database = 0) :=   redis.GetString(key, options, database, password, timeout);
-  EXPORT     GetUtf8(VARSTRING key, UNSIGNED database = 0) :=     redis.GetUtf8(key, options, database, password, timeout);
-  EXPORT  GetBoolean(VARSTRING key, UNSIGNED database = 0) :=  redis.GetBoolean(key, options, database, password, timeout);
-  EXPORT     GetReal(VARSTRING key, UNSIGNED database = 0) :=     redis.GetReal(key, options, database, password, timeout);
-  EXPORT  GetInteger(VARSTRING key, UNSIGNED database = 0) :=  redis.GetInteger(key, options, database, password, timeout);
-  EXPORT GetUnsigned(VARSTRING key, UNSIGNED database = 0) := redis.GetUnsigned(key, options, database, password, timeout);
-  EXPORT     GetData(VARSTRING key, UNSIGNED database = 0) :=     redis.GetData(key, options, database, password, timeout);
-
-  EXPORT Exists(VARSTRING key, UNSIGNED database = 0) := redis.Exists(key, options, database, password, timeout);
-  EXPORT FlushDB(UNSIGNED database = 0) := redis.FlushDB(options, database, password, timeout);
-  EXPORT Delete(VARSTRING key, UNSIGNED database = 0) := redis.Delete(key, options, database, password, timeout);
-  EXPORT Del(VARSTRING key, UNSIGNED database = 0) := redis.Del(key, options, database, password, timeout);
-  EXPORT Persist(VARSTRING key, UNSIGNED database = 0) := redis.Persist(key, options, database, password, timeout);
-  EXPORT Expire(VARSTRING key, UNSIGNED database = 0, UNSIGNED4 expire)  := redis.Expire(key, options, database, expire, password, timeout);
-  EXPORT DBSize(UNSIGNED database = 0) := redis.DBSize(options, database, password, timeout);
-
-  EXPORT  SetAndPublishUnicode(VARSTRING key, UNICODE value,  UNSIGNED database = 0, UNSIGNED4 expire = 0) := redis.SetAndPublishUnicode (key, value, options, database, expire, password, timeout);
-  EXPORT   SetAndPublishString(VARSTRING key, STRING value,   UNSIGNED database = 0, UNSIGNED4 expire = 0) := redis.SetAndPublishString  (key, value, options, database, expire, password, timeout);
-  EXPORT     SetAndPublishUtf8(VARSTRING key, UTF8 value,     UNSIGNED database = 0, UNSIGNED4 expire = 0) := redis.SetAndPublishUtf8    (key, value, options, database, expire, password, timeout);
-
-  EXPORT  GetOrLockUnicode(VARSTRING key, UNSIGNED database = 0, UNSIGNED4 expire = 1000) :=  redis.GetOrLockUnicode(key, options, database, password, timeout, expire);
-  EXPORT   GetOrLockString(VARSTRING key, UNSIGNED database = 0, UNSIGNED4 expire = 1000) :=   redis.GetOrLockString(key, options, database, password, timeout, expire);
-  EXPORT     GetOrLockUtf8(VARSTRING key, UNSIGNED database = 0, UNSIGNED4 expire = 1000) :=     redis.GetOrLockUtf8(key, options, database, password, timeout, expire);
-
-  EXPORT Unlock(VARSTRING key, UNSIGNED database = 0) := redis.Unlock(key, options, database, password, timeout);
+EXPORT RedisServer(VARSTRING options, VARSTRING password = '', UNSIGNED4 timeout = 1000) := MODULE
+  EXPORT  SetUnicode(VARSTRING key, UNICODE value,  INTEGER4 database = 0, UNSIGNED4 expire = 0) := redis.SetUnicode (key, value, options, database, expire, password, timeout);
+  EXPORT   SetString(VARSTRING key, STRING value,   INTEGER4 database = 0, UNSIGNED4 expire = 0) := redis.SetString  (key, value, options, database, expire, password, timeout);
+  EXPORT     SetUtf8(VARSTRING key, UTF8 value,     INTEGER4 database = 0, UNSIGNED4 expire = 0) := redis.SetUtf8    (key, value, options, database, expire, password, timeout);
+  EXPORT  SetBoolean(VARSTRING key, BOOLEAN value,  INTEGER4 database = 0, UNSIGNED4 expire = 0) := redis.SetBoolean (key, value, options, database, expire, password, timeout);
+  EXPORT     SetReal(VARSTRING key, REAL value,     INTEGER4 database = 0, UNSIGNED4 expire = 0) := redis.SetReal    (key, value, options, database, expire, password, timeout);
+  EXPORT  SetInteger(VARSTRING key, INTEGER value,  INTEGER4 database = 0, UNSIGNED4 expire = 0) := redis.SetInteger (key, value, options, database, expire, password, timeout);
+  EXPORT SetUnsigned(VARSTRING key, UNSIGNED value, INTEGER4 database = 0, UNSIGNED4 expire = 0) := redis.SetUnsigned(key, value, options, database, expire, password, timeout);
+  EXPORT     SetData(VARSTRING key, DATA value,     INTEGER4 database = 0, UNSIGNED4 expire = 0) := redis.SetData    (key, value, options, database, expire, password, timeout);
+
+  EXPORT  GetUnicode(VARSTRING key, INTEGER4 database = 0) :=  redis.GetUnicode(key, options, database, password, timeout);
+  EXPORT   GetString(VARSTRING key, INTEGER4 database = 0) :=   redis.GetString(key, options, database, password, timeout);
+  EXPORT     GetUtf8(VARSTRING key, INTEGER4 database = 0) :=     redis.GetUtf8(key, options, database, password, timeout);
+  EXPORT  GetBoolean(VARSTRING key, INTEGER4 database = 0) :=  redis.GetBoolean(key, options, database, password, timeout);
+  EXPORT     GetReal(VARSTRING key, INTEGER4 database = 0) :=     redis.GetReal(key, options, database, password, timeout);
+  EXPORT  GetInteger(VARSTRING key, INTEGER4 database = 0) :=  redis.GetInteger(key, options, database, password, timeout);
+  EXPORT GetUnsigned(VARSTRING key, INTEGER4 database = 0) := redis.GetUnsigned(key, options, database, password, timeout);
+  EXPORT     GetData(VARSTRING key, INTEGER4 database = 0) :=     redis.GetData(key, options, database, password, timeout);
+
+  EXPORT Exists(VARSTRING key, INTEGER4 database = 0) := redis.Exists(key, options, database, password, timeout);
+  EXPORT FlushDB(INTEGER4 database = 0) := redis.FlushDB(options, database, password, timeout);
+  EXPORT Delete(VARSTRING key, INTEGER4 database = 0) := redis.Delete(key, options, database, password, timeout);
+  EXPORT Del(VARSTRING key, INTEGER4 database = 0) := redis.Del(key, options, database, password, timeout);
+  EXPORT Persist(VARSTRING key, INTEGER4 database = 0) := redis.Persist(key, options, database, password, timeout);
+  EXPORT Expire(VARSTRING key, INTEGER4 database = 0, UNSIGNED4 expire)  := redis.Expire(key, options, database, expire, password, timeout);
+  EXPORT DBSize(INTEGER4 database = 0) := redis.DBSize(options, database, password, timeout);
+
+  EXPORT  SetAndPublishUnicode(VARSTRING key, UNICODE value,  INTEGER4 database = 0, UNSIGNED4 expire = 0) := redis.SetAndPublishUnicode (key, value, options, database, expire, password, timeout);
+  EXPORT   SetAndPublishString(VARSTRING key, STRING value,   INTEGER4 database = 0, UNSIGNED4 expire = 0) := redis.SetAndPublishString  (key, value, options, database, expire, password, timeout);
+  EXPORT     SetAndPublishUtf8(VARSTRING key, UTF8 value,     INTEGER4 database = 0, UNSIGNED4 expire = 0) := redis.SetAndPublishUtf8    (key, value, options, database, expire, password, timeout);
+
+  EXPORT  GetOrLockUnicode(VARSTRING key, INTEGER4 database = 0, UNSIGNED4 expire = 1000) :=  redis.GetOrLockUnicode(key, options, database, password, timeout, expire);
+  EXPORT   GetOrLockString(VARSTRING key, INTEGER4 database = 0, UNSIGNED4 expire = 1000) :=   redis.GetOrLockString(key, options, database, password, timeout, expire);
+  EXPORT     GetOrLockUtf8(VARSTRING key, INTEGER4 database = 0, UNSIGNED4 expire = 1000) :=     redis.GetOrLockUtf8(key, options, database, password, timeout, expire);
+
+  EXPORT Unlock(VARSTRING key, INTEGER4 database = 0) := redis.Unlock(key, options, database, password, timeout);
+END;
+
+EXPORT RedisServerWithoutTimeout(VARSTRING options, VARSTRING password = '') := MODULE
+  EXPORT  SetUnicode(VARSTRING key, UNICODE value,  INTEGER4 database = 0, UNSIGNED4 expire = 0, UNSIGNED4 timeout = 1000) := redis.SetUnicode (key, value, options, database, expire, password, timeout);
+  EXPORT   SetString(VARSTRING key, STRING value,   INTEGER4 database = 0, UNSIGNED4 expire = 0, UNSIGNED4 timeout = 1000) := redis.SetString  (key, value, options, database, expire, password, timeout);
+  EXPORT     SetUtf8(VARSTRING key, UTF8 value,     INTEGER4 database = 0, UNSIGNED4 expire = 0, UNSIGNED4 timeout = 1000) := redis.SetUtf8    (key, value, options, database, expire, password, timeout);
+  EXPORT  SetBoolean(VARSTRING key, BOOLEAN value,  INTEGER4 database = 0, UNSIGNED4 expire = 0, UNSIGNED4 timeout = 1000) := redis.SetBoolean (key, value, options, database, expire, password, timeout);
+  EXPORT     SetReal(VARSTRING key, REAL value,     INTEGER4 database = 0, UNSIGNED4 expire = 0, UNSIGNED4 timeout = 1000) := redis.SetReal    (key, value, options, database, expire, password, timeout);
+  EXPORT  SetInteger(VARSTRING key, INTEGER value,  INTEGER4 database = 0, UNSIGNED4 expire = 0, UNSIGNED4 timeout = 1000) := redis.SetInteger (key, value, options, database, expire, password, timeout);
+  EXPORT SetUnsigned(VARSTRING key, UNSIGNED value, INTEGER4 database = 0, UNSIGNED4 expire = 0, UNSIGNED4 timeout = 1000) := redis.SetUnsigned(key, value, options, database, expire, password, timeout);
+  EXPORT     SetData(VARSTRING key, DATA value,     INTEGER4 database = 0, UNSIGNED4 expire = 0, UNSIGNED4 timeout = 1000) := redis.SetData    (key, value, options, database, expire, password, timeout);
+
+  EXPORT  GetUnicode(VARSTRING key, INTEGER4 database = 0, UNSIGNED4 timeout = 1000) :=  redis.GetUnicode(key, options, database, password, timeout);
+  EXPORT   GetString(VARSTRING key, INTEGER4 database = 0, UNSIGNED4 timeout = 1000) :=   redis.GetString(key, options, database, password, timeout);
+  EXPORT     GetUtf8(VARSTRING key, INTEGER4 database = 0, UNSIGNED4 timeout = 1000) :=     redis.GetUtf8(key, options, database, password, timeout);
+  EXPORT  GetBoolean(VARSTRING key, INTEGER4 database = 0, UNSIGNED4 timeout = 1000) :=  redis.GetBoolean(key, options, database, password, timeout);
+  EXPORT     GetReal(VARSTRING key, INTEGER4 database = 0, UNSIGNED4 timeout = 1000) :=     redis.GetReal(key, options, database, password, timeout);
+  EXPORT  GetInteger(VARSTRING key, INTEGER4 database = 0, UNSIGNED4 timeout = 1000) :=  redis.GetInteger(key, options, database, password, timeout);
+  EXPORT GetUnsigned(VARSTRING key, INTEGER4 database = 0, UNSIGNED4 timeout = 1000) := redis.GetUnsigned(key, options, database, password, timeout);
+  EXPORT     GetData(VARSTRING key, INTEGER4 database = 0, UNSIGNED4 timeout = 1000) :=     redis.GetData(key, options, database, password, timeout);
+
+  EXPORT Exists(VARSTRING key, INTEGER4 database = 0, UNSIGNED4 timeout = 1000) := redis.Exists(key, options, database, password, timeout);
+  EXPORT FlushDB(INTEGER4 database = 0, UNSIGNED4 timeout = 1000) := redis.FlushDB(options, database, password, timeout);
+  EXPORT Delete(VARSTRING key, INTEGER4 database = 0, UNSIGNED4 timeout = 1000) := redis.Delete(key, options, database, password, timeout);
+  EXPORT Del(VARSTRING key, INTEGER4 database = 0, UNSIGNED4 timeout = 1000) := redis.Del(key, options, database, password, timeout);
+  EXPORT Persist(VARSTRING key, INTEGER4 database = 0, UNSIGNED4 timeout = 1000) := redis.Persist(key, options, database, password, timeout);
+  EXPORT Expire(VARSTRING key, INTEGER4 database = 0, UNSIGNED4 expire, UNSIGNED4 timeout = 1000)  := redis.Expire(key, options, database, expire, password, timeout);
+  EXPORT DBSize(INTEGER4 database = 0, UNSIGNED4 timeout = 1000) := redis.DBSize(options, database, password, timeout);
+
+  EXPORT  SetAndPublishUnicode(VARSTRING key, UNICODE value,  INTEGER4 database = 0, UNSIGNED4 expire = 0, UNSIGNED4 timeout = 1000) := redis.SetAndPublishUnicode (key, value, options, database, expire, password, timeout);
+  EXPORT   SetAndPublishString(VARSTRING key, STRING value,   INTEGER4 database = 0, UNSIGNED4 expire = 0, UNSIGNED4 timeout = 1000) := redis.SetAndPublishString  (key, value, options, database, expire, password, timeout);
+  EXPORT     SetAndPublishUtf8(VARSTRING key, UTF8 value,     INTEGER4 database = 0, UNSIGNED4 expire = 0, UNSIGNED4 timeout = 1000) := redis.SetAndPublishUtf8    (key, value, options, database, expire, password, timeout);
+
+  EXPORT  GetOrLockUnicode(VARSTRING key, INTEGER4 database = 0, UNSIGNED4 expire = 1000, UNSIGNED4 timeout = 1000) :=  redis.GetOrLockUnicode(key, options, database, password, timeout, expire);
+  EXPORT   GetOrLockString(VARSTRING key, INTEGER4 database = 0, UNSIGNED4 expire = 1000, UNSIGNED4 timeout = 1000) :=   redis.GetOrLockString(key, options, database, password, timeout, expire);
+  EXPORT     GetOrLockUtf8(VARSTRING key, INTEGER4 database = 0, UNSIGNED4 expire = 1000, UNSIGNED4 timeout = 1000) :=     redis.GetOrLockUtf8(key, options, database, password, timeout, expire);
+
+  EXPORT Unlock(VARSTRING key, INTEGER4 database = 0, UNSIGNED4 timeout = 1000) := redis.Unlock(key, options, database, password, timeout);
 END;

plugins/redis/redis.cpp

@@ -81,17 +81,36 @@ private :
 };
 typedef Owned<RedisPlugin::Reply> OwnedReply;
 
+class TimeoutHandler
+{
+public :
+    TimeoutHandler(unsigned _timeout) : timeout(_timeout), t0(msTick()) { }
+    inline void reset(unsigned _timeout) { timeout = _timeout; t0 = msTick(); }
+    unsigned timeLeft() const
+    {
+        unsigned dt = msTick() - t0;
+        if (dt < timeout)
+            return timeout - dt;
+        return 0;
+    }
+    inline unsigned getTimeout() { return timeout; }
+
+private :
+    unsigned timeout;
+    unsigned t0;
+};
+
 class Connection : public CInterface
 {
 public :
-    Connection(ICodeContext * ctx, const char * options, unsigned __int64 database, const char * password, unsigned __int64 _timeout);
-    Connection(ICodeContext * ctx, const char * _options, const char * _ip, int _port, unsigned _serverIpPortPasswordHash, unsigned __int64 _database, const char * password, unsigned __int64 _timeout);
+    Connection(ICodeContext * ctx, const char * options, int database, const char * password, unsigned _timeout);
+    Connection(ICodeContext * ctx, const char * _options, const char * _ip, int _port, unsigned _serverIpPortPasswordHash, int _database, const char * password, unsigned _timeout);
     ~Connection()
     {
         if (context)
             redisFree(context);
     }
-    static Connection * createConnection(ICodeContext * ctx, const char * options, unsigned __int64 database, const char * password, unsigned __int64 _timeout);
+    static Connection * createConnection(ICodeContext * ctx, const char * options, int database, const char * password, unsigned _timeout);
 
     //set
     template <class type> void set(ICodeContext * ctx, const char * key, type value, unsigned expire);
@@ -114,21 +133,24 @@ public :
     bool exists(ICodeContext * ctx, const char * key);
 
 protected :
+    void redisSetTimeout();
+    void redisConnect();
+    unsigned timeLeft();
     void parseOptions(ICodeContext * ctx, const char * _options);
-    void connect(ICodeContext * ctx, unsigned __int64 _database, const char * password);
-    void selectDB(ICodeContext * ctx, unsigned __int64 _database);
+    void connect(ICodeContext * ctx, int _database, const char * password);
+    void selectDB(ICodeContext * ctx, int _database);
     void resetContextErr();
     void readReply(Reply * reply);
     void readReplyAndAssert(Reply * reply, const char * msg);
-    void readReplyAndAssertWithKey(Reply * reply, const char * msg, const char * key);
+    void readReplyAndAssertWithCmdMsg(Reply * reply, const char * msg, const char * key = NULL);
     void assertKey(const redisReply * reply, const char * key);
     void assertAuthorization(const redisReply * reply);
     void assertOnError(const redisReply * reply, const char * _msg);
-    void assertOnCommandError(const redisReply * reply, const char * cmd);
-    void assertOnCommandErrorWithDatabase(const redisReply * reply, const char * cmd);
-    void assertOnCommandErrorWithKey(const redisReply * reply, const char * cmd, const char * key);
-    void assertConnection();
-    void updateTimeout(unsigned __int64 _timeout);
+    void assertOnErrorWithCmdMsg(const redisReply * reply, const char * cmd, const char * key = NULL);
+    void assertConnection(const char * _msg);
+    void assertConnectionWithCmdMsg(const char * cmd, const char * key = NULL);
+    void fail(const char * cmd, const char * errmsg, const char * key = NULL);
+    void * redisCommand(redisContext * context, const char * format, ...);
     static unsigned hashServerIpPortPassword(ICodeContext * ctx, const char * _options, const char * password);
     bool isSameConnection(ICodeContext * ctx, const char * _options, const char * password) const;
 
@@ -146,8 +168,8 @@ protected :
     StringAttr ip;
     unsigned serverIpPortPasswordHash;
     int port;
-    unsigned __int64 timeout;
-    unsigned __int64 database;
+    TimeoutHandler timeout;
+    int database; //NOTE: redis stores the maximum number of dbs as an 'int'.
 };
 
 //The following class is here to ensure destruction of the cachedConnection within the main thread
@@ -180,26 +202,35 @@ static void releaseContext()
     }
     threadHooked = false;
 }
-Connection::Connection(ICodeContext * ctx, const char * _options, unsigned __int64 _database, const char * password, unsigned __int64 _timeout)
+Connection::Connection(ICodeContext * ctx, const char * _options, int _database, const char * password, unsigned _timeout)
   : database(0), timeout(_timeout), port(0), serverIpPortPasswordHash(hashServerIpPortPassword(ctx, _options, password))
 {
     options.set(_options, strlen(_options));
     parseOptions(ctx, _options);
     connect(ctx, _database, password);
 }
-Connection::Connection(ICodeContext * ctx, const char * _options, const char * _ip, int _port, unsigned _serverIpPortPasswordHash, unsigned __int64 _database, const char * password, unsigned __int64 _timeout)
+Connection::Connection(ICodeContext * ctx, const char * _options, const char * _ip, int _port, unsigned _serverIpPortPasswordHash, int _database, const char * password, unsigned _timeout)
   : database(0), timeout(_timeout), serverIpPortPasswordHash(_serverIpPortPasswordHash), port(_port)
 {
     options.set(_options, strlen(_options));
     ip.set(_ip, strlen(_ip));
     connect(ctx, _database, password);
 }
-void Connection::connect(ICodeContext * ctx, unsigned __int64 _database, const char * password)
+void Connection::redisConnect()
 {
-    struct timeval to = { timeout/1000, (timeout%1000)*1000 };
-    context = redisConnectWithTimeout(ip.str(), port, to);
-    assertConnection();
-    redisSetTimeout(context, to);
+    if (timeout.getTimeout() == 0)
+        context = ::redisConnect(ip.str(), port);
+    else
+    {
+        unsigned _timeLeft = timeLeft();
+        struct timeval to = { _timeLeft/1000, (_timeLeft%1000)*1000 };
+        context = ::redisConnectWithTimeout(ip.str(), port, to);
+    }
+    assertConnection("connection");
+}
+void Connection::connect(ICodeContext * ctx, int _database, const char * password)
+{
+    redisConnect();
 
     //The following is the dissemination of the two methods authenticate(ctx, password) & selectDB(ctx, _database)
     //such that they may be pipelined to save an extra round trip to the server and back.
@@ -208,21 +239,54 @@ void Connection::connect(ICodeContext * ctx, unsigned __int64 _database, const c
 
     if (database != _database)
     {
-        VStringBuffer cmd("SELECT %" I64F "u", _database);
+        VStringBuffer cmd("SELECT %d", _database);
         redisAppendCommand(context, cmd.str());
     }
 
     //Now read replies.
     OwnedReply reply = new Reply();
     if (password && *password)
-        readReplyAndAssert(reply, "server authentication failed");
+        readReplyAndAssert(reply, "server authentication");
 
     if (database != _database)
     {
-        readReplyAndAssert(reply, "request to SELECT database failed");
+        VStringBuffer cmd("SELECT %d", _database);
+        readReplyAndAssertWithCmdMsg(reply, cmd.str());
         database = _database;
     }
 }
+void * Connection::redisCommand(redisContext * context, const char * format, ...)
+{
+    //Copied from https://github.com/redis/hiredis/blob/master/hiredis.c ~line:1008 void * redisCommand(redisContext * context, const char * format, ...)
+    //with redisSetTimeout(); added.
+    va_list parameters;
+    void * reply = NULL;
+    va_start(parameters, format);
+    redisSetTimeout();
+    reply = ::redisvCommand(context, format, parameters);
+    va_end(parameters);
+    return reply;
+}
+unsigned Connection::timeLeft()
+{
+    unsigned _timeLeft = timeout.timeLeft();
+    if (_timeLeft == 0 && timeout.getTimeout() != 0)
+        ::rtlFail(0, "Redis Plugin: ERROR - function timed out internally.");
+    return _timeLeft;
+}
+void Connection::redisSetTimeout()
+{
+    unsigned _timeLeft = timeLeft();
+    if (_timeLeft == 0)
+        return;
+    struct timeval to = { _timeLeft/1000, (_timeLeft%1000)*1000 };
+    assertex(context);
+    if (::redisSetTimeout(context, to) != REDIS_OK)
+    {
+        assertConnection("request to set timeout");
+        throwUnexpected();//In case there is a bug in hiredis such that the above err is not reflected in the 'context' (checked in assertConnection) as expected.
+    }
+}
 bool Connection::isSameConnection(ICodeContext * ctx, const char * _options, const char * password) const
 {
     return (hashServerIpPortPassword(ctx, _options, password) == serverIpPortPasswordHash);
@@ -251,8 +315,8 @@ void Connection::parseOptions(ICodeContext * ctx, const char * _options)
         }
         else
         {
-            VStringBuffer err("Redis Plugin: unsupported option string %s", opt);
-            rtlFail(0, err.str());
+            VStringBuffer err("Redis Plugin: ERROR - unsupported option string '%s'", opt);
+            ::rtlFail(0, err.str());
         }
     }
     if (ip.isEmpty())
@@ -261,7 +325,7 @@ void Connection::parseOptions(ICodeContext * ctx, const char * _options)
         port = 6379;
         if (ctx)
         {
-            VStringBuffer msg("Redis Plugin: WARNING - using default server (%s:%d)", ip.str(), port);
+            VStringBuffer msg("Redis Plugin: WARNING - using default cache (%s:%d)", ip.str(), port);
             ctx->logString(msg.str());
         }
     }
@@ -274,23 +338,21 @@ void Connection::resetContextErr()
 void Connection::readReply(Reply * reply)
 {
     redisReply * nakedReply = NULL;
+    redisSetTimeout();
     redisGetReply(context, (void**)&nakedReply);
-    assertex(reply);
     reply->setClear(nakedReply);
 }
 void Connection::readReplyAndAssert(Reply * reply, const char * msg)
 {
     readReply(reply);
-    assertex(reply);
     assertOnError(reply->query(), msg);
 }
-void Connection::readReplyAndAssertWithKey(Reply * reply, const char * msg, const char * key)
+void Connection::readReplyAndAssertWithCmdMsg(Reply * reply, const char * msg, const char * key)
 {
     readReply(reply);
-    assertex(reply);
-    assertOnCommandErrorWithKey(reply->query(), msg, key);
+    assertOnErrorWithCmdMsg(reply->query(), msg, key);
 }
-Connection * Connection::createConnection(ICodeContext * ctx, const char * options, unsigned __int64 _database, const char * password, unsigned __int64 _timeout)
+Connection * Connection::createConnection(ICodeContext * ctx, const char * options, int _database, const char * password, unsigned _timeout)
 {
     if (!cachedConnection)
     {
@@ -306,9 +368,8 @@ Connection * Connection::createConnection(ICodeContext * ctx, const char * optio
     if (cachedConnection->isSameConnection(ctx, options, password))
     {
         //MORE: should perhaps check that the connection has not expired (think hiredis REDIS_KEEPALIVE_INTERVAL is defaulted to 15s).
-        //At present updateTimeout calls assertConnection.
         cachedConnection->resetContextErr();//reset the context err to allow reuse when an error previously occurred.
-        cachedConnection->updateTimeout(_timeout);
+        cachedConnection->timeout.reset(_timeout);
         cachedConnection->selectDB(ctx, _database);
         return LINK(cachedConnection);
     }
@@ -318,122 +379,80 @@ Connection * Connection::createConnection(ICodeContext * ctx, const char * optio
     cachedConnection = new Connection(ctx, options, _database, password, _timeout);
     return LINK(cachedConnection);
 }
-void Connection::selectDB(ICodeContext * ctx, unsigned __int64 _database)
+void Connection::selectDB(ICodeContext * ctx, int _database)
 {
     if (database == _database)
         return;
     database = _database;
-    VStringBuffer cmd("SELECT %" I64F "u", database);
+    VStringBuffer cmd("SELECT %d", database);
     OwnedReply reply = Reply::createReply(redisCommand(context, cmd.str()));
-    assertOnCommandError(reply->query(), "SELECT");
+    assertOnErrorWithCmdMsg(reply->query(), cmd.str());
 }
-void Connection::updateTimeout(unsigned __int64 _timeout)
+void Connection::fail(const char * cmd, const char * errmsg, const char * key)
 {
-    if (timeout == _timeout)
-        return;
-    assertConnection();
-    timeout = _timeout;
-    struct timeval to = { timeout/1000, (timeout%1000)*1000 };
-    assertex(context);
-    if (redisSetTimeout(context, to) != REDIS_OK)
+    if (key)
     {
-        if (context->err)
-        {
-            VStringBuffer msg("Redis Plugin: failed to set timeout - %s", context->errstr);
-            rtlFail(0, msg.str());
-        }
-        else
-            rtlFail(0, "Redis Plugin: failed to set timeout - no message available");
+        VStringBuffer msg("Redis Plugin: ERROR - %s '%s' on database %d for %s:%d failed : %s", cmd, key, database, ip.str(), port, errmsg);
+        ::rtlFail(0, msg.str());
     }
+    VStringBuffer msg("Redis Plugin: ERROR - %s on database %d for %s:%d failed : %s", cmd, database, ip.str(), port, errmsg);
+    ::rtlFail(0, msg.str());
 }
 void Connection::assertOnError(const redisReply * reply, const char * _msg)
 {
-    if (!reply)//MORE: should this be assertex(reply) instead?
-    {
-        //There should always be a context error if no reply error
-        assertConnection();
-        VStringBuffer msg("Redis Plugin: %s - %s", _msg, "neither 'reply' nor connection error available");
-        rtlFail(0, msg.str());
-    }
+    if (!reply)
+        assertConnection(_msg);
     else if (reply->type == REDIS_REPLY_ERROR)
     {
         assertAuthorization(reply);
         VStringBuffer msg("Redis Plugin: %s - %s", _msg, reply->str);
-        rtlFail(0, msg.str());
-    }
-}
-void Connection::assertOnCommandErrorWithKey(const redisReply * reply, const char * cmd, const char * key)
-{
-    if (!reply)//MORE: should this be assertex(reply) instead?
-    {
-        //There should always be a context error if no reply error
-        assertConnection();
-        VStringBuffer msg("Redis Plugin: ERROR - %s '%s' on database %" I64F "u failed with neither 'reply' nor connection error available", cmd, key, database);
-        rtlFail(0, msg.str());
-    }
-    else if (reply->type == REDIS_REPLY_ERROR)
-    {
-        assertAuthorization(reply);
-        VStringBuffer msg("Redis Plugin: ERROR - %s '%s' on database %" I64F "u failed : %s", cmd, key, database, reply->str);
-        rtlFail(0, msg.str());
-    }
-}
-void Connection::assertOnCommandErrorWithDatabase(const redisReply * reply, const char * cmd)
-{
-    if (!reply)//assertex(reply)?
-    {
-        //There should always be a context error if no reply error
-        assertConnection();
-        VStringBuffer msg("Redis Plugin: ERROR - %s on database %" I64F "u failed with neither 'reply' nor connection error available", cmd, database);
-        rtlFail(0, msg.str());
-    }
-    else if (reply->type == REDIS_REPLY_ERROR)
-    {
-        assertAuthorization(reply);
-        VStringBuffer msg("Redis Plugin: ERROR - %s on database %" I64F "u failed : %s", cmd, database, reply->str);
-        rtlFail(0, msg.str());
+        ::rtlFail(0, msg.str());
     }
 }
-void Connection::assertOnCommandError(const redisReply * reply, const char * cmd)
+void Connection::assertOnErrorWithCmdMsg(const redisReply * reply, const char * cmd, const char * key)
 {
-    if (!reply)//assertex(reply)?
-    {
-        //There should always be a context error if no reply error
-        assertConnection();
-        VStringBuffer msg("Redis Plugin: ERROR - %s failed with neither 'reply' nor connection error available", cmd);
-        rtlFail(0, msg.str());
-    }
+    if (!reply)
+        assertConnectionWithCmdMsg(cmd, key);
     else if (reply->type == REDIS_REPLY_ERROR)
     {
         assertAuthorization(reply);
-        VStringBuffer msg("Redis Plugin: ERROR - %s failed : %s", cmd, reply->str);
-        rtlFail(0, msg.str());
+        fail(cmd, reply->str, key);
     }
 }
 void Connection::assertAuthorization(const redisReply * reply)
 {
     if (reply && reply->str && ( strncmp(reply->str, "NOAUTH", 6) == 0 || strncmp(reply->str, "ERR operation not permitted", 27) == 0 ))
     {
-        VStringBuffer msg("Redis Plugin: server authentication failed - %s", reply->str);
-        rtlFail(0, msg.str());
+        VStringBuffer msg("Redis Plugin: ERROR - authentication for %s:%d failed : %s", ip.str(), port, reply->str);
+        ::rtlFail(0, msg.str());
     }
 }
 void Connection::assertKey(const redisReply * reply, const char * key)
 {
     if (reply && reply->type == REDIS_REPLY_NIL)
     {
-        VStringBuffer msg("Redis Plugin: ERROR - the requested key '%s' does not exist on database %" I64F "u", key, database);
-        rtlFail(0, msg.str());
+        VStringBuffer msg("Redis Plugin: ERROR - the requested key '%s' does not exist on database %d on %s:%d", key, database, ip.str(), port);
+        ::rtlFail(0, msg.str());
     }
 }
-void Connection::assertConnection()
+void Connection::assertConnectionWithCmdMsg(const char * cmd, const char * key)
+{
+    if (!context)
+        fail(cmd, "neither 'reply' nor connection error available", key);
+    else if (context->err)
+        fail(cmd, context->errstr, key);
+}
+void Connection::assertConnection(const char * _msg)
 {
     if (!context)
-        rtlFail(0, "Redis Plugin: 'redisConnect' failed - no error available.");
+    {
+        VStringBuffer msg("Redis Plugin: ERROR - %s for %s:%d failed : neither 'reply' nor connection error available", _msg, ip.str(), port);
+        ::rtlFail(0, msg.str());
+    }
     else if (context->err)
     {
-        VStringBuffer msg("Redis Plugin: Connection failed - %s for %s:%u", context->errstr, ip.str(),  port);
-        rtlFail(0, msg.str());
+        VStringBuffer msg("Redis Plugin: ERROR - %s for %s:%d failed : %s", _msg, ip.str(), port, context->errstr);
+        ::rtlFail(0, msg.str());
     }
 }
 void Connection::clear(ICodeContext * ctx)
@@ -441,44 +460,44 @@ void Connection::clear(ICodeContext * ctx)
     //NOTE: flush is the actual cache flush/clear/delete and not an io buffer flush.
     OwnedReply reply = Reply::createReply(redisCommand(context, "FLUSHDB"));//NOTE: FLUSHDB deletes current database where as FLUSHALL deletes all dbs.
     //NOTE: documented as never failing, but in case
-    assertOnCommandErrorWithDatabase(reply->query(), "FlushDB");
+    assertOnErrorWithCmdMsg(reply->query(), "FlushDB");
 }
 void Connection::del(ICodeContext * ctx, const char * key)
 {
     OwnedReply reply = Reply::createReply(redisCommand(context, "DEL %b", key, strlen(key)));
-    assertOnCommandErrorWithKey(reply->query(), "Del", key);
+    assertOnErrorWithCmdMsg(reply->query(), "Del", key);
 }
 void Connection::persist(ICodeContext * ctx, const char * key)
 {
     OwnedReply reply = Reply::createReply(redisCommand(context, "PERSIST %b", key, strlen(key)));
-    assertOnCommandErrorWithKey(reply->query(), "Persist", key);
+    assertOnErrorWithCmdMsg(reply->query(), "Persist", key);
 }
 void Connection::expire(ICodeContext * ctx, const char * key, unsigned _expire)
 {
     OwnedReply reply = Reply::createReply(redisCommand(context, "PEXPIRE %b %u", key, strlen(key), _expire));
-    assertOnCommandErrorWithKey(reply->query(), "Expire", key);
+    assertOnErrorWithCmdMsg(reply->query(), "Expire", key);
 }
 bool Connection::exists(ICodeContext * ctx, const char * key)
 {
     OwnedReply reply = Reply::createReply(redisCommand(context, "EXISTS %b", key, strlen(key)));
-    assertOnCommandErrorWithKey(reply->query(), "Exists", key);
+    assertOnErrorWithCmdMsg(reply->query(), "Exists", key);
     return (reply->query()->integer != 0);
 }
 unsigned __int64 Connection::dbSize(ICodeContext * ctx)
 {
     OwnedReply reply = Reply::createReply(redisCommand(context, "DBSIZE"));
-    assertOnCommandErrorWithDatabase(reply->query(), "DBSIZE");
+    assertOnErrorWithCmdMsg(reply->query(), "DBSIZE");
     return reply->query()->integer;
 }
 //-------------------------------------------SET-----------------------------------------
 //--OUTER--
-template<class type> void SyncRSet(ICodeContext * ctx, const char * _options, const char * key, type value, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 _timeout)
+template<class type> void SyncRSet(ICodeContext * ctx, const char * _options, const char * key, type value, int database, unsigned expire, const char * password, unsigned _timeout)
 {
     Owned<Connection> master = Connection::createConnection(ctx, _options, database, password, _timeout);
     master->set(ctx, key, value, expire);
 }
 //Set pointer types
-template<class type> void SyncRSet(ICodeContext * ctx, const char * _options, const char * key, size32_t valueSize, const type * value, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 _timeout)
+template<class type> void SyncRSet(ICodeContext * ctx, const char * _options, const char * key, size32_t valueSize, const type * value, int database, unsigned expire, const char * password, unsigned _timeout)
 {
     Owned<Connection> master = Connection::createConnection(ctx, _options, database, password, _timeout);
     master->set(ctx, key, valueSize, value, expire);
@@ -492,7 +511,7 @@ template<class type> void Connection::set(ICodeContext * ctx, const char * key,
     appendExpire(cmd, expire);
 
     OwnedReply reply = Reply::createReply(redisCommand(context, cmd.str(), key, strlen(key), _value, sizeof(type)));
-    assertOnCommandErrorWithKey(reply->query(), "SET", key);
+    assertOnErrorWithCmdMsg(reply->query(), "SET", key);
 }
 template<class type> void Connection::set(ICodeContext * ctx, const char * key, size32_t valueSize, const type * value, unsigned expire)
 {
@@ -501,16 +520,16 @@ template<class type> void Connection::set(ICodeContext * ctx, const char * key,
     StringBuffer cmd("SET %b %b");
     appendExpire(cmd, expire);
     OwnedReply reply = Reply::createReply(redisCommand(context, cmd.str(), key, strlen(key), _value, (size_t)valueSize));
-    assertOnCommandErrorWithKey(reply->query(), "SET", key);
+    assertOnErrorWithCmdMsg(reply->query(), "SET", key);
 }
 //-------------------------------------------GET-----------------------------------------
 //--OUTER--
-template<class type> void SyncRGet(ICodeContext * ctx, const char * options, const char * key, type & returnValue, unsigned __int64 database, const char * password, unsigned __int64 _timeout)
+template<class type> void SyncRGet(ICodeContext * ctx, const char * options, const char * key, type & returnValue, int database, const char * password, unsigned _timeout)
 {
     Owned<Connection> master = Connection::createConnection(ctx, options, database, password, _timeout);
     master->get(ctx, key, returnValue);
 }
-template<class type> void SyncRGet(ICodeContext * ctx, const char * options, const char * key, size_t & returnSize, type * & returnValue, unsigned __int64 database, const char * password, unsigned __int64 _timeout)
+template<class type> void SyncRGet(ICodeContext * ctx, const char * options, const char * key, size_t & returnSize, type * & returnValue, int database, const char * password, unsigned _timeout)
 {
     Owned<Connection> master = Connection::createConnection(ctx, options, database, password, _timeout);
     master->get(ctx, key, returnSize, returnValue);
@@ -520,14 +539,14 @@ template<class type> void Connection::get(ICodeContext * ctx, const char * key,
 {
     OwnedReply reply = Reply::createReply(redisCommand(context, "GET %b", key, strlen(key)));
 
-    assertOnError(reply->query(), "GET");
+    assertOnErrorWithCmdMsg(reply->query(), "GET", key);
     assertKey(reply->query(), key);
 
     size_t returnSize = reply->query()->len;
     if (sizeof(type)!=returnSize)
     {
-        VStringBuffer msg("Redis Plugin: ERROR - Requested type of different size (%uB) from that stored (%uB).", (unsigned)sizeof(type), (unsigned)returnSize);
-        rtlFail(0, msg.str());
+        VStringBuffer msg("requested type of different size (%uB) from that stored (%uB)", (unsigned)sizeof(type), (unsigned)returnSize);
+        fail("GET", msg.str(), key);
     }
     memcpy(&returnValue, reply->query()->str, returnSize);
 }
@@ -535,7 +554,7 @@ template<class type> void Connection::get(ICodeContext * ctx, const char * key,
 {
     OwnedReply reply = Reply::createReply(redisCommand(context, "GET %b", key, strlen(key)));
 
-    assertOnError(reply->query(), "GET");
+    assertOnErrorWithCmdMsg(reply->query(), "GET", key);
     assertKey(reply->query(), key);
 
     returnSize = reply->query()->len;
@@ -544,113 +563,113 @@ template<class type> void Connection::get(ICodeContext * ctx, const char * key,
 //--------------------------------------------------------------------------------
 //                           ECL SERVICE ENTRYPOINTS
 //--------------------------------------------------------------------------------
-ECL_REDIS_API void ECL_REDIS_CALL RClear(ICodeContext * ctx, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
+ECL_REDIS_API void ECL_REDIS_CALL RClear(ICodeContext * ctx, const char * options, int database, const char * password, unsigned timeout)
 {
     Owned<Connection> master = Connection::createConnection(ctx, options, database, password, timeout);
     master->clear(ctx);
 }
-ECL_REDIS_API bool ECL_REDIS_CALL RExist(ICodeContext * ctx, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
+ECL_REDIS_API bool ECL_REDIS_CALL RExist(ICodeContext * ctx, const char * key, const char * options, int database, const char * password, unsigned timeout)
 {
     Owned<Connection> master = Connection::createConnection(ctx, options, database, password, timeout);
     return master->exists(ctx, key);
 }
-ECL_REDIS_API void ECL_REDIS_CALL RDel(ICodeContext * ctx, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
+ECL_REDIS_API void ECL_REDIS_CALL RDel(ICodeContext * ctx, const char * key, const char * options, int database, const char * password, unsigned timeout)
 {
     Owned<Connection> master = Connection::createConnection(ctx, options, database, password, timeout);
     master->del(ctx, key);
 }
-ECL_REDIS_API void ECL_REDIS_CALL RPersist(ICodeContext * ctx, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
+ECL_REDIS_API void ECL_REDIS_CALL RPersist(ICodeContext * ctx, const char * key, const char * options, int database, const char * password, unsigned timeout)
 {
     Owned<Connection> master = Connection::createConnection(ctx, options, database, password, timeout);
     master->persist(ctx, key);
 }
-ECL_REDIS_API void ECL_REDIS_CALL RExpire(ICodeContext * ctx, const char * key, const char * options, unsigned __int64 database, unsigned _expire, const char * password, unsigned __int64 timeout)
+ECL_REDIS_API void ECL_REDIS_CALL RExpire(ICodeContext * ctx, const char * key, const char * options, int database, unsigned _expire, const char * password, unsigned timeout)
 {
     Owned<Connection> master = Connection::createConnection(ctx, options, database, password, timeout);
     master->expire(ctx, key, _expire);
 }
-ECL_REDIS_API unsigned __int64 ECL_REDIS_CALL RDBSize(ICodeContext * ctx, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
+ECL_REDIS_API unsigned __int64 ECL_REDIS_CALL RDBSize(ICodeContext * ctx, const char * options, int database, const char * password, unsigned timeout)
 {
     Owned<Connection> master = Connection::createConnection(ctx, options, database, password, timeout);
     return master->dbSize(ctx);
 }
 //-----------------------------------SET------------------------------------------
-ECL_REDIS_API void ECL_REDIS_CALL SyncRSetStr(ICodeContext * ctx, const char * key, size32_t valueSize, const char * value, const char * options, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 timeout)
+ECL_REDIS_API void ECL_REDIS_CALL SyncRSetStr(ICodeContext * ctx, const char * key, size32_t valueSize, const char * value, const char * options, int database, unsigned expire, const char * password, unsigned timeout)
 {
     SyncRSet(ctx, options, key, valueSize, value, database, expire, password, timeout);
 }
-ECL_REDIS_API void ECL_REDIS_CALL SyncRSetUChar(ICodeContext * ctx, const char * key, size32_t valueLength, const UChar * value, const char * options, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 timeout)
+ECL_REDIS_API void ECL_REDIS_CALL SyncRSetUChar(ICodeContext * ctx, const char * key, size32_t valueLength, const UChar * value, const char * options, int database, unsigned expire, const char * password, unsigned timeout)
 {
     SyncRSet(ctx, options, key, (valueLength)*sizeof(UChar), value, database, expire, password, timeout);
 }
-ECL_REDIS_API void ECL_REDIS_CALL SyncRSetInt(ICodeContext * ctx, const char * key, signed __int64 value, const char * options, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 timeout)
+ECL_REDIS_API void ECL_REDIS_CALL SyncRSetInt(ICodeContext * ctx, const char * key, signed __int64 value, const char * options, int database, unsigned expire, const char * password, unsigned timeout)
 {
     SyncRSet(ctx, options, key, value, database, expire, password, timeout);
 }
-ECL_REDIS_API void ECL_REDIS_CALL SyncRSetUInt(ICodeContext * ctx, const char * key, unsigned __int64 value, const char * options, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 timeout)
+ECL_REDIS_API void ECL_REDIS_CALL SyncRSetUInt(ICodeContext * ctx, const char * key, unsigned __int64 value, const char * options, int database, unsigned expire, const char * password, unsigned timeout)
 {
     SyncRSet(ctx, options, key, value, database, expire, password, timeout);
 }
-ECL_REDIS_API void ECL_REDIS_CALL SyncRSetReal(ICodeContext * ctx, const char * key, double value, const char * options, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 timeout)
+ECL_REDIS_API void ECL_REDIS_CALL SyncRSetReal(ICodeContext * ctx, const char * key, double value, const char * options, int database, unsigned expire, const char * password, unsigned timeout)
 {
     SyncRSet(ctx, options, key, value, database, expire, password, timeout);
 }
-ECL_REDIS_API void ECL_REDIS_CALL SyncRSetBool(ICodeContext * ctx, const char * key, bool value, const char * options, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 timeout)
+ECL_REDIS_API void ECL_REDIS_CALL SyncRSetBool(ICodeContext * ctx, const char * key, bool value, const char * options, int database, unsigned expire, const char * password, unsigned timeout)
 {
     SyncRSet(ctx, options, key, value, database, expire, password, timeout);
 }
-ECL_REDIS_API void ECL_REDIS_CALL SyncRSetData(ICodeContext * ctx, const char * key, size32_t valueSize, const void * value, const char * options, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 timeout)
+ECL_REDIS_API void ECL_REDIS_CALL SyncRSetData(ICodeContext * ctx, const char * key, size32_t valueSize, const void * value, const char * options, int database, unsigned expire, const char * password, unsigned timeout)
 {
     SyncRSet(ctx, options, key, valueSize, value, database, expire, password, timeout);
 }
-ECL_REDIS_API void ECL_REDIS_CALL SyncRSetUtf8(ICodeContext * ctx, const char * key, size32_t valueLength, const char * value, const char * options, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 timeout)
+ECL_REDIS_API void ECL_REDIS_CALL SyncRSetUtf8(ICodeContext * ctx, const char * key, size32_t valueLength, const char * value, const char * options, int database, unsigned expire, const char * password, unsigned timeout)
 {
     SyncRSet(ctx, options, key, rtlUtf8Size(valueLength, value), value, database, expire, password, timeout);
 }
 //-------------------------------------GET----------------------------------------
-ECL_REDIS_API bool ECL_REDIS_CALL SyncRGetBool(ICodeContext * ctx, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
+ECL_REDIS_API bool ECL_REDIS_CALL SyncRGetBool(ICodeContext * ctx, const char * key, const char * options, int database, const char * password, unsigned timeout)
 {
     bool value;
     SyncRGet(ctx, options, key, value, database, password, timeout);
     return value;
 }
-ECL_REDIS_API double ECL_REDIS_CALL SyncRGetDouble(ICodeContext * ctx, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
+ECL_REDIS_API double ECL_REDIS_CALL SyncRGetDouble(ICodeContext * ctx, const char * key, const char * options, int database, const char * password, unsigned timeout)
 {
     double value;
     SyncRGet(ctx, options, key, value, database, password, timeout);
     return value;
 }
-ECL_REDIS_API signed __int64 ECL_REDIS_CALL SyncRGetInt8(ICodeContext * ctx, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
+ECL_REDIS_API signed __int64 ECL_REDIS_CALL SyncRGetInt8(ICodeContext * ctx, const char * key, const char * options, int database, const char * password, unsigned timeout)
 {
     signed __int64 value;
     SyncRGet(ctx, options, key, value, database, password, timeout);
     return value;
 }
-ECL_REDIS_API unsigned __int64 ECL_REDIS_CALL SyncRGetUint8(ICodeContext * ctx, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
+ECL_REDIS_API unsigned __int64 ECL_REDIS_CALL SyncRGetUint8(ICodeContext * ctx, const char * key, const char * options, int database, const char * password, unsigned timeout)
 {
     unsigned __int64 value;
     SyncRGet(ctx, options, key, value, database, password, timeout);
     return value;
 }
-ECL_REDIS_API void ECL_REDIS_CALL SyncRGetStr(ICodeContext * ctx, size32_t & returnSize, char * & returnValue, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
+ECL_REDIS_API void ECL_REDIS_CALL SyncRGetStr(ICodeContext * ctx, size32_t & returnSize, char * & returnValue, const char * key, const char * options, int database, const char * password, unsigned timeout)
 {
     size_t _returnSize;
     SyncRGet(ctx, options, key, _returnSize, returnValue, database, password, timeout);
     returnSize = static_cast<size32_t>(_returnSize);
 }
-ECL_REDIS_API void ECL_REDIS_CALL SyncRGetUChar(ICodeContext * ctx, size32_t & returnLength, UChar * & returnValue,  const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
+ECL_REDIS_API void ECL_REDIS_CALL SyncRGetUChar(ICodeContext * ctx, size32_t & returnLength, UChar * & returnValue,  const char * key, const char * options, int database, const char * password, unsigned timeout)
 {
     size_t returnSize;
     SyncRGet(ctx, options, key, returnSize, returnValue, database, password, timeout);
     returnLength = static_cast<size32_t>(returnSize/sizeof(UChar));
 }
-ECL_REDIS_API void ECL_REDIS_CALL SyncRGetUtf8(ICodeContext * ctx, size32_t & returnLength, char * & returnValue, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
+ECL_REDIS_API void ECL_REDIS_CALL SyncRGetUtf8(ICodeContext * ctx, size32_t & returnLength, char * & returnValue, const char * key, const char * options, int database, const char * password, unsigned timeout)
 {
     size_t returnSize;
     SyncRGet(ctx, options, key, returnSize, returnValue, database, password, timeout);
     returnLength = static_cast<size32_t>(rtlUtf8Length(returnSize, returnValue));
 }
-ECL_REDIS_API void ECL_REDIS_CALL SyncRGetData(ICodeContext * ctx, size32_t & returnSize, void * & returnValue, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
+ECL_REDIS_API void ECL_REDIS_CALL SyncRGetData(ICodeContext * ctx, size32_t & returnSize, void * & returnValue, const char * key, const char * options, int database, const char * password, unsigned timeout)
 {
     size_t _returnSize;
     SyncRGet(ctx, options, key, _returnSize, returnValue, database, password, timeout);
@@ -659,7 +678,7 @@ ECL_REDIS_API void ECL_REDIS_CALL SyncRGetData(ICodeContext * ctx, size32_t & re
 //----------------------------------LOCK------------------------------------------
 //-----------------------------------SET-----------------------------------------
 //Set pointer types
-void SyncLockRSet(ICodeContext * ctx, const char * _options, const char * key, size32_t valueSize, const char * value, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 _timeout)
+void SyncLockRSet(ICodeContext * ctx, const char * _options, const char * key, size32_t valueSize, const char * value, int database, unsigned expire, const char * password, unsigned _timeout)
 {
     Owned<Connection> master = Connection::createConnection(ctx, _options, database, password, _timeout);
     master->lockSet(ctx, key, valueSize, value, expire);
@@ -672,7 +691,7 @@ void Connection::lockSet(ICodeContext * ctx, const char * key, size32_t valueSiz
 }
 //-------------------------------------------GET-----------------------------------------
 //--OUTER--
-void SyncLockRGet(ICodeContext * ctx, const char * options, const char * key, size_t & returnSize, char * & returnValue, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 _timeout)
+void SyncLockRGet(ICodeContext * ctx, const char * options, const char * key, size_t & returnSize, char * & returnValue, int database, unsigned expire, const char * password, unsigned _timeout)
 {
     Owned<Connection> master = Connection::createConnection(ctx, options, database, password, _timeout);
     master->lockGet(ctx, key, returnSize, returnValue, password, expire);
@@ -693,12 +712,12 @@ void Connection::encodeChannel(StringBuffer & channel, const char * key) const
 bool Connection::lock(ICodeContext * ctx, const char * key, const char * channel, unsigned expire)
 {
     if (expire == 0)
-        rtlFail(0, "Redis Plugin: ERROR - invalid value for 'expire', persistent locks not allowed.");
+        fail("GetOrLock<type>", "invalid value for 'expire', persistent locks not allowed.", key);
     StringBuffer cmd("SET %b %b NX PX ");
     cmd.append(expire);
 
     OwnedReply reply = Reply::createReply(redisCommand(context, cmd.str(), key, strlen(key), channel, strlen(channel)));
-    assertOnError(reply->query(), cmd.append(" of the key '").append(key).append("' failed"));
+    assertOnErrorWithCmdMsg(reply->query(), cmd.str(), key);
 
     return (reply->query()->type == REDIS_REPLY_STATUS && strcmp(reply->query()->str, "OK") == 0);
 }
@@ -710,8 +729,8 @@ void Connection::unlock(ICodeContext * ctx, const char * key)
 
     //Read replies
     OwnedReply reply = new Reply();
-    readReplyAndAssertWithKey(reply.get(), "manual unlock", key);//WATCH reply
-    readReplyAndAssertWithKey(reply.get(), "manual unlock", key);//GET reply
+    readReplyAndAssertWithCmdMsg(reply.get(), "manual unlock", key);//WATCH reply
+    readReplyAndAssertWithCmdMsg(reply.get(), "manual unlock", key);//GET reply
 
     //check if locked
     if (strncmp(reply->query()->str, REDIS_LOCK_PREFIX, strlen(REDIS_LOCK_PREFIX)) == 0)
@@ -723,9 +742,9 @@ void Connection::unlock(ICodeContext * ctx, const char * key)
 #if(0)//Quick draw! You have 10s to manually send (via redis-cli) "set testlock foobar". The second myRedis.Exists('testlock') in redislockingtest.ecl should now return TRUE.
         sleep(10);
 #endif
-        readReplyAndAssertWithKey(reply.get(), "manual unlock", key);//MULTI reply
-        readReplyAndAssertWithKey(reply.get(), "manual unlock", key);//DEL reply
-        readReplyAndAssertWithKey(reply.get(), "manual unlock", key);//EXEC reply
+        readReplyAndAssertWithCmdMsg(reply.get(), "manual unlock", key);//MULTI reply
+        readReplyAndAssertWithCmdMsg(reply.get(), "manual unlock", key);//DEL reply
+        readReplyAndAssertWithCmdMsg(reply.get(), "manual unlock", key);//EXEC reply
     }
     //If the above is aborted, let the lock expire.
 }
@@ -749,14 +768,11 @@ void Connection::handleLockOnGet(ICodeContext * ctx, const char * key, MemoryAtt
 
     //SUB before GET
     //Requires separate connection from GET so that the replies are not mangled. This could be averted
-    Owned<Connection> subConnection = new Connection(ctx, options.str(), ip.str(), port, serverIpPortPasswordHash, database, password, timeout);
+    Owned<Connection> subConnection = new Connection(ctx, options.str(), ip.str(), port, serverIpPortPasswordHash, database, password, timeLeft());
     OwnedReply reply = Reply::createReply(redisCommand(subConnection->context, "SUBSCRIBE %b", channel.str(), (size_t)channel.length()));
-    assertOnCommandErrorWithKey(reply->query(), "GET", key);
+    assertOnErrorWithCmdMsg(reply->query(), "GetOrLock<type>", key);
     if (reply->query()->type == REDIS_REPLY_ARRAY && strcmp("subscribe", reply->query()->element[0]->str) != 0 )
-    {
-        VStringBuffer msg("Redis Plugin: ERROR - GET '%s' on database %" I64F "u failed : failed to register SUB", key, database);
-        rtlFail(0, msg.str());
-    }
+        fail("GetOrLock<type>", "failed to register SUB", key);
 
 #if(0)//Test publish before GET.
     {
@@ -767,7 +783,7 @@ void Connection::handleLockOnGet(ICodeContext * ctx, const char * key, MemoryAtt
 
     //Now GET
     reply->setClear((redisReply*)redisCommand(context, "GET %b", key, strlen(key)));
-    assertOnCommandErrorWithKey(reply->query(), "GET", key);
+    assertOnErrorWithCmdMsg(reply->query(), "GetOrLock<type>", key);
 
 #if(0)//Test publish after GET.
     {
@@ -788,8 +804,8 @@ void Connection::handleLockOnGet(ICodeContext * ctx, const char * key, MemoryAtt
         //Check that the lock was set by this plugin and thus that we subscribed to the expected channel.
         if (reply->query()->str && strcmp(reply->query()->str, channel.str()) !=0 )
         {
-            VStringBuffer msg("Redis Plugin: ERROR - the key '%s', on database %" I64F "u, is locked with a channel ('%s') different to that subscribed to (%s).", key, database, reply->query()->str, channel.str());
-            rtlFail(0, msg.str());
+            VStringBuffer msg("key locked with a channel ('%s') different to that subscribed to (%s).", reply->query()->str, channel.str());
+            fail("GetOrLock<type>", msg.str(), key);
             //MORE: In theory, it is possible to recover at this stage by subscribing to the channel that the key was actually locked with.
             //However, we may have missed the massage publication already or by then, but could SUB again in case we haven't.
             //More importantly and furthermore, the publication (in SetAndPublish<type>) will only publish to the channel encoded by
@@ -797,16 +813,13 @@ void Connection::handleLockOnGet(ICodeContext * ctx, const char * key, MemoryAtt
         }
 #if(0)//Added to allow for manual pub testing via redis-cli
         struct timeval to = { 10, 0 };//10secs
-        redisSetTimeout(subConnection->context, to);
+        ::redisSetTimeout(subConnection->context, to);
 #endif
         //Locked so SUBSCRIBE
-        redisReply * nakedReply = NULL;
-        bool err = redisGetReply(subConnection->context, (void**)&nakedReply);
-        reply->setClear(nakedReply);
-        if (err != REDIS_OK)
-            rtlFail(0, "Redis Plugin: ERROR - GET timed out.");
-        assertOnCommandErrorWithKey(nakedReply, "GET", key);
-        if (nakedReply->type == REDIS_REPLY_ARRAY && strcmp("message", nakedReply->element[0]->str) == 0)
+        subConnection->readReply(reply);
+        subConnection->assertOnErrorWithCmdMsg(reply->query(), "GetOrLock<type>", key);
+
+        if (reply->query()->type == REDIS_REPLY_ARRAY && strcmp("message", reply->query()->element[0]->str) == 0)
         {
             //We are about to return a value, to conform with other Get<type> functions, fail if the key did not exist.
             //Since the value is sent via a published message, there is no direct reply struct so assume that an empty
@@ -814,13 +827,15 @@ void Connection::handleLockOnGet(ICodeContext * ctx, const char * key, MemoryAtt
             //More importantly, it is paramount that this routine only return an empty string under one condition, that
             //which indicates to the caller that the key was successfully locked.
             //NOTE: it is possible for an empty message to have been PUBLISHed.
-            if (nakedReply->element[2]->len > 0)
+            if (reply->query()->element[2]->len > 0)
             {
-                retVal->set(nakedReply->element[2]->len, nakedReply->element[2]->str);//return the published value rather than another (WATCHed) GET.
+                retVal->set(reply->query()->element[2]->len, reply->query()->element[2]->str);//return the published value rather than another (WATCHed) GET.
                 return;
             }
-            VStringBuffer msg("Redis Plugin: ERROR - the requested key '%s' does not exist on database %" I64F "u", key, database);
-            rtlFail(0, msg.str());
+            //fail that key does not exist
+            redisReply fakeReply;
+            fakeReply.type = REDIS_REPLY_NIL;
+            assertKey(&fakeReply, key);
         }
     }
     throwUnexpected();
@@ -854,7 +869,7 @@ void Connection::handleLockOnSet(ICodeContext * ctx, const char * key, const cha
                 replyContainer->setClear((redisReply*)redisCommand(context, "EVAL %b %d %b %b %b %d", luaScriptWithExpire, strlen(luaScriptWithExpire), 1, key, strlen(key), channel.str(), (size_t)channel.length(), value, size, expire));
             }
         }
-        assertOnCommandErrorWithKey(replyContainer->query(), "SET", key);
+        assertOnErrorWithCmdMsg(replyContainer->query(), "SET", key);
     }
     else
     {
@@ -867,10 +882,10 @@ void Connection::handleLockOnSet(ICodeContext * ctx, const char * key, const cha
 
         //Now read and assert replies
         OwnedReply reply = new Reply();
-        readReplyAndAssertWithKey(reply, "SET", key);//MULTI reply
-        readReplyAndAssertWithKey(reply, "SET", key);//SET reply
-        readReplyAndAssertWithKey(reply, "PUB for the key", key);//PUB reply
-        readReplyAndAssertWithKey(reply, "SET", key);//EXEC reply
+        readReplyAndAssertWithCmdMsg(reply, "SET", key);//MULTI reply
+        readReplyAndAssertWithCmdMsg(reply, "SET", key);//SET reply
+        readReplyAndAssertWithCmdMsg(reply, "PUB for the key", key);//PUB reply
+        readReplyAndAssertWithCmdMsg(reply, "SET", key);//EXEC reply
     }
 
     //NOTE: When setting and publishing the data with a pipelined MULTI-SET-PUB-EXEC, the data is sent twice, once with the SET and again with the PUBLISH.
@@ -897,20 +912,20 @@ bool Connection::noScript(const redisReply * reply) const
 //                           ECL SERVICE ENTRYPOINTS
 //--------------------------------------------------------------------------------
 //-----------------------------------SET------------------------------------------
-ECL_REDIS_API void ECL_REDIS_CALL SyncLockRSetStr(ICodeContext * ctx, size32_t & returnLength, char * & returnValue, const char * key, size32_t valueLength, const char * value, const char * options, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 timeout)
+ECL_REDIS_API void ECL_REDIS_CALL SyncLockRSetStr(ICodeContext * ctx, size32_t & returnLength, char * & returnValue, const char * key, size32_t valueLength, const char * value, const char * options, int database, unsigned expire, const char * password, unsigned timeout)
 {
     SyncLockRSet(ctx, options, key, valueLength, value, database, expire, password, timeout);
     returnLength = valueLength;
     returnValue = (char*)allocateAndCopy(value, valueLength);
 }
-ECL_REDIS_API void ECL_REDIS_CALL SyncLockRSetUChar(ICodeContext * ctx, size32_t & returnLength, UChar * & returnValue, const char * key, size32_t valueLength, const UChar * value, const char * options, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 timeout)
+ECL_REDIS_API void ECL_REDIS_CALL SyncLockRSetUChar(ICodeContext * ctx, size32_t & returnLength, UChar * & returnValue, const char * key, size32_t valueLength, const UChar * value, const char * options, int database, unsigned expire, const char * password, unsigned timeout)
 {
     unsigned valueSize = (valueLength)*sizeof(UChar);
     SyncLockRSet(ctx, options, key, valueSize, (char*)value, database, expire, password, timeout);
     returnLength= valueLength;
     returnValue = (UChar*)allocateAndCopy(value, valueSize);
 }
-ECL_REDIS_API void ECL_REDIS_CALL SyncLockRSetUtf8(ICodeContext * ctx, size32_t & returnLength, char * & returnValue, const char * key, size32_t valueLength, const char * value, const char * options, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 timeout)
+ECL_REDIS_API void ECL_REDIS_CALL SyncLockRSetUtf8(ICodeContext * ctx, size32_t & returnLength, char * & returnValue, const char * key, size32_t valueLength, const char * value, const char * options, int database, unsigned expire, const char * password, unsigned timeout)
 {
     unsigned valueSize = rtlUtf8Size(valueLength, value);
     SyncLockRSet(ctx, options, key, valueSize, value, database, expire, password, timeout);
@@ -918,13 +933,13 @@ ECL_REDIS_API void ECL_REDIS_CALL SyncLockRSetUtf8(ICodeContext * ctx, size32_t
     returnValue = (char*)allocateAndCopy(value, valueSize);
 }
 //-------------------------------------GET----------------------------------------
-ECL_REDIS_API void ECL_REDIS_CALL SyncLockRGetStr(ICodeContext * ctx, size32_t & returnSize, char * & returnValue, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout, unsigned expire)
+ECL_REDIS_API void ECL_REDIS_CALL SyncLockRGetStr(ICodeContext * ctx, size32_t & returnSize, char * & returnValue, const char * key, const char * options, int database, const char * password, unsigned timeout, unsigned expire)
 {
     size_t _returnSize;
     SyncLockRGet(ctx, options, key, _returnSize, returnValue, database, expire, password, timeout);
     returnSize = static_cast<size32_t>(_returnSize);
 }
-ECL_REDIS_API void ECL_REDIS_CALL SyncLockRGetUChar(ICodeContext * ctx, size32_t & returnLength, UChar * & returnValue,  const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout, unsigned expire)
+ECL_REDIS_API void ECL_REDIS_CALL SyncLockRGetUChar(ICodeContext * ctx, size32_t & returnLength, UChar * & returnValue,  const char * key, const char * options, int database, const char * password, unsigned timeout, unsigned expire)
 {
     size_t returnSize;
     char  * _returnValue;
@@ -932,13 +947,13 @@ ECL_REDIS_API void ECL_REDIS_CALL SyncLockRGetUChar(ICodeContext * ctx, size32_t
     returnValue = (UChar*)_returnValue;
     returnLength = static_cast<size32_t>(returnSize/sizeof(UChar));
 }
-ECL_REDIS_API void ECL_REDIS_CALL SyncLockRGetUtf8(ICodeContext * ctx, size32_t & returnLength, char * & returnValue, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout, unsigned expire)
+ECL_REDIS_API void ECL_REDIS_CALL SyncLockRGetUtf8(ICodeContext * ctx, size32_t & returnLength, char * & returnValue, const char * key, const char * options, int database, const char * password, unsigned timeout, unsigned expire)
 {
     size_t returnSize;
     SyncLockRGet(ctx, options, key, returnSize, returnValue, database, expire, password, timeout);
     returnLength = static_cast<size32_t>(rtlUtf8Length(returnSize, returnValue));
 }
-ECL_REDIS_API void ECL_REDIS_CALL SyncLockRUnlock(ICodeContext * ctx, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)
+ECL_REDIS_API void ECL_REDIS_CALL SyncLockRUnlock(ICodeContext * ctx, const char * key, const char * options, int database, const char * password, unsigned timeout)
 {
     Owned<Connection> master = Connection::createConnection(ctx, options, database, password, timeout);
     master->unlock(ctx, key);

plugins/redis/redis.hpp

@@ -43,42 +43,42 @@ extern "C++"
 {
 namespace RedisPlugin {
     //--------------------------SET----------------------------------------
-    ECL_REDIS_API void ECL_REDIS_CALL SyncRSetBool (ICodeContext * _ctx, const char * key, bool value, const char * options, unsigned __int64 database, unsigned expire, const char * pswd, unsigned timeout);
-    ECL_REDIS_API void ECL_REDIS_CALL SyncRSetInt  (ICodeContext * _ctx, const char * key, signed __int64 value, const char * options, unsigned __int64 database, unsigned expire, const char * pswd, unsigned timeout);
-    ECL_REDIS_API void ECL_REDIS_CALL SyncRSetUInt (ICodeContext * _ctx, const char * key, unsigned __int64 value, const char * options, unsigned __int64 database, unsigned expire, const char * pswd, unsigned timeout);
-    ECL_REDIS_API void ECL_REDIS_CALL SyncRSetReal (ICodeContext * _ctx, const char * key, double value, const char * options, unsigned __int64 database, unsigned expire, const char * pswd, unsigned timeout);
-    ECL_REDIS_API void ECL_REDIS_CALL SyncRSetUtf8 (ICodeContext * _ctx, const char * key, size32_t valueLength, const char * value, const char * options, unsigned __int64 database, unsigned expire, const char * pswd, unsigned timeout);
-    ECL_REDIS_API void ECL_REDIS_CALL SyncRSetStr  (ICodeContext * _ctx, const char * key, size32_t valueLength, const char * value, const char * options, unsigned __int64 database, unsigned expire, const char * pswd, unsigned timeout);
-    ECL_REDIS_API void ECL_REDIS_CALL SyncRSetUChar(ICodeContext * _ctx, const char * key, size32_t valueLength, const UChar * value, const char * options, unsigned __int64 database, unsigned expire, const char * pswd, unsigned timeout);
-    ECL_REDIS_API void ECL_REDIS_CALL SyncRSetData (ICodeContext * _ctx, const char * key, size32_t valueLength, const void * value, const char * options, unsigned __int64 database, unsigned expire, const char * pswd, unsigned timeout);
+    ECL_REDIS_API void ECL_REDIS_CALL SyncRSetBool (ICodeContext * _ctx, const char * key, bool value, const char * options, int database, unsigned expire, const char * pswd, unsigned timeout);
+    ECL_REDIS_API void ECL_REDIS_CALL SyncRSetInt  (ICodeContext * _ctx, const char * key, signed __int64 value, const char * options, int database, unsigned expire, const char * pswd, unsigned timeout);
+    ECL_REDIS_API void ECL_REDIS_CALL SyncRSetUInt (ICodeContext * _ctx, const char * key, unsigned __int64 value, const char * options, int database, unsigned expire, const char * pswd, unsigned timeout);
+    ECL_REDIS_API void ECL_REDIS_CALL SyncRSetReal (ICodeContext * _ctx, const char * key, double value, const char * options, int database, unsigned expire, const char * pswd, unsigned timeout);
+    ECL_REDIS_API void ECL_REDIS_CALL SyncRSetUtf8 (ICodeContext * _ctx, const char * key, size32_t valueLength, const char * value, const char * options, int database, unsigned expire, const char * pswd, unsigned timeout);
+    ECL_REDIS_API void ECL_REDIS_CALL SyncRSetStr  (ICodeContext * _ctx, const char * key, size32_t valueLength, const char * value, const char * options, int database, unsigned expire, const char * pswd, unsigned timeout);
+    ECL_REDIS_API void ECL_REDIS_CALL SyncRSetUChar(ICodeContext * _ctx, const char * key, size32_t valueLength, const UChar * value, const char * options, int database, unsigned expire, const char * pswd, unsigned timeout);
+    ECL_REDIS_API void ECL_REDIS_CALL SyncRSetData (ICodeContext * _ctx, const char * key, size32_t valueLength, const void * value, const char * options, int database, unsigned expire, const char * pswd, unsigned timeout);
     //--------------------------GET----------------------------------------
-    ECL_REDIS_API bool             ECL_REDIS_CALL SyncRGetBool  (ICodeContext * _ctx, const char * key, const char * options, unsigned __int64 database, const char * pswd, unsigned timeout);
-    ECL_REDIS_API signed __int64   ECL_REDIS_CALL SyncRGetInt8  (ICodeContext * _ctx, const char * key, const char * options, unsigned __int64 database, const char * pswd, unsigned timeout);
-    ECL_REDIS_API unsigned __int64 ECL_REDIS_CALL SyncRGetUint8 (ICodeContext * _ctx, const char * key, const char * options, unsigned __int64 database, const char * pswd, unsigned timeout);
-    ECL_REDIS_API double           ECL_REDIS_CALL SyncRGetDouble(ICodeContext * _ctx, const char * key, const char * options, unsigned __int64 database, const char * pswd, unsigned timeout);
-    ECL_REDIS_API void             ECL_REDIS_CALL SyncRGetUtf8  (ICodeContext * _ctx, size32_t & returnLength, char * & returnValue, const char * key, const char * options, unsigned __int64 database, const char * pswd, unsigned timeout);
-    ECL_REDIS_API void             ECL_REDIS_CALL SyncRGetStr   (ICodeContext * _ctx, size32_t & returnLength, char * & returnValue, const char * key, const char * options, unsigned __int64 database, const char * pswd, unsigned timeout);
-    ECL_REDIS_API void             ECL_REDIS_CALL SyncRGetUChar (ICodeContext * _ctx, size32_t & returnLength, UChar * & returnValue, const char * key, const char * options, unsigned __int64 database, const char * pswd, unsigned timeout);
-    ECL_REDIS_API void             ECL_REDIS_CALL SyncRGetData  (ICodeContext * _ctx,size32_t & returnLength, void * & returnValue, const char * key, const char * options, unsigned __int64 database, const char * pswd, unsigned timeout);
+    ECL_REDIS_API bool             ECL_REDIS_CALL SyncRGetBool  (ICodeContext * _ctx, const char * key, const char * options, int database, const char * pswd, unsigned timeout);
+    ECL_REDIS_API signed __int64   ECL_REDIS_CALL SyncRGetInt8  (ICodeContext * _ctx, const char * key, const char * options, int database, const char * pswd, unsigned timeout);
+    ECL_REDIS_API unsigned __int64 ECL_REDIS_CALL SyncRGetUint8 (ICodeContext * _ctx, const char * key, const char * options, int database, const char * pswd, unsigned timeout);
+    ECL_REDIS_API double           ECL_REDIS_CALL SyncRGetDouble(ICodeContext * _ctx, const char * key, const char * options, int database, const char * pswd, unsigned timeout);
+    ECL_REDIS_API void             ECL_REDIS_CALL SyncRGetUtf8  (ICodeContext * _ctx, size32_t & returnLength, char * & returnValue, const char * key, const char * options, int database, const char * pswd, unsigned timeout);
+    ECL_REDIS_API void             ECL_REDIS_CALL SyncRGetStr   (ICodeContext * _ctx, size32_t & returnLength, char * & returnValue, const char * key, const char * options, int database, const char * pswd, unsigned timeout);
+    ECL_REDIS_API void             ECL_REDIS_CALL SyncRGetUChar (ICodeContext * _ctx, size32_t & returnLength, UChar * & returnValue, const char * key, const char * options, int database, const char * pswd, unsigned timeout);
+    ECL_REDIS_API void             ECL_REDIS_CALL SyncRGetData  (ICodeContext * _ctx,size32_t & returnLength, void * & returnValue, const char * key, const char * options, int database, const char * pswd, unsigned timeout);
 
     //--------------------------------AUXILLARIES---------------------------
-    ECL_REDIS_API bool             ECL_REDIS_CALL RExist  (ICodeContext * _ctx, const char * key, const char * options, unsigned __int64 database, const char * pswd, unsigned timeout);
-    ECL_REDIS_API void             ECL_REDIS_CALL RClear  (ICodeContext * _ctx, const char * options, unsigned __int64 database, const char * pswd, unsigned timeout);
-    ECL_REDIS_API void             ECL_REDIS_CALL RDel    (ICodeContext * _ctx, const char * key, const char * options, unsigned __int64 database, const char * pswd, unsigned timeout);
-    ECL_REDIS_API void             ECL_REDIS_CALL RPersist(ICodeContext * _ctx, const char * key, const char * options, unsigned __int64 database, const char * pswd, unsigned timeout);
-    ECL_REDIS_API void             ECL_REDIS_CALL RExpire (ICodeContext * _ctx, const char * key, const char * options, unsigned expire, unsigned __int64 database, const char * pswd, unsigned timeout);
-    ECL_REDIS_API unsigned __int64 ECL_REDIS_CALL RDBSize (ICodeContext * _ctx, const char * options, unsigned __int64 database, const char * pswd, unsigned timeout);
+    ECL_REDIS_API bool             ECL_REDIS_CALL RExist  (ICodeContext * _ctx, const char * key, const char * options, int database, const char * pswd, unsigned timeout);
+    ECL_REDIS_API void             ECL_REDIS_CALL RClear  (ICodeContext * _ctx, const char * options, int database, const char * pswd, unsigned timeout);
+    ECL_REDIS_API void             ECL_REDIS_CALL RDel    (ICodeContext * _ctx, const char * key, const char * options, int database, const char * pswd, unsigned timeout);
+    ECL_REDIS_API void             ECL_REDIS_CALL RPersist(ICodeContext * _ctx, const char * key, const char * options, int database, const char * pswd, unsigned timeout);
+    ECL_REDIS_API void             ECL_REDIS_CALL RExpire (ICodeContext * _ctx, const char * key, const char * options, unsigned expire, int database, const char * pswd, unsigned timeout);
+    ECL_REDIS_API unsigned __int64 ECL_REDIS_CALL RDBSize (ICodeContext * _ctx, const char * options, int database, const char * pswd, unsigned timeout);
 
     //--------------------------SET----------------------------------------
-    ECL_REDIS_API void             ECL_REDIS_CALL SyncLockRSetUtf8 (ICodeContext * _ctx, size32_t & returnLength, char * & returnValue, const char * key, size32_t valueLength, const char * value, const char * options, unsigned __int64 database, unsigned expire, const char * pswd, unsigned timeout);
-    ECL_REDIS_API void             ECL_REDIS_CALL SyncLockRSetStr  (ICodeContext * _ctx, size32_t & returnLength, char * & returnValue, const char * key, size32_t valueLength, const char * value, const char * options, unsigned __int64 database, unsigned expire, const char * pswd, unsigned timeout);
-    ECL_REDIS_API void             ECL_REDIS_CALL SyncLockRSetUChar(ICodeContext * _ctx, size32_t & returnLength, UChar * & returnValue, const char * key, size32_t valueLength, const UChar * value, const char * options, unsigned __int64 database, unsigned expire, const char * pswd, unsigned timeout);
+    ECL_REDIS_API void             ECL_REDIS_CALL SyncLockRSetUtf8 (ICodeContext * _ctx, size32_t & returnLength, char * & returnValue, const char * key, size32_t valueLength, const char * value, const char * options, int database, unsigned expire, const char * pswd, unsigned timeout);
+    ECL_REDIS_API void             ECL_REDIS_CALL SyncLockRSetStr  (ICodeContext * _ctx, size32_t & returnLength, char * & returnValue, const char * key, size32_t valueLength, const char * value, const char * options, int database, unsigned expire, const char * pswd, unsigned timeout);
+    ECL_REDIS_API void             ECL_REDIS_CALL SyncLockRSetUChar(ICodeContext * _ctx, size32_t & returnLength, UChar * & returnValue, const char * key, size32_t valueLength, const UChar * value, const char * options, int database, unsigned expire, const char * pswd, unsigned timeout);
     //--------------------------GET----------------------------------------
-    ECL_REDIS_API void             ECL_REDIS_CALL SyncLockRGetUtf8  (ICodeContext * _ctx, size32_t & returnLength, char * & returnValue, const char * key, const char * options, unsigned __int64 database, const char * pswd, unsigned timeout, unsigned expire);
-    ECL_REDIS_API void             ECL_REDIS_CALL SyncLockRGetStr   (ICodeContext * _ctx, size32_t & returnLength, char * & returnValue, const char * key, const char * options, unsigned __int64 database, const char * pswd, unsigned timeout, unsigned expire);
-    ECL_REDIS_API void             ECL_REDIS_CALL SyncLockRGetUChar (ICodeContext * _ctx, size32_t & returnLength, UChar * & returnValue, const char * key, const char * options, unsigned __int64 database, const char * pswd, unsigned timeout, unsigned expire);
+    ECL_REDIS_API void             ECL_REDIS_CALL SyncLockRGetUtf8  (ICodeContext * _ctx, size32_t & returnLength, char * & returnValue, const char * key, const char * options, int database, const char * pswd, unsigned timeout, unsigned expire);
+    ECL_REDIS_API void             ECL_REDIS_CALL SyncLockRGetStr   (ICodeContext * _ctx, size32_t & returnLength, char * & returnValue, const char * key, const char * options, int database, const char * pswd, unsigned timeout, unsigned expire);
+    ECL_REDIS_API void             ECL_REDIS_CALL SyncLockRGetUChar (ICodeContext * _ctx, size32_t & returnLength, UChar * & returnValue, const char * key, const char * options, int database, const char * pswd, unsigned timeout, unsigned expire);
 
-    ECL_REDIS_API bool ECL_REDIS_CALL SyncLockRMissThenLock(ICodeContext * _ctx, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout);
+    ECL_REDIS_API bool ECL_REDIS_CALL SyncLockRMissThenLock(ICodeContext * _ctx, const char * key, const char * options, int database, const char * password, unsigned __int64 timeout);
 }
 }
 #endif

testing/regress/ecl/key/redislockingtest.xml

@@ -71,10 +71,10 @@
  <Row><Result_24>false</Result_24></Row>
 </Dataset>
 <Dataset name='Result 25'>
- <Row><value>Redis Plugin: ERROR - the key &apos;channelTest1&apos;, on database 0, is locked with a channel (&apos;redis_ecl_lock_blah_blah_blah&apos;) different to that subscribed to (redis_ecl_lock_channelTest1_0).</value></Row>
+ <Row><value>Redis Plugin: ERROR - GetOrLock&lt;type&gt; &apos;channelTest1&apos; on database 0 for 127.0.0.1:6379 failed : key locked with a channel (&apos;redis_ecl_lock_blah_blah_blah&apos;) different to that subscribed to (redis_ecl_lock_channelTest1_0).</value></Row>
 </Dataset>
 <Dataset name='Result 26'>
- <Row><value>Redis Plugin: ERROR - GET timed out.</value></Row>
+ <Row><value>Redis Plugin: ERROR - GetOrLock&lt;type&gt; &apos;channelTest2&apos; on database 0 for 127.0.0.1:6379 failed : Resource temporarily unavailable</value></Row>
 </Dataset>
 <Dataset name='Result 27'>
  <Row><Result_27>databaseThenExpire</Result_27></Row>
@@ -127,3 +127,15 @@
 <Dataset name='Result 43'>
  <Row><Result_43>supercalifragilisticexpialidocious</Result_43></Row>
 </Dataset>
+<Dataset name='Result 44'>
+ <Row><Result_44></Result_44></Row>
+</Dataset>
+<Dataset name='Result 45'>
+ <Row><value>Redis Plugin: ERROR - GetOrLock&lt;type&gt; &apos;timeoutTest1&apos; on database 0 for 127.0.0.1:6379 failed : Resource temporarily unavailable</value></Row>
+</Dataset>
+<Dataset name='Result 46'>
+ <Row><Result_46></Result_46></Row>
+</Dataset>
+<Dataset name='Result 47'>
+ <Row><value>Timed Out</value></Row>
+</Dataset>

testing/regress/ecl/key/redissynctest.xml

@@ -92,8 +92,14 @@
  <Row><value>Auth Failed</value></Row>
 </Dataset>
 <Dataset name='Result 32'>
- <Row><value>Redis Plugin: ERROR - the requested key &apos;authTest1&apos; does not exist on database 0</value></Row>
+ <Row><value>Redis Plugin: ERROR - the requested key &apos;authTest1&apos; does not exist on database 0 on 127.0.0.1:6379</value></Row>
 </Dataset>
 <Dataset name='Result 33'>
- <Row><value>Redis Plugin: Connection failed - Connection refused for 127.0.0.1:9999</value></Row>
+ <Row><value>Redis Plugin: ERROR - connection for 127.0.0.1:9999 failed : Connection refused</value></Row>
+</Dataset>
+<Dataset name='Result 34'>
+ <Row><value>Redis Plugin: ERROR - unsupported option string &apos;blahblahblah&apos;</value></Row>
+</Dataset>
+<Dataset name='Result 35'>
+ <Row><value>Redis Plugin: ERROR - SELECT 16 on database 0 for 127.0.0.1:6379 failed : ERR invalid DB index</value></Row>
 </Dataset>

testing/regress/ecl/redislockingtest.ecl

@@ -20,7 +20,7 @@
 
 //nothor
 
-IMPORT redisServer FROM lib_redis;
+IMPORT * FROM lib_redis;
 IMPORT Std;
 
 STRING server := '--SERVER=127.0.0.1:6379';
@@ -183,4 +183,22 @@ SEQUENTIAL(
     myRedis.FlushDB(1);
     );
 
+//Test timeout
+myRedisNoTO := redisServerWithoutTimeout(server, password);
+dsTO := DATASET(NOFOLD(1), TRANSFORM({string value}, SELF.value := myRedisNoTO.GetOrLockString('timeoutTest' + (string)COUNTER,,,1000)));
+SEQUENTIAL(
+    myRedis.FlushDB();
+    myRedis.GetOrLockString('timeoutTest1');
+    OUTPUT(CATCH(dsTO, ONFAIL(TRANSFORM({ STRING value }, SELF.value := FAILMESSAGE))));
+    );
+
+STRING pluginTO := 'Redis Plugin: ERROR - function timed out internally.';
+STRING redisTO := 'Redis Plugin: ERROR - GetOrLock<type> \'timeoutTest2\' on database 0 for 127.0.0.1:6379 failed : Resource temporarily unavailable';
+dsTO2 := DATASET(NOFOLD(1), TRANSFORM({string value}, SELF.value := redis.GetOrLockString('timeoutTest' + (string)(1+COUNTER), server, /*database*/, password, 1/*ms*/)));
+SEQUENTIAL(
+    myRedis.FlushDB();
+    myRedis.GetOrLockString('timeoutTest2');
+    OUTPUT(CATCH(dsTO2, ONFAIL(TRANSFORM({ STRING value }, SELF.value := IF(FAILMESSAGE = pluginTO OR FAILMESSAGE = redisTO, 'Timed Out', 'Unexpected Error - ' + FAILMESSAGE)))));
+    );
+
 myRedis.FlushDB();

testing/regress/ecl/redissynctest.ecl

@@ -160,14 +160,14 @@ SEQUENTIAL(
     OUTPUT(SUM(NOFOLD(s1 + s2), a))//answer = (x+x/2)*N, in this case 300.
     );
 
-//Test some authentication exceptions
+//Test some exceptions
 myRedis4 := RedisServer(server);
-STRING noauth := 'Redis Plugin: server authentication failed - NOAUTH Authentication required.';
-STRING opNotPerm :=  'Redis Plugin: server authentication failed - ERR operation not permitted';
+STRING noauth := 'Redis Plugin: ERROR - authentication for 127.0.0.1:6379 failed : NOAUTH Authentication required.';
+STRING opNotPerm :=  'Redis Plugin: ERROR - authentication for 127.0.0.1:6379 failed : ERR operation not permitted';
 ds1 := DATASET(NOFOLD(1), TRANSFORM({string value}, SELF.value := myRedis4.GetString('authTest' + (string)COUNTER)));
 SEQUENTIAL(
     myRedis.FlushDB();
-    OUTPUT(CATCH(ds1, ONFAIL(TRANSFORM({ STRING value }, SELF.value := IF(FAILMESSAGE = noauth OR FAILMESSAGE = opNotPerm, 'Auth Failed', 'Unexpected Error')))));
+    OUTPUT(CATCH(ds1, ONFAIL(TRANSFORM({ STRING value }, SELF.value := IF(FAILMESSAGE = noauth OR FAILMESSAGE = opNotPerm, 'Auth Failed', 'Unexpected Error - ' + FAILMESSAGE)))));
     );
 
 ds2 := DATASET(NOFOLD(1), TRANSFORM({string value}, SELF.value := myRedis.GetString('authTest' + (string)COUNTER)));
@@ -183,5 +183,15 @@ SEQUENTIAL(
     OUTPUT(CATCH(ds3, ONFAIL(TRANSFORM({ STRING value }, SELF.value := FAILMESSAGE))));
     );
 
+ds4 := DATASET(NOFOLD(1), TRANSFORM({string value}, SELF.value := redis.GetString('option' + (string)COUNTER, 'blahblahblah')));
+SEQUENTIAL(
+    OUTPUT(CATCH(ds4, ONFAIL(TRANSFORM({ STRING value }, SELF.value := FAILMESSAGE))));
+    );
+
+ds5 := DATASET(NOFOLD(1), TRANSFORM({string value}, SELF.value := myRedis.GetString('maxDB' + (string)COUNTER, 16)));
+SEQUENTIAL(
+    OUTPUT(CATCH(ds5, ONFAIL(TRANSFORM({ STRING value }, SELF.value := FAILMESSAGE))));
+    );
+
 myRedis.FlushDB();
 myRedis2.FlushDB();