:bug: To avoid ambiguous, data type is mandatory in OR map SET operation.

7 files changed, 96 insertions, 35 deletions

diff --git a/CRDT/crdt_gtest.cpp b/CRDT/crdt_gtest.cpp
index b578e08..94515d4 100644
--- a/CRDT/crdt_gtest.cpp
+++ b/CRDT/crdt_gtest.cpp
@@ -1068,7 +1068,7 @@ TEST(ORMap, Basic)
 
 	const char *key2="lisi";
 	const char *val1="rich", *val2="poor";
-	ret=OR_map_set(map, key2, strlen(key2)+1, val1, strlen(val1)+1);
+	ret=OR_map_set_string(map, key2, strlen(key2)+1, val1, strlen(val1)+1);
 	EXPECT_EQ(ret, 1);
 	struct OR_map_kv *result=NULL;
 	result=OR_map_get(map,  key2, strlen(key2)+1);
@@ -1076,14 +1076,14 @@ TEST(ORMap, Basic)
 	OR_map_kv_free(result);
 	result=NULL;
 	
-	ret=OR_map_set(map, key2, strlen(key2)+1, val2, strlen(val2)+1);
+	ret=OR_map_set_string(map, key2, strlen(key2)+1, val2, strlen(val2)+1);
 	EXPECT_EQ(ret, 0);
 	result=OR_map_get(map,  key2, strlen(key2)+1);
 	EXPECT_STREQ(val2, result->string);
 	OR_map_kv_free(result);
 	result=NULL;
 
-	ret=OR_map_set(map, key1, strlen(key1)+1, val1, strlen(val1)+1);
+	ret=OR_map_set_string(map, key1, strlen(key1)+1, val1, strlen(val1)+1);
 	EXPECT_EQ(ret, 0);
 	long long integer_result=0;
 	ret=OR_map_incrby(map, key1, strlen(key1)+1, 0, &integer_result);
@@ -1138,7 +1138,7 @@ int my_key_filter(const char *key, size_t key_len, void *cb_arg)
 	}
 }
 
-TEST(ORMap, ReSetAndGet)
+TEST(ORMap, RepetitiveGet)
 {
 	uuid_t uuid;
 	uuid_generate(uuid);
@@ -1153,22 +1153,16 @@ TEST(ORMap, ReSetAndGet)
 	{
 		uuid_generate(keys[i]);
 		uuid_generate(values[i]);
+		ret = OR_map_set_string(map, (const char*)&keys[i], sizeof(keys[i]), (const char*)values[i], sizeof(values[i]));
+		EXPECT_EQ(ret, 1);
 	}
-
 	for(int j=0; j<ROUND; j++)
 	{
 		for(int i=0; i<KV_NUM; i++)
 		{
 			result=OR_map_get(map,  (const char*)&keys[i], sizeof(keys[i]));
-			if(result == NULL)
-			{
-				ret = 0;
-				ret = OR_map_set(map, (const char*)&keys[i], sizeof(keys[i]), (const char*)values[i], sizeof(values[i]));
-				EXPECT_EQ(ret, 1);
-				result=OR_map_get(map,  (const char*)&keys[i], sizeof(keys[i]));
-			}
-			EXPECT_TRUE(result);
-			EXPECT_EQ(result->type, TYPE_STRING);
+			ASSERT_TRUE(result);
+			ASSERT_EQ(result->type, TYPE_STRING);
 			EXPECT_EQ(memcmp(result->string, values[i], sizeof(values[i])), 0);
 			OR_map_kv_free(result);
 			result = NULL;
@@ -1185,20 +1179,21 @@ TEST(ORMap, Filter)
 
 	struct OR_map *map=OR_map_new(uuid);
 	int ret=0;
-	char key[32], value[32];
+	char key[32];
+	long long value=0;
 	size_t n_prefixed_key=8;
 	const char *prefix="a_";
 	for(size_t i=0; i<n_prefixed_key; i++)
 	{
 		snprintf(key, sizeof(key), "%s%zu", prefix, i);
-		snprintf(value, sizeof(value), "%zu", i);
-		ret=OR_map_set(map, key, strlen(key)+1, value, strlen(value)+1);
+		value=i;
+		ret=OR_map_set_integer(map, key, strlen(key)+1, value);
 	}
 	for(size_t i=0; i<12; i++)
 	{
 		snprintf(key, sizeof(key), "b_%zu", i);
-		snprintf(value, sizeof(value), "%zu", i);
-		ret=OR_map_set(map, key, strlen(key)+1, value, strlen(value)+1);
+		value=i;
+		ret=OR_map_set_integer(map, key, strlen(key)+1, value);
 	}
 
 	struct OR_map_kv **kv_list=NULL;
@@ -1271,7 +1266,7 @@ TEST(ORMap, SetGet)
 	{
 		snprintf(key_buffer, sizeof(key_buffer), "%s-%zu", key_prefix, i);
 		snprintf(val_buffer, sizeof(val_buffer), "%s-%zu", val_prefix, i);
-		OR_map_set(map[0], key_buffer, strlen(key_buffer)+1, val_buffer, strlen(val_buffer)+1);
+		OR_map_set_string(map[0], key_buffer, strlen(key_buffer)+1, val_buffer, strlen(val_buffer)+1);
 	}
 	char *blob=NULL;
 	size_t blob_sz=0;
@@ -1400,7 +1395,7 @@ TEST(ORMap, Remove)
 		else
 		{
 			snprintf(val_buffer, sizeof(val_buffer), "%s-%zu", val_prefix, i);			
-			OR_map_set(map[i%2], key_buffer, strlen(key_buffer)+1, val_buffer, strlen(val_buffer)+1);
+			OR_map_set_string(map[i%2], key_buffer, strlen(key_buffer)+1, val_buffer, strlen(val_buffer)+1);
 		}
 	}
 	char *blob=NULL;
@@ -1467,7 +1462,7 @@ TEST(ORMap, ObservedRemove)
 	{
 		snprintf(key_buffer, sizeof(key_buffer), "%s-%zu", key_prefix, i);
 		snprintf(val_buffer, sizeof(val_buffer), "%s-%zu", val_prefix, i);			
-		OR_map_set(map[i%2], key_buffer, strlen(key_buffer)+1, val_buffer, strlen(val_buffer)+1);
+		OR_map_set_string(map[i%2], key_buffer, strlen(key_buffer)+1, val_buffer, strlen(val_buffer)+1);
 		
 	}
 	char *blob=NULL;
@@ -1483,7 +1478,7 @@ TEST(ORMap, ObservedRemove)
 	ret=OR_map_remove(map[0], key_buffer, strlen(key_buffer)+1);
 	EXPECT_EQ(ret, 1);
 	const char *val="Hello Stellar!";
-	ret=OR_map_set(map[1], key_buffer, strlen(key_buffer)+1, val, strlen(val)+1);
+	ret=OR_map_set_string(map[1], key_buffer, strlen(key_buffer)+1, val, strlen(val)+1);
 
 	OR_map_serialize(map[0], &blob, &blob_sz);
 	OR_map_merge_blob(map[1], blob, blob_sz);
@@ -1532,7 +1527,7 @@ TEST(ORMap, Replica32)
 				break;
 			case 1:
 				snprintf(val_buffer, sizeof(val_buffer), "%s-%zu", val_prefix, i);			
-				OR_map_set(map[i%replica_number], key_buffer, strlen(key_buffer)+1, val_buffer, strlen(val_buffer)+1);
+				OR_map_set_string(map[i%replica_number], key_buffer, strlen(key_buffer)+1, val_buffer, strlen(val_buffer)+1);
 				break;
 			case 2:
 				OR_map_add(map[i%replica_number], key_buffer, strlen(key_buffer)+1);
@@ -1587,7 +1582,7 @@ TEST(ORMap, Replica32)
 				break;
 			case 1:
 				snprintf(val_buffer, sizeof(val_buffer), "%s-%zu-updated", val_prefix, i);			
-				OR_map_set(map[i%replica_number], key_buffer, strlen(key_buffer)+1, val_buffer, strlen(val_buffer)+1);
+				OR_map_set_string(map[i%replica_number], key_buffer, strlen(key_buffer)+1, val_buffer, strlen(val_buffer)+1);
 				break;
 			case 2:
 				OR_map_remove(map[i%replica_number], key_buffer, strlen(key_buffer)+1);
diff --git a/CRDT/or_map.c b/CRDT/or_map.c
index eeaa661..beeae4c 100644
--- a/CRDT/or_map.c
+++ b/CRDT/or_map.c
@@ -231,8 +231,28 @@ int OR_map_set(struct OR_map *map, const char *key, size_t key_len, const char *
 	int added=0, ret=0;
 	added=OR_record_create_if_not_exist(map, key, key_len, &record);
 	char *endptr=NULL;
-	long long int_value=strtol(value, &endptr, 10);
-	if(*endptr=='\0')
+	int is_integer=0;
+	if(value_sz<21)
+	{
+		for(size_t i=0; i<value_sz; i++)
+		{
+			if((value[i]<'0' || value[i]>'9') && value[i]!='-')
+			{
+				continue;
+			}
+			else
+			{
+				is_integer=0;
+				break;
+			}
+		}
+	}
+	long long int_value=0;
+	if(is_integer)
+	{
+		int_value=strtol(value, &endptr, 10);
+	}
+	if(*endptr=='\0' && is_integer)
 	{
 		ret=OR_record_set_int(record, int_value);
 	}
@@ -246,6 +266,28 @@ int OR_map_set(struct OR_map *map, const char *key, size_t key_len, const char *
 	
 	return added;
 }
+int OR_map_set_string(struct OR_map *map, const char *key, size_t key_len, const char *value, size_t value_sz)
+{
+	struct OR_record *record=NULL;
+	int added=0, ret=0;
+	added=OR_record_create_if_not_exist(map, key, key_len, &record);
+	ret=OR_record_set_str(record, value, value_sz);
+	if(ret<0) return ret;
+	unsigned long long new_sequence=OR_map_step(map);
+	OR_record_step(record, map->my_id, new_sequence);
+	return added;
+}
+int OR_map_set_integer(struct OR_map *map, const char *key, size_t key_len, long long int_value)
+{
+	struct OR_record *record=NULL;
+	int added=0, ret=0;
+	added=OR_record_create_if_not_exist(map, key, key_len, &record);
+	ret=OR_record_set_int(record, int_value);
+	if(ret<0) return ret;
+	unsigned long long new_sequence=OR_map_step(map);
+	OR_record_step(record, map->my_id, new_sequence);
+	return added;
+}
 int OR_map_set_score(struct OR_map *map, const char *key, size_t key_len, long long score)
 {
 	struct OR_record *record=NULL;
diff --git a/CRDT/or_map.h b/CRDT/or_map.h
index f2bfd85..82522a3 100644
--- a/CRDT/or_map.h
+++ b/CRDT/or_map.h
@@ -15,7 +15,8 @@ struct OR_map;
 struct OR_map *OR_map_new(uuid_t uuid);
 void OR_map_free(struct OR_map *map);
 int OR_map_add(struct OR_map *map, const char *key, size_t key_len);
-int OR_map_set(struct OR_map *map, const char *key, size_t key_len, const char *value, size_t value_len);
+int OR_map_set_string(struct OR_map *map, const char *key, size_t key_len, const char *value, size_t value_sz);
+int OR_map_set_integer(struct OR_map *map, const char *key, size_t key_len, long long int_value);
 int OR_map_incrby(struct OR_map *map, const char *key, size_t key_len, long long increment, long long *result);
 int OR_map_remove(struct OR_map *map, const char *key, size_t key_len);
 int OR_map_exisits(const struct OR_map *map, const char* key, size_t key_len);
diff --git a/src/inc_internal/swarmkv_utils.h b/src/inc_internal/swarmkv_utils.h
index 5209305..55f5dbc 100644
--- a/src/inc_internal/swarmkv_utils.h
+++ b/src/inc_internal/swarmkv_utils.h
@@ -1,6 +1,7 @@
 #pragma once
 
 #include <assert.h>
+#include <stddef.h>
 
 #define ALLOC(type, number) ((type *)calloc(sizeof(type), number))
 #define	FREE(p)	{free(*p);*p=NULL;}
@@ -41,5 +42,5 @@ char *toLower(char* s);
 char *toUpper(char* s);
 int stringmatchlen(const char *pattern, int patternLen, const char *string, int stringLen, int nocase);
 long long ustime(void);
-int is_number(const char *string, long long *number);
+int is_number(const char *string, size_t sz, long long *number);
 int is_double(const char *string, double *number);
\ No newline at end of file
diff --git a/src/swarmkv_utils.c b/src/swarmkv_utils.c
index e54118d..ab0b920 100644
--- a/src/swarmkv_utils.c
+++ b/src/swarmkv_utils.c
@@ -188,13 +188,26 @@ long long ustime(void) {
     ust += tv.tv_nsec/1000;
     return ust;
 }
-int is_number(const char *string, long long *number)
+int is_number(const char *string, size_t sz, long long *number)
 {
+    if(sz>20) return 0;
+
+    for(size_t i=0; i<sz ; i++)
+    {
+        if((string[i]<'0' || string[i]>'9') && string[i]!='-')
+        {
+            return 0;
+        }
+        if(string[i]=='\0' && i<sz-1)
+        {
+            return 0;
+        }
+    }
 	char *endptr=NULL;
 	*number=strtol(string, &endptr, 10);
 	if(*endptr=='\0')
 	{
-		return 1;		
+		return 1;
 	}
     else
     {
diff --git a/src/t_hash.c b/src/t_hash.c
index 3124ea4..68584a0 100644
--- a/src/t_hash.c
+++ b/src/t_hash.c
@@ -1,4 +1,5 @@
 #include "swarmkv_common.h"
+#include "swarmkv_utils.h"
 #include "swarmkv_store.h"
 #include "swarmkv_error.h"
 #include "or_map.h"
@@ -48,16 +49,24 @@ enum cmd_exec_result hset_command(struct swarmkv_module *mod_store, const struct
 
 	sds field;
     sds value;
-		
+	long long int_val=0;
+
 	if(obj->type==OBJ_TYPE_HASH)
 	{
 		for(i=0; i<n_to_add; i++)
 		{
             field=cmd->argv[2+i*2];
             value=cmd->argv[2+i*2+1];
-
-            ret=OR_map_set(obj->hash, field, sdslen(field),
+			if(is_number(value, sdslen(value), &int_val))
+			{
+				ret=OR_map_set_integer(obj->hash, field, sdslen(field), int_val);
+			}
+			else
+			{
+           		ret=OR_map_set_string(obj->hash, field, sdslen(field),
                             value, sdslen(value));
+			}
+
 			if(ret>0) n_added++;
 		}
 		*reply=swarmkv_reply_new_integer(n_added);
diff --git a/tools/swarmkv_cli.c b/tools/swarmkv_cli.c
index 3419b49..c04961a 100644
--- a/tools/swarmkv_cli.c
+++ b/tools/swarmkv_cli.c
@@ -1297,7 +1297,7 @@ int main(int argc, char * argv[])
 		run_times=1;
 		run_interval_sec=0;
 		offset=0;
-		if(is_number(exec_argv[0], &run_times))
+		if(is_number(exec_argv[0], strlen(exec_argv[0])+1, &run_times))
 		{
 			offset++;
 			if(exec_argc>1 && is_double(exec_argv[1], &run_interval_sec))