Merge pull request #2 from tlexy/develop

Develop

Author: tlexy

rclientpp/core/async_socket_client.cpp

@@ -11,8 +11,13 @@ namespace rcpp {
 		//FD_ZERO(&_efds);
 	}
 
-	AsyncSocketClient::AsyncSocketClient(int sockfd)
-	{}
+	/*AsyncSocketClient::AsyncSocketClient(int sockfd)
+	{}*/
+
+	AsyncSocketClient::~AsyncSocketClient()
+	{
+		close();
+	}
 
 	int AsyncSocketClient::sockfd() const
 	{

rclientpp/core/async_socket_client.h

@@ -16,7 +16,7 @@ namespace rcpp {
 	{
 	public:
 		AsyncSocketClient();
-		AsyncSocketClient(int sockfd);
+		//AsyncSocketClient(int sockfd);
 
 		int sockfd() const;
 
@@ -30,6 +30,8 @@ namespace rcpp {
 
 		void close();
 
+		~AsyncSocketClient();
+
 	private:
 		int _sockfd{ -1 };
 		fd_set _rfds;

rclientpp/core/rclient.cpp

@@ -19,16 +19,18 @@ int RClient::_read_timeout = 30000;
 RClient::RClient(const std::string& ipstr, int port)
 	:_ipstr(ipstr),
 	_port(port),
-	_bufptr(std::make_shared<RClientBuffer>(1024)),
+	_bufptr(std::make_shared<RClientBuffer>(1024*1024)),
 	_map_parser(std::make_shared<MapParser>()),
 	_array_parser(std::make_shared<ArrayParser>()),
 	_aclient(std::make_shared<AsyncSocketClient>())
 {
-	_strerr.resize(128, '\0');
+	_strerr.resize(256, '\0');
 }
 
 int RClient::connect(const std::string& username, const std::string& auth, int timeoutms)
 {
+	_user = username;
+	_pass = auth;
 	bool flag = _aclient->connect(_ipstr.c_str(), _port, timeoutms);
 	if (!flag)
 	{
@@ -52,6 +54,7 @@ int RClient::connect(const std::string& username, const std::string& auth, int t
 
 int RClient::connect(const std::string& auth, int timeoutms)
 {
+	_pass = auth;
 	bool flag = _aclient->connect(_ipstr.c_str(), _port, timeoutms);
 	if (!flag)
 	{
@@ -91,6 +94,20 @@ int RClient::connect(int timeoutms)
 	return _sock_param();
 }
 
+int RClient::reconnect(int timeoutms)
+{
+	_aclient = std::make_shared<AsyncSocketClient>();
+	if (_user.size() == 0 && _pass.size() == 0)
+	{
+		return connect(timeoutms);
+	}
+	if (_pass.size() > 0 && _user.size() > 0)
+	{
+		return connect(_user, _pass, timeoutms);
+	}
+	return connect(_pass, timeoutms);
+}
+
 int RClient::_sock_param()
 {
 	/*int ret = sockets::SetRecvTimeout(_sockfd, _read_timeout);
@@ -166,6 +183,11 @@ std::string RClient::strerror()
 	return _strerr;
 }
 
+void RClient::set_error_str(const std::string& str)
+{
+	_strerr = str;
+}
+
 void RClient::set_read_timeout(int millisec)
 {
 	_read_timeout = millisec;
@@ -407,6 +429,16 @@ std::shared_ptr<BaseValue> RClient::get_results(int& ret_code)
 			//ret_code = PARSE_FORMAT_ERROR;
 		}
 	}
+	else if (text[0] == '>')
+	{
+		_bufptr->has_read(1);
+		result = std::make_shared<RedisComplexValue>(ParserType::Push);
+		ret = _array_parser->parse(_bufptr, result);
+		if (ret != 0)
+		{
+			//ret_code = PARSE_FORMAT_ERROR;
+		}
+	}
 	else
 	{
 		return nullptr;

rclientpp/core/rclient.h

@@ -20,6 +20,8 @@ class RClient
 	int connect(const std::string& auth, int timeoutms = 10000);
 	int connect(int timeoutms = 10000);
 
+	int reconnect(int timeoutms = 500);
+
 	int use_resp3();
 	int use_resp2();
 
@@ -38,6 +40,8 @@ class RClient
 
 	std::string strerror();
 
+	void set_error_str(const std::string&);
+
 private:
 	int do_connect(const std::string& cmd);
 
@@ -51,6 +55,9 @@ class RClient
 
 	std::string _ipstr;
 	int _port;
+	std::string _user;
+	std::string _pass;
+
 	int _err_code;
 	static int _read_timeout;
 	std::string _strerr;

rclientpp/core/rclient_def.h

@@ -134,6 +134,7 @@ enum class ParserType
 	BlobError,
 	Verbatim,
 	BigNumber,
+	Push,
 
 };
 
@@ -144,6 +145,9 @@ class BaseValue
 		:type_(type)
 	{}
 
+	virtual int64_t get_number() = 0;
+	virtual std::string get_string() = 0;
+
 	ParserType value_type() const
 	{
 		return type_;
@@ -172,6 +176,15 @@ class BaseValue
 		return false;
 	}
 
+	bool is_number() const
+	{
+		if (type_ == ParserType::Number)
+		{
+			return true;
+		}
+		return false;
+	}
+
 	bool is_digit() const
 	{
 		if (type_ == ParserType::Number
@@ -233,6 +246,16 @@ class RedisValue : public BaseValue
 	{
 	}
 
+	virtual int64_t get_number()
+	{
+		return u.int_val_;
+	}
+
+	virtual std::string get_string()
+	{
+		return str_val_;
+	}
+
 	virtual bool is_ok()
 	{
 		if (value_type() == ParserType::SimpleString)
@@ -262,6 +285,15 @@ class RedisComplexValue : public BaseValue
 		:BaseValue(type)
 	{
 	}
+	virtual int64_t get_number()
+	{
+		return 0;
+	}
+
+	virtual std::string get_string()
+	{
+		return std::string("");
+	}
 	std::list<std::shared_ptr<BaseValue>> results;
 	int count{0};
 };

rclientpp/ds/rd.cpp

@@ -13,6 +13,16 @@ void Rd::set_connection(std::shared_ptr<RClient> connection)
 	_client = connection;
 }
 
+void Rd::set_err_str(const std::string& str)
+{
+	_client->set_error_str(str);
+}
+
+std::string Rd::get_err_str()
+{
+	return _client->strerror();
+}
+
 std::shared_ptr<BaseValue> Rd::redis_command(const char* cmd, int len, int& ret_code)
 {
 	int ret = _client->command(cmd, len);

rclientpp/ds/rd.h

@@ -23,6 +23,9 @@ class Rd
 	bool get_boolean(const std::string& cmd);
 	//int get_integer(const std::string& cmd);
 
+	void set_err_str(const std::string&);
+	std::string get_err_str();
+
 protected:
 	std::shared_ptr<RClient> _client;
 	std::string _crlf;