Added Missing old function && Fixed the error showing an error from kafka broker

.gitignore

@@ -1,2 +1,10 @@
-include/k.h
-build/
+qpbuild/
+clib/transformer*
+clib/kurl*
+clib/include/k.h
+clib/build/
+clib/qpbuild/
+clib/kafkakdb.so
+clib/libs/qtfm*
+q/kurl*
+tests/bench_producer.log

.travis.yml

@@ -56,17 +56,18 @@ before_install:
     fi
   - export FILE_NAME=${FILE_ROOT}-${TRAVIS_OS_NAME}-${TRAVIS_BRANCH}.${FILE_TAIL}
   # Build package
+  - cd clib
   - mkdir build
   - if [[ $BUILD == "True" && $TRAVIS_OS_NAME == "windows" ]]; then
       cd build;
-      cmake -G "Visual Studio 15 2017 Win64" --config Release .. -DENABLE_SSL:BOOL=OFF;
+      cmake -G "Visual Studio 15 2017 Win64" --config Release ../.. -DENABLE_SSL:BOOL=OFF;
       cmake --build . --config Release --target install;
-      cd ..;
+      cd ../..;
     elif [[ $BUILD == "True" && ( $TRAVIS_OS_NAME == "linux" || $TRAVIS_OS_NAME == "osx" ) ]]; then
       cd build;
-      cmake .. -DENABLE_SSL:BOOL=OFF;
+      cmake ../.. -DENABLE_SSL:BOOL=OFF;
       cmake --build . --target install;
-      cd ..;
+      cd ../..;
     fi
 
 script:
@@ -75,9 +76,9 @@ script:
       q test.q tests/ -q;
     fi
   - if [[ $TRAVIS_OS_NAME == "windows" ]]; then
-      7z a -tzip -r $FILE_NAME ./build/$FILE_ROOT/*;
+      7z a -tzip -r $FILE_NAME ./clib/build/$FILE_ROOT/*;
     else
-      tar -zcvf $FILE_NAME -C build/$FILE_ROOT .;
+      tar -zcvf $FILE_NAME -C clib/build/$FILE_ROOT .;
     fi
 
 deploy:

CMakeLists.txt

@@ -1,13 +1,13 @@
 ##%% General Settings %%##vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv#
 
 cmake_minimum_required(VERSION 3.1)
-project(kafkakdb C)
 
-# Set library name
-set(MY_LIBRARY_NAME kafkakdb)
+# Set project name
+# Declaration for installing files.
+project(kafkakdb C)
 
 # Add src directry
-add_subdirectory(src)
+add_subdirectory(clib)
 
 # Default option is Release
 if(NOT CMAKE_BUILD_TYPE)
@@ -24,9 +24,9 @@ else()
 endif()
 
 # Build package always
-file(COPY README.md LICENSE ${INSTALL_SCRIPT} DESTINATION ${PROJECT_BINARY_DIR}/${CMAKE_PROJECT_NAME})
-file(COPY examples/ DESTINATION ${PROJECT_BINARY_DIR}/${CMAKE_PROJECT_NAME}/examples/)
-file(COPY q/ DESTINATION ${PROJECT_BINARY_DIR}/${CMAKE_PROJECT_NAME}/q/)
+file(COPY README.md LICENSE ${INSTALL_SCRIPT} DESTINATION clib/${CMAKE_PROJECT_NAME})
+file(COPY examples/ DESTINATION clib/${CMAKE_PROJECT_NAME}/examples/)
+file(COPY q/ DESTINATION clib/${CMAKE_PROJECT_NAME}/q/)
 
 # Copy q files to QHOME
-install(DIRECTORY q/ DESTINATION $ENV{QHOME}/ CONFIGURATIONS Release)
+install(DIRECTORY q/ DESTINATION $ENV{QHOME}/ CONFIGURATIONS Release)

README.md

@@ -57,9 +57,9 @@ Then add the path to `LD_LIBRARY_PATH` (Linux) or `DYLD_LIBRRAY_PATH` (MacOSX) a
 ```bash
 
 # Linux
-$ export LD_LIBRARY_PATH=${KAFKA_LINSTALL_DIR}/lib:${LD_LIBRARY_PATH}
+$ export LD_LIBRARY_PATH=${KAFKA_INSTALL_DIR}/lib:${LD_LIBRARY_PATH}
 # MacOSX
-$ export DYLD_LIBRARY_PATH=${KAFKA_LINSTALL_DIR}/lib:${DYLD_LIBRARY_PATH}
+$ export DYLD_LIBRARY_PATH=${KAFKA_INSTALL_DIR}/lib:${DYLD_LIBRARY_PATH}
 
 ```
 
@@ -107,7 +107,6 @@ $ make install
 2. If using OpenSSL, remove `--disable-ssl` from configure command below
 3. On macOS with OpenSSL you might need to set `export OPENSSL_ROOT_DIR=/usr/local/Cellar/openssl` before proceeding
 
-
 #### Windows
 
 ##### Install rdkafka
@@ -209,6 +208,10 @@ w64> MKLINK rdkafka.dll %KAFKA_INSTALL_DIR%\bin\rdkafka.dll
 
 ```
 
+### Install qtfm (optional)
+
+Message transformer is supported only for Linux. If you intend to use transformer with kafkakdb, you need to install it to a location where q can serach it (for eaxmple, under `${QHOME}/[os]64`). After installing qtfm.so you need to copy `qtfm.h` to `clib/include` and `transformer.q` to `q/`.
+
 ### Install kafkakdb
 
 To use OpenSSL set `OPENSSL_ROOT_DIR` to the install directory of OpenSSL.
@@ -219,24 +222,28 @@ To use OpenSSL set `OPENSSL_ROOT_DIR` to the install directory of OpenSSL.
 
 ]$ git clone https://github.com/KxSystems/kafka.git
 ]$ cd kafka
-]$ mkdir build && cd build
-build]$ cmake .. -DENABLE_SSL:BOOL=OFF
+]$ mkdir -p clib/build && cd clib/build
+build]$ cmake ../.. -DENABLE_SSL:BOOL=OFF
 build]$ cmake --build . --target install
 
 ```
 
 **Notes:**
 1. `cmake --build . --target install` as used in the Linux/MacOS builds installs the required share object and q files to the `QHOME/[ml]64` and `QHOME` directories respectively. If you do not wish to install these files directly, you can execute `cmake --build .` instead of `cmake --build . --target install` and move the files from their build location at `build/kafkakdb`.
 2. If you use TLS remove the flag `-DENABLE_SSL:BOOL=OFF`.
+3. If you use a transformer library, use a flag `-DUSE_TRANSFORMER:BOOL=ON`, i.e., the build phase may become:
+
+        cmake ../../ -DUSE_TRANSFORMER:BOOL=ON
 
 #### Windows
 
 ```bat
 
 > git clone https://github.com/KxSystems/kafka.git
 > cd kafka
+> cd clib
 > mkdir build && cd build
-build> cmake --config Release .. -DENABLE_SSL:BOOL=OFF
+build> cmake --config Release ..\.. -DENABLE_SSL:BOOL=OFF
 build> cmake --build . --config Release --target install
 
 ```
@@ -246,6 +253,7 @@ build> cmake --build . --config Release --target install
 1. `cmake --build . --config Release --target install` installs the required share object and q files to the `QHOME\w64` and `QHOME` directories respectively. If you do not wish to install these files directly, you can execute `cmake --build . --config Release` instead of `cmake --build . --config Release --target install` and move the files from their build location at `build/kafkakdb`.
 2. You can use flag `cmake -G "Visual Studio 16 2019" -A Win32` if building 32-bit version.
 3. If you use TLS remove the flag `-DENABLE_SSL:BOOL=OFF`.
+4. Windows does not support transformer library.
 
 ## Quick Start
 

clib/CMakeLists.txt

@@ -0,0 +1,18 @@
+##%% General Settings %%##vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv#
+
+cmake_minimum_required(VERSION 3.1)
+
+# Set project name
+# Don't move this declaration. `PROJECT_SOURCE_DIR` must be this directorty.
+project(kafkakdb C)
+
+# Set library name
+set(MY_LIBRARY_NAME kafkakdb)
+
+# Add src directry
+add_subdirectory(src)
+
+# Default option is Release
+if(NOT CMAKE_BUILD_TYPE)
+	set(CMAKE_BUILD_TYPE Release)
+endif()

clib/include/kafkakdb_client.h

@@ -0,0 +1,89 @@
+#ifndef __KAFKAKDB_CLIENT_H__
+#define __KAFKAKDB_CLIENT_H__
+
+//+++++++++++++++++++++++++++++++++++++++++++++++++++++++//
+//                     Load Libraries                    //
+//+++++++++++++++++++++++++++++++++++++++++++++++++++++++//
+
+#include <rdkafka.h>
+#include "socketpair.h"
+#include "k.h"
+
+//+++++++++++++++++++++++++++++++++++++++++++++++++++++++//
+//                        Macros                         //
+//+++++++++++++++++++++++++++++++++++++++++++++++++++++++//
+
+#ifdef _WIN32
+#define EXP __declspec(dllexport)
+#else
+#define EXP
+#endif
+
+//++++++++++++++++++++++++++++++++++++++++++++++++++++++++//
+//                    Global Variables                    //
+//++++++++++++++++++++++++++++++++++++++++++++++++++++++++//
+
+//%% Number to String %%//vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv/
+
+/**
+ * @brief Buffer to write out an integer number.
+ */
+static char NUMBER[11];
+
+//%% Interface %%//vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv/
+
+#ifdef _WIN32
+extern SOCKET spair[2];
+#else
+#define SOCKET_ERROR -1
+extern I spair[2];
+#endif
+
+/**
+ * @brief Thread pool for polling client.
+ */
+static K ALL_THREADS;
+
+/**
+ * @brief Client handles expressed in symbol list
+ */
+extern K CLIENTS;
+
+//+++++++++++++++++++++++++++++++++++++++++++++++++++++++//
+//                   Private Functions                   //
+//+++++++++++++++++++++++++++++++++++++++++++++++++++++++//
+
+//%% Index Conversion %%//vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv/
+
+/**
+ * @brief Retrieve client handle from a given index.
+ * @param client_idx: Index of client.
+ * @return 
+ * - symbol: client handle if index is valid
+ * - null: error message if index is not valid
+ */
+rd_kafka_t *index_to_handle(K client_idx);
+
+/**
+ * @brief Retrieve index from a given client handle.
+ * @param handle: Client handle.
+ * @return 
+ * - int: Index of the given client in `CLIENTS`.
+ * - null int: if the client handle is not a registered one.
+ */
+I handle_to_index(const rd_kafka_t *handle);
+
+//+++++++++++++++++++++++++++++++++++++++++++++++++++++++//
+//                      Interface                        //
+//+++++++++++++++++++++++++++++++++++++++++++++++++++++++//
+
+//%% Create/Delete %%//vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv/
+
+/**
+ * @brief Destroy client handle and remove from `CLIENTS`.
+ * @param client_idx: Index of client in `CLIENTS`.
+ */
+EXP K delete_client(K client_idx);
+
+// __KAFKAKDB_CIENT_H__
+#endif

clib/include/kafkakdb_topic.h

@@ -0,0 +1,34 @@
+#ifndef __KAFKAKDB_TOPIC_H__
+#define __KAFKAKDB_TOPIC_H__
+
+//+++++++++++++++++++++++++++++++++++++++++++++++++++++++//
+//                     Load Libraries                    //
+//+++++++++++++++++++++++++++++++++++++++++++++++++++++++//
+
+#include <rdkafka.h>
+#include "k.h"
+
+//+++++++++++++++++++++++++++++++++++++++++++++++++++++++//
+//                    Global Variables                   //
+//+++++++++++++++++++++++++++++++++++++++++++++++++++++++//
+
+/**
+ * @brief Topic names expressed in symbol list
+ */
+extern K TOPICS;
+
+//+++++++++++++++++++++++++++++++++++++++++++++++++++++++//
+//                   Private Functions                   //
+//+++++++++++++++++++++++++++++++++++++++++++++++++++++++//
+
+/**
+ * @brief Retrieve topic object by topic index
+ * @param index: Index of topic
+ * @return 
+ * - symbol: Topic
+ * - error if index is out of range or topic for the index is null
+ */
+rd_kafka_topic_t *index_to_topic_handle(K topic_idx);
+
+// __KAFKAKDB_TOPIC_H__
+#endif