Implement RedisWriter::ExecuteQuery

refs #4991

lib/redis/rediswriter.cpp

@@ -331,3 +331,36 @@ void RedisWriter::AssertOnWorkQueue(void)
 {
 	ASSERT(m_WorkQueue.IsWorkerThread());
 }
+
+boost::shared_ptr<redisReply> RedisWriter::ExecuteQuery(const std::vector<String>& query)
+{
+	const char **argv;
+	size_t *argvlen;
+
+	argv = new const char *[query.size()];
+	argvlen = new size_t[query.size()];
+
+	for (std::vector<String>::size_type i = 0; i < query.size(); i++) {
+		argv[i] = query[i].CStr();
+		argvlen[i] = query[i].GetLength();
+	}
+
+	redisReply *reply = reinterpret_cast<redisReply *>(redisCommandArgv(m_Context, query.size(), argv, argvlen));
+
+	if (reply->type == REDIS_REPLY_ERROR) {
+		Log(LogInformation, "RedisWriter")
+		    << "Redis query failed: " << reply->str;
+
+		String msg = reply->str;
+
+		freeReplyObject(reply);
+
+		BOOST_THROW_EXCEPTION(
+		    redis_error()
+			<< errinfo_message(msg)
+			<< errinfo_redis_query(Utility::Join(Array::FromVector(query), ' ', false))
+		);
+	}
+
+	return boost::shared_ptr<redisReply>(reply);
+}

lib/redis/rediswriter.hpp

@@ -81,6 +81,8 @@ private:
 
 	void AssertOnWorkQueue(void);
 
+	boost::shared_ptr<redisReply> ExecuteQuery(const std::vector<String>& query);
+
 	Timer::Ptr m_ReconnectTimer;
 	Timer::Ptr m_SubscriptionTimer;
 	WorkQueue m_WorkQueue;
@@ -88,6 +90,11 @@ private:
 	std::map<String, RedisSubscriptionInfo> m_Subscriptions;
 };
 
+struct redis_error : virtual std::exception, virtual boost::exception { };
+
+struct errinfo_redis_query_;
+typedef boost::error_info<struct errinfo_redis_query_, std::string> errinfo_redis_query;
+
 }
 
 #endif /* REDISWRITER_H */