refactored tests (#147)
This commit is contained in:
Timofey Turenko
21
maxscale-system-test/cdc_connector/CMakeLists.txt
Normal file
21
maxscale-system-test/cdc_connector/CMakeLists.txt
Normal file
@ -0,0 +1,21 @@
|
||||
project(cdc_connector)
|
||||
cmake_minimum_required(VERSION 2.8)
|
||||
|
||||
enable_testing()
|
||||
|
||||
include(ExternalProject)
|
||||
include(cmake/BuildJansson.cmake)
|
||||
|
||||
set(CMAKE_CXX_FLAGS "-fPIC -std=c++0x")
|
||||
set(CMAKE_CXX_FLAGS_DEBUG "-fPIC -std=c++0x -ggdb")
|
||||
set(CMAKE_CXX_FLAGS_RELEASE "-fPIC -std=c++0x -O2")
|
||||
set(CMAKE_CXX_FLAGS_RELWITHDEBINFO "-fPIC -std=c++0x -O2")
|
||||
|
||||
include_directories(${JANSSON_INCLUDE_DIR})
|
||||
add_library(cdc_connector SHARED cdc_connector.cpp)
|
||||
target_link_libraries(cdc_connector ${JANSSON_LIBRARIES} crypto)
|
||||
add_dependencies(cdc_connector jansson)
|
||||
install(TARGETS cdc_connector DESTINATION lib)
|
||||
install(FILES cdc_connector.h DESTINATION include)
|
||||
|
||||
include(CTest)
|
||||
454
maxscale-system-test/cdc_connector/cdc_connector.cpp
Normal file
454
maxscale-system-test/cdc_connector/cdc_connector.cpp
Normal file
@ -0,0 +1,454 @@
|
||||
#include "cdc_connector.h"
|
||||
|
||||
#include <arpa/inet.h>
|
||||
#include <stdexcept>
|
||||
#include <unistd.h>
|
||||
#include <string.h>
|
||||
#include <sstream>
|
||||
#include <openssl/sha.h>
|
||||
#include <sys/socket.h>
|
||||
#include <sys/types.h>
|
||||
#include <jansson.h>
|
||||
|
||||
#define CDC_CONNECTOR_VERSION "1.0.0"
|
||||
|
||||
#define ERRBUF_SIZE 512
|
||||
#define READBUF_SIZE 1024
|
||||
|
||||
static const char OK_RESPONSE[] = "OK\n";
|
||||
|
||||
static const char CLOSE_MSG[] = "CLOSE";
|
||||
static const char REGISTER_MSG[] = "REGISTER UUID=CDC_CONNECTOR-" CDC_CONNECTOR_VERSION ", TYPE=";
|
||||
static const char REQUEST_MSG[] = "REQUEST-DATA ";
|
||||
|
||||
namespace
|
||||
{
|
||||
|
||||
static inline int nointr_read(int fd, void *dest, size_t size)
|
||||
{
|
||||
int rc = read(fd, dest, size);
|
||||
|
||||
while (rc == -1 && errno == EINTR)
|
||||
{
|
||||
rc = read(fd, dest, size);
|
||||
}
|
||||
|
||||
return rc;
|
||||
}
|
||||
|
||||
static inline int nointr_write(int fd, const void *src, size_t size)
|
||||
{
|
||||
int rc = write(fd, src, size);
|
||||
|
||||
while (rc == -1 && errno == EINTR)
|
||||
{
|
||||
rc = write(fd, src, size);
|
||||
}
|
||||
|
||||
return rc;
|
||||
}
|
||||
|
||||
static std::string bin2hex(const uint8_t *data, size_t len)
|
||||
{
|
||||
std::string result;
|
||||
static const char hexconvtab[] = "0123456789abcdef";
|
||||
|
||||
for (int i = 0; i < len; i++)
|
||||
{
|
||||
result += hexconvtab[data[i] >> 4];
|
||||
result += hexconvtab[data[i] & 0x0f];
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
std::string generateAuthString(const std::string& user, const std::string& password)
|
||||
{
|
||||
uint8_t digest[SHA_DIGEST_LENGTH];
|
||||
SHA1(reinterpret_cast<const uint8_t*> (password.c_str()), password.length(), digest);
|
||||
|
||||
std::string auth_str = user;
|
||||
auth_str += ":";
|
||||
|
||||
std::string part1 = bin2hex((uint8_t*)auth_str.c_str(), auth_str.length());
|
||||
std::string part2 = bin2hex(digest, sizeof(digest));
|
||||
|
||||
return part1 + part2;
|
||||
}
|
||||
|
||||
|
||||
std::string json_to_string(json_t* json)
|
||||
{
|
||||
std::stringstream ss;
|
||||
|
||||
switch (json_typeof(json))
|
||||
{
|
||||
case JSON_STRING:
|
||||
ss << json_string_value(json);
|
||||
break;
|
||||
|
||||
case JSON_INTEGER:
|
||||
ss << json_integer_value(json);
|
||||
break;
|
||||
|
||||
case JSON_REAL:
|
||||
ss << json_real_value(json);
|
||||
break;
|
||||
|
||||
case JSON_TRUE:
|
||||
ss << "true";
|
||||
break;
|
||||
|
||||
case JSON_FALSE:
|
||||
ss << "false";
|
||||
break;
|
||||
|
||||
case JSON_NULL:
|
||||
break;
|
||||
|
||||
default:
|
||||
break;
|
||||
|
||||
}
|
||||
|
||||
return ss.str();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
namespace CDC
|
||||
{
|
||||
|
||||
/**
|
||||
* Public functions
|
||||
*/
|
||||
|
||||
Connection::Connection(const std::string& address,
|
||||
uint16_t port,
|
||||
const std::string& user,
|
||||
const std::string& password,
|
||||
uint32_t flags) :
|
||||
m_fd(-1),
|
||||
m_address(address),
|
||||
m_port(port),
|
||||
m_user(user),
|
||||
m_password(password),
|
||||
m_flags(flags)
|
||||
{
|
||||
}
|
||||
|
||||
Connection::~Connection()
|
||||
{
|
||||
closeConnection();
|
||||
}
|
||||
|
||||
bool Connection::createConnection()
|
||||
{
|
||||
bool rval = false;
|
||||
struct sockaddr_in remote = {};
|
||||
|
||||
remote.sin_port = htons(m_port);
|
||||
remote.sin_family = AF_INET;
|
||||
|
||||
if (inet_aton(m_address.c_str(), (struct in_addr*)&remote.sin_addr.s_addr) == 0)
|
||||
{
|
||||
m_error = "Invalid address: ";
|
||||
m_error += m_address;
|
||||
}
|
||||
else
|
||||
{
|
||||
int fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
|
||||
|
||||
if (fd == -1)
|
||||
{
|
||||
char err[ERRBUF_SIZE];
|
||||
m_error = "Failed to create socket: ";
|
||||
m_error += strerror_r(errno, err, sizeof (err));
|
||||
}
|
||||
|
||||
m_fd = fd;
|
||||
|
||||
if (connect(fd, (struct sockaddr*) &remote, sizeof (remote)) == -1)
|
||||
{
|
||||
char err[ERRBUF_SIZE];
|
||||
m_error = "Failed to connect: ";
|
||||
m_error += strerror_r(errno, err, sizeof (err));
|
||||
}
|
||||
else if (doAuth())
|
||||
{
|
||||
rval = doRegistration();
|
||||
}
|
||||
}
|
||||
|
||||
return rval;
|
||||
}
|
||||
|
||||
void Connection::closeConnection()
|
||||
{
|
||||
if (m_fd != -1)
|
||||
{
|
||||
nointr_write(m_fd, CLOSE_MSG, sizeof (CLOSE_MSG) - 1);
|
||||
close(m_fd);
|
||||
m_fd = -1;
|
||||
}
|
||||
}
|
||||
|
||||
bool Connection::requestData(const std::string& table, const std::string& gtid)
|
||||
{
|
||||
bool rval = true;
|
||||
|
||||
std::string req_msg(REQUEST_MSG);
|
||||
req_msg += table;
|
||||
|
||||
if (gtid.length())
|
||||
{
|
||||
req_msg += " ";
|
||||
req_msg += gtid;
|
||||
}
|
||||
|
||||
if (nointr_write(m_fd, req_msg.c_str(), req_msg.length()) == -1)
|
||||
{
|
||||
rval = false;
|
||||
char err[ERRBUF_SIZE];
|
||||
m_error = "Failed to write request: ";
|
||||
m_error += strerror_r(errno, err, sizeof (err));
|
||||
}
|
||||
|
||||
return rval;
|
||||
}
|
||||
|
||||
static inline bool is_schema(json_t* json)
|
||||
{
|
||||
bool rval = false;
|
||||
json_t* j = json_object_get(json, "fields");
|
||||
|
||||
if (j && json_is_array(j) && json_array_size(j))
|
||||
{
|
||||
rval = json_object_get(json_array_get(j, 0), "name") != NULL;
|
||||
}
|
||||
|
||||
return rval;
|
||||
}
|
||||
|
||||
void Connection::processSchema(json_t* json)
|
||||
{
|
||||
m_keys.clear();
|
||||
m_types.clear();
|
||||
|
||||
json_t* arr = json_object_get(json, "fields");
|
||||
char* raw = json_dumps(json, 0);
|
||||
size_t i;
|
||||
json_t* v;
|
||||
|
||||
json_array_foreach(arr, i, v)
|
||||
{
|
||||
json_t* name = json_object_get(v, "name");
|
||||
json_t* type = json_object_get(v, "real_type");
|
||||
std::string nameval = name ? json_string_value(name) : "";
|
||||
std::string typeval = type ? json_string_value(type) : "undefined";
|
||||
m_keys.push_back(nameval);
|
||||
m_types.push_back(typeval);
|
||||
}
|
||||
}
|
||||
|
||||
Row Connection::processRow(json_t* js)
|
||||
{
|
||||
ValueList values;
|
||||
m_error.clear();
|
||||
|
||||
for (ValueList::iterator it = m_keys.begin();
|
||||
it != m_keys.end(); it++)
|
||||
{
|
||||
json_t* v = json_object_get(js, it->c_str());
|
||||
|
||||
if (v)
|
||||
{
|
||||
values.push_back(json_to_string(v));
|
||||
}
|
||||
else
|
||||
{
|
||||
m_error = "No value for key found: ";
|
||||
m_error += *it;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
Row rval;
|
||||
|
||||
if (m_error.empty())
|
||||
{
|
||||
rval = Row(new InternalRow(m_keys, m_types, values));
|
||||
}
|
||||
|
||||
return rval;
|
||||
}
|
||||
|
||||
Row Connection::read()
|
||||
{
|
||||
Row rval;
|
||||
std::string row;
|
||||
|
||||
if (readRow(row))
|
||||
{
|
||||
json_error_t err;
|
||||
json_t* js = json_loads(row.c_str(), JSON_ALLOW_NUL, &err);
|
||||
|
||||
if (js)
|
||||
{
|
||||
if (is_schema(js))
|
||||
{
|
||||
processSchema(js);
|
||||
rval = Connection::read();
|
||||
}
|
||||
else
|
||||
{
|
||||
rval = processRow(js);
|
||||
}
|
||||
|
||||
json_decref(js);
|
||||
}
|
||||
else
|
||||
{
|
||||
m_error = "Failed to parse JSON: ";
|
||||
m_error += err.text;
|
||||
}
|
||||
}
|
||||
|
||||
return rval;
|
||||
}
|
||||
|
||||
/**
|
||||
* Private functions
|
||||
*/
|
||||
|
||||
bool Connection::doAuth()
|
||||
{
|
||||
bool rval = false;
|
||||
std::string auth_str = generateAuthString(m_user, m_password);
|
||||
|
||||
/** Send the auth string */
|
||||
if (nointr_write(m_fd, auth_str.c_str(), auth_str.length()) == -1)
|
||||
{
|
||||
char err[ERRBUF_SIZE];
|
||||
m_error = "Failed to write authentication data: ";
|
||||
m_error += strerror_r(errno, err, sizeof (err));
|
||||
}
|
||||
else
|
||||
{
|
||||
/** Read the response */
|
||||
char buf[READBUF_SIZE];
|
||||
int bytes;
|
||||
|
||||
if ((bytes = nointr_read(m_fd, buf, sizeof (buf))) == -1)
|
||||
{
|
||||
char err[ERRBUF_SIZE];
|
||||
m_error = "Failed to read authentication response: ";
|
||||
m_error += strerror_r(errno, err, sizeof (err));
|
||||
}
|
||||
else if (memcmp(buf, OK_RESPONSE, sizeof (OK_RESPONSE) - 1) != 0)
|
||||
{
|
||||
buf[bytes] = '\0';
|
||||
m_error = "Authentication failed: ";
|
||||
m_error += buf;
|
||||
}
|
||||
else
|
||||
{
|
||||
rval = true;
|
||||
}
|
||||
}
|
||||
|
||||
return rval;
|
||||
}
|
||||
|
||||
bool Connection::doRegistration()
|
||||
{
|
||||
bool rval = false;
|
||||
std::string reg_msg(REGISTER_MSG);
|
||||
|
||||
const char *type = "";
|
||||
|
||||
if (m_flags & CDC_REQUEST_TYPE_JSON)
|
||||
{
|
||||
type = "JSON";
|
||||
}
|
||||
else if (m_flags & CDC_REQUEST_TYPE_AVRO)
|
||||
{
|
||||
type = "AVRO";
|
||||
}
|
||||
|
||||
reg_msg += type;
|
||||
|
||||
/** Send the registration message */
|
||||
if (nointr_write(m_fd, reg_msg.c_str(), reg_msg.length()) == -1)
|
||||
{
|
||||
char err[ERRBUF_SIZE];
|
||||
m_error = "Failed to write registration message: ";
|
||||
m_error += strerror_r(errno, err, sizeof (err));
|
||||
}
|
||||
else
|
||||
{
|
||||
/** Read the response */
|
||||
char buf[READBUF_SIZE];
|
||||
int bytes;
|
||||
|
||||
if ((bytes = nointr_read(m_fd, buf, sizeof (buf))) == -1)
|
||||
{
|
||||
char err[ERRBUF_SIZE];
|
||||
m_error = "Failed to read registration response: ";
|
||||
m_error += strerror_r(errno, err, sizeof (err));
|
||||
}
|
||||
else if (memcmp(buf, OK_RESPONSE, sizeof (OK_RESPONSE) - 1) != 0)
|
||||
{
|
||||
buf[bytes] = '\0';
|
||||
m_error = "Registration failed: ";
|
||||
m_error += buf;
|
||||
}
|
||||
else
|
||||
{
|
||||
rval = true;
|
||||
}
|
||||
}
|
||||
|
||||
return rval;
|
||||
}
|
||||
|
||||
bool Connection::readRow(std::string& dest)
|
||||
{
|
||||
bool rval = true;
|
||||
|
||||
while (true)
|
||||
{
|
||||
char buf;
|
||||
int rc = nointr_read(m_fd, &buf, 1);
|
||||
|
||||
if (rc == -1)
|
||||
{
|
||||
rval = false;
|
||||
char err[ERRBUF_SIZE];
|
||||
m_error = "Failed to read row: ";
|
||||
m_error += strerror_r(errno, err, sizeof (err));
|
||||
break;
|
||||
}
|
||||
|
||||
if (buf == '\n')
|
||||
{
|
||||
break;
|
||||
}
|
||||
else
|
||||
{
|
||||
dest += buf;
|
||||
|
||||
if (dest[0] == 'E' && dest[1] == 'R' & dest[2] == 'R')
|
||||
{
|
||||
m_error = "Server responded with an error: ";
|
||||
m_error += dest;
|
||||
rval = false;
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return rval;
|
||||
}
|
||||
|
||||
}
|
||||
122
maxscale-system-test/cdc_connector/cdc_connector.h
Normal file
122
maxscale-system-test/cdc_connector/cdc_connector.h
Normal file
@ -0,0 +1,122 @@
|
||||
#include <cstdint>
|
||||
#include <string>
|
||||
#include <tr1/memory>
|
||||
#include <vector>
|
||||
#include <algorithm>
|
||||
#include <jansson.h>
|
||||
|
||||
/** Request format flags */
|
||||
#define CDC_REQUEST_TYPE_JSON (1 << 0)
|
||||
#define CDC_REQUEST_TYPE_AVRO (1 << 1)
|
||||
|
||||
namespace CDC
|
||||
{
|
||||
|
||||
// The typedef for the Row type
|
||||
class InternalRow;
|
||||
typedef std::tr1::shared_ptr<InternalRow> Row;
|
||||
|
||||
typedef std::vector<std::string> ValueList;
|
||||
|
||||
// A class that represents a CDC connection
|
||||
class Connection
|
||||
{
|
||||
public:
|
||||
Connection(const std::string& address,
|
||||
uint16_t port,
|
||||
const std::string& user,
|
||||
const std::string& password,
|
||||
uint32_t flags = CDC_REQUEST_TYPE_JSON);
|
||||
virtual ~Connection();
|
||||
bool createConnection();
|
||||
bool requestData(const std::string& table, const std::string& gtid = "");
|
||||
Row read();
|
||||
void closeConnection();
|
||||
const std::string& getSchema() const
|
||||
{
|
||||
return m_schema;
|
||||
}
|
||||
const std::string& getError() const
|
||||
{
|
||||
return m_error;
|
||||
}
|
||||
|
||||
private:
|
||||
int m_fd;
|
||||
uint32_t m_flags;
|
||||
uint16_t m_port;
|
||||
std::string m_address;
|
||||
std::string m_user;
|
||||
std::string m_password;
|
||||
std::string m_error;
|
||||
std::string m_schema;
|
||||
ValueList m_keys;
|
||||
ValueList m_types;
|
||||
|
||||
bool doAuth();
|
||||
bool doRegistration();
|
||||
bool readRow(std::string& dest);
|
||||
void processSchema(json_t* json);
|
||||
Row processRow(json_t*);
|
||||
};
|
||||
|
||||
// Internal representation of a row, used via the Row type
|
||||
class InternalRow
|
||||
{
|
||||
public:
|
||||
|
||||
size_t fieldCount() const
|
||||
{
|
||||
return m_values.size();
|
||||
}
|
||||
|
||||
const std::string& value(size_t i) const
|
||||
{
|
||||
return m_values[i];
|
||||
}
|
||||
|
||||
const std::string& value(const std::string& str) const
|
||||
{
|
||||
ValueList::const_iterator it = std::find(m_keys.begin(), m_keys.end(), str);
|
||||
return m_values[it - m_keys.begin()];
|
||||
}
|
||||
|
||||
const std::string& key(size_t i) const
|
||||
{
|
||||
return m_keys[i];
|
||||
}
|
||||
|
||||
const std::string& type(size_t i) const
|
||||
{
|
||||
return m_types[i];
|
||||
}
|
||||
|
||||
~InternalRow()
|
||||
{
|
||||
}
|
||||
|
||||
private:
|
||||
ValueList m_keys;
|
||||
ValueList m_types;
|
||||
ValueList m_values;
|
||||
|
||||
// Not intended to be copied
|
||||
InternalRow(const InternalRow&);
|
||||
InternalRow& operator=(const InternalRow&);
|
||||
InternalRow();
|
||||
|
||||
// Only a Connection should construct an InternalRow
|
||||
friend class Connection;
|
||||
|
||||
InternalRow(const ValueList& keys,
|
||||
const ValueList& types,
|
||||
const ValueList& values):
|
||||
m_keys(keys),
|
||||
m_types(types),
|
||||
m_values(values)
|
||||
{
|
||||
}
|
||||
|
||||
};
|
||||
|
||||
}
|
||||
20
maxscale-system-test/cdc_connector/cmake/BuildJansson.cmake
Normal file
20
maxscale-system-test/cdc_connector/cmake/BuildJansson.cmake
Normal file
@ -0,0 +1,20 @@
|
||||
# If the Jansson library is not found, download it and build it from source.
|
||||
|
||||
set(JANSSON_REPO "https://github.com/akheron/jansson.git" CACHE STRING "Jansson Git repository")
|
||||
|
||||
# Release 2.9 of Jansson
|
||||
set(JANSSON_TAG "v2.9" CACHE STRING "Jansson Git tag")
|
||||
|
||||
ExternalProject_Add(jansson
|
||||
GIT_REPOSITORY ${JANSSON_REPO}
|
||||
GIT_TAG ${JANSSON_TAG}
|
||||
CMAKE_ARGS -DCMAKE_INSTALL_PREFIX=${CMAKE_BINARY_DIR}/jansson/install -DCMAKE_C_FLAGS=-fPIC -DJANSSON_BUILD_DOCS=OFF
|
||||
BINARY_DIR ${CMAKE_BINARY_DIR}/jansson
|
||||
INSTALL_DIR ${CMAKE_BINARY_DIR}/jansson/install
|
||||
UPDATE_COMMAND "")
|
||||
|
||||
set(JANSSON_FOUND TRUE CACHE INTERNAL "")
|
||||
set(JANSSON_STATIC_FOUND TRUE CACHE INTERNAL "")
|
||||
set(JANSSON_INCLUDE_DIR ${CMAKE_BINARY_DIR}/jansson/install/include CACHE INTERNAL "")
|
||||
set(JANSSON_STATIC_LIBRARIES ${CMAKE_BINARY_DIR}/jansson/install/lib/libjansson.a CACHE INTERNAL "")
|
||||
set(JANSSON_LIBRARIES ${JANSSON_STATIC_LIBRARIES} CACHE INTERNAL "")
|
||||
Reference in New Issue
Block a user