Use correct CDC connector for cdc_datatypes

The cdc_datatypes test did not use the correct connector and instead it
used a stale version of the MaxScale CDC Connector. The connector should
be treated as an external dependency and thus cloned at configuration
time.

maxscale-system-test/CMakeLists.txt

@@ -50,7 +50,7 @@ add_library(testcore SHARED testconnections.cpp nodes.cpp mariadb_nodes.cpp maxs
   mariadb_func.cpp get_com_select_insert.cpp maxadmin_operations.cpp big_transaction.cpp
   sql_t1.cpp test_binlog_fnc.cpp get_my_ip.cpp big_load.cpp  get_com_select_insert.cpp
   different_size.cpp fw_copy_rules maxinfo_func.cpp config_operations.cpp rds_vpc.cpp execute_cmd.cpp
-  blob_test.cpp cdc_connector.cpp)
+  blob_test.cpp)
 target_link_libraries(testcore ${MYSQL_CLIENT} ${CDC_CONNECTOR_LIBRARIES} ${JANSSON_LIBRARIES} z nsl m pthread ssl dl rt crypto crypt)
 install(TARGETS testcore DESTINATION system-test)
 add_dependencies(testcore connector-c cdc_connector)

maxscale-system-test/cdc_connector/CMakeLists.txt

@@ -1,21 +0,0 @@
-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)

maxscale-system-test/cdc_connector/cdc_connector.cpp

@@ -1,454 +0,0 @@
-#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;
-}
-
-}

maxscale-system-test/cdc_connector/cdc_connector.h

@@ -1,122 +0,0 @@
-#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)
-    {
-    }
-
-};
-
-}

maxscale-system-test/cdc_connector/cmake/BuildJansson.cmake

@@ -1,20 +0,0 @@
-# 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 "")

maxscale-system-test/cdc_datatypes/cdc_datatypes.cpp

@@ -178,7 +178,7 @@ bool run_test(TestConnections& test)
             std::string name = type_to_table_name(test_set[x].types[i]);
             CDC::Connection conn(test.maxscales->IP[0], 4001, "skysql", "skysql");
 
-            if (conn.createConnection() && conn.requestData(name))
+            if (conn.connect(name))
             {
                 for (int j = 0; test_set[x].values[j]; j++)
                 {
@@ -198,14 +198,14 @@ bool run_test(TestConnections& test)
                     }
                     else
                     {
-                        std::string err = conn.getError();
+                        std::string err = conn.error();
                         test.tprintf("Failed to read data: %s", err.c_str());
                     }
                 }
             }
             else
             {
-                std::string err = conn.getError();
+                std::string err = conn.error();
                 test.tprintf("Failed to request data: %s", err.c_str());
                 rval = false;
                 break;

maxscale-system-test/utilities.cmake

@@ -105,15 +105,16 @@ set(MYSQL_CLIENT ${CMAKE_BINARY_DIR}/lib/mariadb/libmariadbclient.a CACHE INTERN
 
 # Build the CDC connector
 ExternalProject_Add(cdc_connector
-  SOURCE_DIR ${CMAKE_SOURCE_DIR}/cdc_connector/
+  GIT_REPOSITORY https://github.com/mariadb-corporation/maxscale-cdc-connector.git
   CMAKE_ARGS -DCMAKE_INSTALL_PREFIX=${CMAKE_BINARY_DIR}/cdc_connector/
   BUILD_COMMAND make
-  INSTALL_COMMAND make install)
+  INSTALL_COMMAND make install
-
+  UPDATE_COMMAND "")
-set(CDC_CONNECTOR_INCLUDE ${CMAKE_BINARY_DIR}/cdc_connector/include/ CACHE INTERNAL "")
-set(CDC_CONNECTOR_LIBRARIES ${CMAKE_BINARY_DIR}/cdc_connector/lib/libcdc_connector.so CACHE INTERNAL "")
-include_directories(${CMAKE_BINARY_DIR}/cdc_connector/include)
 
+include(GNUInstallDirs)
+set(CDC_CONNECTOR_INCLUDE ${CMAKE_BINARY_DIR}/cdc_connector/${CMAKE_INSTALL_INCLUDEDIR}/ CACHE INTERNAL "")
+set(CDC_CONNECTOR_LIBRARIES ${CMAKE_BINARY_DIR}/cdc_connector/${CMAKE_INSTALL_LIBDIR}/libcdc_connector.so CACHE INTERNAL "")
+include_directories(${CMAKE_BINARY_DIR}/cdc_connector/${CMAKE_INSTALL_INCLUDEDIR})
 
 #
 # Check that all required components are present. To build even without them,
@@ -129,3 +130,23 @@ find_program(HAVE_PHP php)
 if (NOT HAVE_PHP)
   message(FATAL_ERROR "Could not find php.")
 endif()
+
+# Build the Jansson library 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 "")