feat: ILP over HTTP (InfluxDB compatible)

CMakeLists.txt

@@ -100,6 +100,9 @@ if (QUESTDB_TESTS_AND_EXAMPLES)
     compile_example(
         line_sender_c_example_auth_tls
         examples/line_sender_c_example_auth_tls.c)
+    compile_example(
+        line_sender_c_example_http
+        examples/line_sender_c_example_http.c)
     compile_example(
         line_sender_cpp_example
         examples/line_sender_cpp_example.cpp)
@@ -112,6 +115,9 @@ if (QUESTDB_TESTS_AND_EXAMPLES)
     compile_example(
         line_sender_cpp_example_auth_tls
         examples/line_sender_cpp_example_auth_tls.cpp)
+    compile_example(
+        line_sender_cpp_example_http
+        examples/line_sender_cpp_example_http.cpp)
 
     # Include Rust tests as part of the tests run
     add_test(

ci/compile.yaml

@@ -0,0 +1,14 @@
+steps:
+  - script: |
+      rustup update $(toolchain)
+      rustup default $(toolchain)
+    condition: ne(variables['toolchain'], '')
+    displayName: "Update and set Rust toolchain"
+  - script: cmake -S . -B build -DCMAKE_BUILD_TYPE=Release -DQUESTDB_TESTS_AND_EXAMPLES=ON
+    env:
+      JAVA_HOME: $(JAVA_HOME_11_X64)
+    displayName: "Build Makefile with CMake"
+  - script: cmake --build build
+    env:
+      JAVA_HOME: $(JAVA_HOME_11_X64)
+    displayName: "Make"

ci/run_all_tests.py

@@ -38,8 +38,9 @@ def main():
     test_line_sender_path = next(iter(
         build_dir.glob(f'**/test_line_sender{exe_suffix}')))
     system_test_path = pathlib.Path('system_test') / 'test.py'
-    qdb_v = '7.3.2'  # The version of QuestDB we'll test against.
+    qdb_v = '7.3.7'  # The version of QuestDB we'll test against.
 
+    run_cmd('cargo', 'test', '--', '--nocapture', cwd='questdb-rs')
     run_cmd('cargo', 'test', '--all-features', '--', '--nocapture', cwd='questdb-rs')
     run_cmd(str(test_line_sender_path))
     run_cmd('python3', str(system_test_path), 'run', '--versions', qdb_v, '-v')

ci/run_tests_pipeline.yaml

@@ -19,38 +19,27 @@ stages:
             linux:
               imageName: "ubuntu-latest"
               poolName: "Azure Pipelines"
-              BUILD_WITH_MINGW: "0"
             linux-stable:
               imageName: "ubuntu-latest"
               poolName: "Azure Pipelines"
-              BUILD_WITH_MINGW: "0"
               toolchain: "stable"
             linux-beta:
               imageName: "ubuntu-latest"
               poolName: "Azure Pipelines"
-              BUILD_WITH_MINGW: "0"
               toolchain: "beta"
             linux-nightly:
               imageName: "ubuntu-latest"
               poolName: "Azure Pipelines"
-              BUILD_WITH_MINGW: "0"
               toolchain: "nightly"
             mac:
               imageName: "macos-latest"
               poolName: "Azure Pipelines"
-              BUILD_WITH_MINGW: "0"
             windows-msvc-2022:
               imageName: "windows-2022"
               poolName: "Azure Pipelines"
-              BUILD_WITH_MINGW: "0"
             windows-msvc-2019:
               imageName: "windows-2019"
               poolName: "Azure Pipelines"
-              BUILD_WITH_MINGW: "0"
-            # windows-mingw:
-            #   imageName: "windows-2022"
-            #   poolName: "Azure Pipelines"
-            #   BUILD_WITH_MINGW: "1"
         pool:
           name: $(poolName)
           vmImage: $(imageName)
@@ -60,34 +49,11 @@ stages:
             fetchDepth: 1
             lfs: false
             submodules: false
-          - script: |
-              rustup update $(toolchain)
-              rustup default $(toolchain)
-            condition: ne(variables['toolchain'], '')
-            displayName: "Update and set Rust toolchain"
+          - template: compile.yaml
           - script: |
               cd questdb-rs
-              cargo build --example basic --features=chrono_timestamp
-              cargo build --example auth --features=chrono_timestamp
-              cargo build --example auth_tls --features=chrono_timestamp
-            displayName: Build Rust examples.
-          - script: cmake -S . -B build -DCMAKE_BUILD_TYPE=Release -DQUESTDB_TESTS_AND_EXAMPLES=ON
-            env:
-              JAVA_HOME: $(JAVA_HOME_11_X64)
-            displayName: "Build Makefile with CMake"
-            condition: eq(variables['BUILD_WITH_MINGW'], '0')
-          # - script: |
-          #     choco upgrade mingw -y
-          #     rustup target add x86_64-pc-windows-gnu
-          #     cmake -S . -B build -DCMAKE_BUILD_TYPE=Release -G "MinGW Makefiles"
-          #   env:
-          #     JAVA_HOME: $(JAVA_HOME_11_X64)
-          #   displayName: "Build Makefile with CMake"
-          #   condition: eq(variables['BUILD_WITH_MINGW'], '1')
-          - script: cmake --build build
-            env:
-              JAVA_HOME: $(JAVA_HOME_11_X64)
-            displayName: "Make"
+              cargo build --examples --all-features
+            displayName: "Build Rust examples"
           - script: python3 ci/run_all_tests.py
             env:
               JAVA_HOME: $(JAVA_HOME_11_X64)
@@ -96,7 +62,6 @@ stages:
             inputs:
               pathToPublish: ./build
             displayName: "Publish build directory"
-            condition: eq(variables['BUILD_WITH_MINGW'], '1')
       - job: CargoFmtAndClippy
         displayName: "cargo fmt and clippy"
         pool:
@@ -134,3 +99,26 @@ stages:
               cd tls_proxy
               cargo clippy --all-targets --all-features -- -D warnings
             displayName: "tls_proxy: clippy"
+      - job: TestVsQuestDBMaster
+        displayName: "Vs QuestDB 'master'"
+        pool:
+          vmImage: 'ubuntu-latest'
+        timeoutInMinutes: 60
+        steps:
+          - checkout: self
+            fetchDepth: 1
+            lfs: false
+            submodules: false
+          - template: compile.yaml
+          - script: |
+              git clone --depth 1 https://github.com/questdb/questdb.git
+            displayName: git clone questdb
+          - task: Maven@3
+            displayName: "Compile QuestDB"
+            inputs:
+              mavenPOMFile: 'questdb/pom.xml'
+              jdkVersionOption: '1.11'
+              options: "install -DskipTests -Pbuild-web-console"
+          - script: |
+              python3 system_test/test.py run --repo ./questdb -v
+            displayName: "integration test"

cpp_test/test_line_sender.cpp

@@ -715,4 +715,19 @@ TEST_CASE("Empty Buffer") {
         sender.flush_and_keep(b1),
         "State error: Bad call to `flush`, should have called `table` instead.",
         questdb::ingress::line_sender_error);
+}
+
+TEST_CASE("HTTP basics") {
+    questdb::ingress::opts opts1{"localhost", 1};
+    questdb::ingress::opts opts2{"localhost", 1};
+    opts1.http().transactional().max_retries(5).retry_interval(10).basic_auth("user", "pass");
+    opts2.http().token_auth("token").min_throughput(1000);
+    questdb::ingress::line_sender sender1{opts1};
+    questdb::ingress::line_sender sender2{opts2};
+
+    questdb::ingress::line_sender_buffer b1;
+    b1.table("test").symbol("a", "b").at_now();
+
+    CHECK_THROWS_AS(sender1.flush(b1), questdb::ingress::line_sender_error);
+    CHECK_THROWS_AS(sender2.flush(b1), questdb::ingress::line_sender_error);
 }

examples/line_sender_c_example_http.c

@@ -0,0 +1,135 @@
+#include <questdb/ingress/line_sender.h>
+#include <stdio.h>
+#include <stdbool.h>
+#include <string.h>
+
+static bool example(const char* host, const char* port)
+{
+    line_sender_error* err = NULL;
+    line_sender_opts* opts = NULL;
+    line_sender* sender = NULL;
+    line_sender_buffer* buffer = NULL;
+
+    line_sender_utf8 host_utf8 = { 0, NULL };
+    if (!line_sender_utf8_init(&host_utf8, strlen(host), host, &err))
+        goto on_error;
+
+    line_sender_utf8 port_utf8 = { 0, NULL };
+    if (!line_sender_utf8_init(&port_utf8, strlen(port), port, &err))
+        goto on_error;
+
+    // Call `line_sender_opts_new` if instead you have an integer port.
+    opts = line_sender_opts_new_service(host_utf8, port_utf8);
+
+    // Use ILP/HTTP instead of ILP/TCP for better error messages.
+    line_sender_opts_http(opts);
+
+    // Ensure that each buffer contains lines for a single table on each flush.
+    line_sender_opts_transactional(opts);
+
+    sender = line_sender_connect(opts, &err);
+    line_sender_opts_free(opts);
+    opts = NULL;
+    if (!sender)
+        goto on_error;
+
+    buffer = line_sender_buffer_new();
+    line_sender_buffer_reserve(buffer, 64 * 1024);  // 64KB buffer initial size.
+
+    // We prepare all our table names and column names in advance.
+    // If we're inserting multiple rows, this allows us to avoid
+    // re-validating the same strings over and over again.
+    line_sender_table_name table_name = QDB_TABLE_NAME_LITERAL("c_cars_http");
+    line_sender_column_name id_name = QDB_COLUMN_NAME_LITERAL("id");
+    line_sender_column_name x_name = QDB_COLUMN_NAME_LITERAL("x");
+    line_sender_column_name y_name = QDB_COLUMN_NAME_LITERAL("y");
+    line_sender_column_name booked_name = QDB_COLUMN_NAME_LITERAL("booked");
+    line_sender_column_name passengers_name = QDB_COLUMN_NAME_LITERAL(
+        "passengers");
+    line_sender_column_name driver_name = QDB_COLUMN_NAME_LITERAL("driver");
+
+    if (!line_sender_buffer_table(buffer, table_name, &err))
+        goto on_error;
+
+    line_sender_utf8 id_value = QDB_UTF8_LITERAL(
+        "d6e5fe92-d19f-482a-a97a-c105f547f721");
+    if (!line_sender_buffer_symbol(buffer, id_name, id_value, &err))
+        goto on_error;
+
+    if (!line_sender_buffer_column_f64(buffer, x_name, 30.5, &err))
+        goto on_error;
+
+    if (!line_sender_buffer_column_f64(buffer, y_name, -150.25, &err))
+        goto on_error;
+
+    if (!line_sender_buffer_column_bool(buffer, booked_name, true, &err))
+        goto on_error;
+
+    if (!line_sender_buffer_column_i64(buffer, passengers_name, 3, &err))
+        goto on_error;
+
+    line_sender_utf8 driver_value = QDB_UTF8_LITERAL("John Doe");
+    if (!line_sender_buffer_column_str(buffer, driver_name, driver_value, &err))
+        goto on_error;
+
+    // 1997-07-04 04:56:55 UTC
+    int64_t designated_timestamp = 867992215000000000;
+    if (!line_sender_buffer_at_nanos(buffer, designated_timestamp, &err))
+        goto on_error;
+
+    //// If you want to get the current system timestamp as nanos, call:
+    // if (!line_sender_buffer_at_nanos(buffer, line_sender_now_nanos(), &err))
+    //     goto on_error;
+
+    // To insert more records, call `line_sender_buffer_table(..)...` again.
+    // It's recommended to keep a timer and/or maximum buffer size to flush
+    // the buffer periodically with any accumulated records.
+    if (!line_sender_flush(sender, buffer, &err))
+        goto on_error;
+
+    line_sender_close(sender);
+
+    return true;
+
+on_error: ;
+    line_sender_opts_free(opts);
+    size_t err_len = 0;
+    const char* err_msg = line_sender_error_msg(err, &err_len);
+    fprintf(stderr, "Error running example: %.*s\n", (int)err_len, err_msg);
+    line_sender_error_free(err);
+    line_sender_buffer_free(buffer);
+    line_sender_close(sender);
+    return false;
+}
+
+static bool displayed_help(int argc, const char* argv[])
+{
+    for (int index = 1; index < argc; ++index)
+    {
+        const char* arg = argv[index];
+        if ((strncmp(arg, "-h", 2) == 0) || (strncmp(arg, "--help", 6) == 0))
+        {
+            fprintf(stderr, "Usage:\n");
+            fprintf(stderr, "line_sender_c_example: [HOST [PORT]]\n");
+            fprintf(stderr, "    HOST: ILP host (defaults to \"localhost\").\n");
+            fprintf(stderr, "    PORT: ILP port (defaults to \"9009\").\n");
+            return true;
+        }
+    }
+    return false;
+}
+
+int main(int argc, const char* argv[])
+{
+    if (displayed_help(argc, argv))
+        return 0;
+
+    const char* host = "localhost";
+    if (argc >= 2)
+        host = argv[1];
+    const char* port = "9009";
+    if (argc >= 3)
+        port = argv[2];
+
+    return !example(host, port);
+}