kapitan/IXWebSocket
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

(ws) add bare bone redis-cli like sub-command, with command line editing powered by libnoise

Browse Source
This commit is contained in:
Benjamin Sergeant
2020-06-11 17:30:42 -07:00
parent 35d76c20dc
commit ac9710d5d6
9 changed files with 182 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
docs/CHANGELOG.md
View File

@@ -1,6 +1,10 @@
# Changelog
All changes to this project will be documented in this file.
## [9.7.2] - 2020-06-11
(ws) add bare bone redis-cli like sub-command, with command line editing powered by libnoise
## [9.7.1] - 2020-06-11
(redis cobra bots) ws cobra metrics to redis / hostname invalid parsing

100
ixredis/ixredis/IXRedisClient.cpp
View File

@@ -354,4 +354,104 @@ namespace ix
return success;
}
std::pair<RespType, std::string> RedisClient::send(
const std::vector<std::string>& args,
std::string& errMsg)
{
std::stringstream ss;
ss << "*";
ss << std::to_string(args.size());
ss << "\r\n";
for (auto&& arg : args)
{
ss << writeString(arg);
}
bool sent = _socket->writeBytes(ss.str(), nullptr);
if (!sent)
{
errMsg = "Cannot write bytes to socket";
return std::make_pair(RespType::Error, "");
}
return readResponse(errMsg);
}
std::pair<RespType, std::string> RedisClient::readResponse(std::string& errMsg)
{
// Read result
auto pollResult = _socket->isReadyToRead(-1);
if (pollResult == PollResultType::Error)
{
errMsg = "Error while polling for result";
return std::make_pair(RespType::Error, "");
}
// First line is the string length
auto lineResult = _socket->readLine(nullptr);
auto lineValid = lineResult.first;
auto line = lineResult.second;
if (!lineValid)
{
errMsg = "Error while polling for result";
return std::make_pair(RespType::Error, "");
}
std::string response;
if (line[0] == '+') // Simple string
{
std::stringstream ss;
response = line.substr(1, line.size() - 3);
return std::make_pair(RespType::String, response);
}
else if (line[0] == '-') // Errors
{
std::stringstream ss;
response = line.substr(1, line.size() - 3);
return std::make_pair(RespType::Error, response);
}
else if (line[0] == ':') // Integers
{
std::stringstream ss;
response = line.substr(1, line.size() - 3);
return std::make_pair(RespType::Integer, response);
}
else if (line[0] == '$') // Bulk strings
{
int stringSize;
{
std::stringstream ss;
ss << line.substr(1, line.size() - 1);
ss >> stringSize;
}
// Read the result, which is the stream id computed by the redis server
lineResult = _socket->readLine(nullptr);
lineValid = lineResult.first;
line = lineResult.second;
std::string str = line.substr(0, stringSize);
return std::make_pair(RespType::String, str);
}
else
{
errMsg = "Unhandled response type";
return std::make_pair(RespType::Unknown, std::string());
}
}
std::string RedisClient::getRespTypeDescription(RespType respType)
{
switch (respType)
{
case RespType::Integer: return "integer";
case RespType::Error: return "error";
case RespType::String: return "string";
default: return "unknown";
}
}
} // namespace ix

20
ixredis/ixredis/IXRedisClient.h
View File

@@ -14,6 +14,14 @@
namespace ix
{
enum class RespType : int
{
String = 0,
Error = 1,
Integer = 2,
Unknown = 3
};
class RedisClient
{
public:
@@ -42,14 +50,20 @@ namespace ix
const std::string& message,
int maxLen,
std::string& errMsg);
std::string prepareXaddCommand(const std::string& stream,
const std::string& message,
int maxLen);
std::string readXaddReply(std::string& errMsg);
bool sendCommand(
const std::string& commands, int commandsCount, std::string& errMsg);
bool sendCommand(const std::string& commands, int commandsCount, std::string& errMsg);
// Arbitrary commands
std::pair<RespType, std::string> send(
const std::vector<std::string>& args,
std::string& errMsg);
std::pair<RespType, std::string> readResponse(std::string& errMsg);
std::string getRespTypeDescription(RespType respType);
void stop();

2
ixwebsocket/IXWebSocketVersion.h
View File

@@ -6,4 +6,4 @@
#pragma once
#define IX_WEBSOCKET_VERSION "9.7.1"
#define IX_WEBSOCKET_VERSION "9.7.2"

3
makefile
View File

@@ -233,6 +233,9 @@ install_cmake_for_linux:
doc:
mkdocs gh-deploy
change:
vi ixwebsocket/IXWebSocketVersion.h docs/CHANGELOG.md
.PHONY: test
.PHONY: build
.PHONY: ws

1
ws/CMakeLists.txt
View File

@@ -45,6 +45,7 @@ add_executable(ws
ws_transfer.cpp
ws_send.cpp
ws_receive.cpp
ws_redis_cli.cpp
ws_redis_publish.cpp
ws_redis_subscribe.cpp
ws_redis_server.cpp

10
ws/ws.cpp
View File

@@ -281,6 +281,12 @@ int main(int argc, char** argv)
httpClientApp->add_option("--transfer-timeout", transferTimeout, "Transfer timeout");
addTLSOptions(httpClientApp);
CLI::App* redisCliApp = app.add_subcommand("redis_cli", "Redis cli");
redisCliApp->fallthrough();
redisCliApp->add_option("--port", redisPort, "Port");
redisCliApp->add_option("--host", hostname, "Hostname");
redisCliApp->add_option("--password", password, "Password");
CLI::App* redisPublishApp = app.add_subcommand("redis_publish", "Redis publisher");
redisPublishApp->fallthrough();
redisPublishApp->add_option("--port", redisPort, "Port");
@@ -531,6 +537,10 @@ int main(int argc, char** argv)
compress,
tlsOptions);
}
else if (app.got_subcommand("redis_cli"))
{
ret = ix::ws_redis_cli_main(hostname, redisPort, password);
}
else if (app.got_subcommand("redis_publish"))
{
ret = ix::ws_redis_publish_main(hostname, redisPort, password, channel, message, count);

4
ws/ws.h
View File

@@ -64,6 +64,10 @@ namespace ix
bool disablePerMessageDeflate,
const ix::SocketTLSOptions& tlsOptions);
int ws_redis_cli_main(const std::string& hostname,
int port,
const std::string& password);
int ws_redis_publish_main(const std::string& hostname,
int port,
const std::string& password,

50
ws/ws_redis_cli.cpp
View File

@@ -8,6 +8,7 @@
#include <spdlog/spdlog.h>
#include <sstream>
#include <iostream>
#include "linenoise.hpp"
namespace ix
{
@@ -36,17 +37,50 @@ namespace ix
while (true)
{
std::string text;
std::cout << "> " << std::flush;
std::getline(std::cin, text);
// Read line
std::string line;
std::string prompt;
prompt += hostname;
prompt += ":";
prompt += std::to_string(port);
prompt += "> ";
auto quit = linenoise::Readline(prompt.c_str(), line);
#if 0
if (!redisClient.send(args, errMsg))
if (quit)
{
spdlog::error("Error", channel, errMsg);
return 1;
break;
}
#endif
std::stringstream ss(line);
std::vector<std::string> args;
std::string arg;
while (ss.good())
{
ss >> arg;
args.push_back(arg);
}
std::string errMsg;
auto response = redisClient.send(args, errMsg);
if (!errMsg.empty())
{
spdlog::error("(error) {}", errMsg);
}
else
{
if (response.first != RespType::String)
{
std::cout << "("
<< redisClient.getRespTypeDescription(response.first)
<< ")"
<< " ";
}
std::cout << response.second << std::endl;
}
linenoise::AddHistory(line.c_str());
}
return 0;