From 7a439a7158f3eae3a2522e88d726896c82ea3b6a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Komor?= Date: Wed, 16 Sep 2026 07:53:56 +0200 Subject: [PATCH 01/10] Add Kafka Gateway transport plugin Implements request/reply over Apache Kafka (librdkafka) with correlation-id, matching rabbitmq/servicebus plugin contracts. Includes CMake FetchContent/vcpkg build, Docker Compose for local broker, and smoke test. --- README.md | 1 + kafka/.gitignore | 9 + kafka/BUILD_NOTES.md | 33 ++ kafka/CMakeLists.txt | 22 ++ kafka/GraftcodePluginsInterfaces/IServer.h | 16 + kafka/GraftcodePluginsInterfaces/ITransport.h | 12 + kafka/GraftcodePluginsInterfaces/_common.h | 9 + kafka/KafkaPlugin/CMakeLists.txt | 99 ++++++ kafka/KafkaPlugin/KafkaClient.cpp | 217 ++++++++++++ kafka/KafkaPlugin/KafkaClient.h | 52 +++ kafka/KafkaPlugin/KafkaServer.cpp | 308 ++++++++++++++++++ kafka/KafkaPlugin/KafkaServer.h | 50 +++ kafka/KafkaPlugin/KafkaUtil.h | 74 +++++ kafka/KafkaPlugin/TransportKafka.cpp | 97 ++++++ kafka/KafkaPlugin/TransportKafka.h | 17 + kafka/KafkaPluginTest/CMakeLists.txt | 6 + kafka/KafkaPluginTest/smoke.cpp | 62 ++++ kafka/Readme.md | 210 ++++++++++++ kafka/docker-compose.redpanda.yml | 10 + kafka/docker-compose.yml | 19 ++ kafka/pluginConfig.json | 8 + kafka/scripts/create-topics.sh | 53 +++ kafka/vcpkg.json | 9 + 23 files changed, 1393 insertions(+) create mode 100644 kafka/.gitignore create mode 100644 kafka/BUILD_NOTES.md create mode 100644 kafka/CMakeLists.txt create mode 100644 kafka/GraftcodePluginsInterfaces/IServer.h create mode 100644 kafka/GraftcodePluginsInterfaces/ITransport.h create mode 100644 kafka/GraftcodePluginsInterfaces/_common.h create mode 100644 kafka/KafkaPlugin/CMakeLists.txt create mode 100644 kafka/KafkaPlugin/KafkaClient.cpp create mode 100644 kafka/KafkaPlugin/KafkaClient.h create mode 100644 kafka/KafkaPlugin/KafkaServer.cpp create mode 100644 kafka/KafkaPlugin/KafkaServer.h create mode 100644 kafka/KafkaPlugin/KafkaUtil.h create mode 100644 kafka/KafkaPlugin/TransportKafka.cpp create mode 100644 kafka/KafkaPlugin/TransportKafka.h create mode 100644 kafka/KafkaPluginTest/CMakeLists.txt create mode 100644 kafka/KafkaPluginTest/smoke.cpp create mode 100644 kafka/Readme.md create mode 100644 kafka/docker-compose.redpanda.yml create mode 100644 kafka/docker-compose.yml create mode 100644 kafka/pluginConfig.json create mode 100644 kafka/scripts/create-topics.sh create mode 100644 kafka/vcpkg.json diff --git a/README.md b/README.md index 80605e2..b903cea 100644 --- a/README.md +++ b/README.md @@ -8,6 +8,7 @@ Official open-source Graftcode Gateway plugins for carrying Graft calls over ext |--------|---------| | [rabbitmq](rabbitmq/) | RabbitMQ (AMQP 0-9-1), request/reply | | [servicebus](servicebus/) | Azure Service Bus (AMQP 1.0), request/reply and one-way | +| [kafka](kafka/) | Apache Kafka, request/reply (correlation-id) | | [observability/opentelemetry](observability/opentelemetry/) | OpenTelemetry / Azure Application Insights connector | Each plugin has its own README with build and configuration steps. For how the Gateway loads a plugin, see the "Plugin server config" section of the [Graftcode Gateway](https://github.com/grft-dev/graftcode-gateway) README. diff --git a/kafka/.gitignore b/kafka/.gitignore new file mode 100644 index 0000000..5c9d681 --- /dev/null +++ b/kafka/.gitignore @@ -0,0 +1,9 @@ +build/ +out/ +.vs/ +CMakeUserPresets.json +cmake-build-*/ +vcpkg/ +vcpkg_installed/ +.cache/ +*.user diff --git a/kafka/BUILD_NOTES.md b/kafka/BUILD_NOTES.md new file mode 100644 index 0000000..5585967 --- /dev/null +++ b/kafka/BUILD_NOTES.md @@ -0,0 +1,33 @@ +# Build notes + +## Verified in this workspace (2026-09-16 Europe/Warsaw) + +| Step | Result | +|------|--------| +| `cmake -S . -B build -DCMAKE_BUILD_TYPE=Release` | **Succeeded** (FetchContent nlohmann/json + librdkafka `v2.8.0`) | +| `cmake --build build -j$(nproc)` | **Succeeded** → `build/KafkaPlugin/libKafkaPlugin.so` | +| `ctest --test-dir build` | **Passed** (`KafkaPluginSmoke`) | +| Exported symbols | `CreateServer`, `DestroyServer`, `CreateTransportChannel`, `DestroyTransportChannel` | + +Toolchain used: GCC 14, CMake 3.31, OpenSSL 3.5, libsasl2, zlib, zstd, libcurl. + +## Include path note + +FetchContent builds expose ``; packaged/vcpkg installs typically use +``. Sources use `__has_include` to accept either. + +## Possible blockers elsewhere + +1. **Network / git** required on first configure for FetchContent. +2. **librdkafka compile time** is several minutes; if the build is OOM-killed, use `cmake --build build -j2`. +3. Missing SSL/SASL packages: install `libssl-dev`, `libsasl2-dev`, `zlib1g-dev`, `libzstd-dev` (and optionally `libcurl4-openssl-dev`). +4. **Windows**: prefer vcpkg + `-DKAFKA_USE_SYSTEM_RDKAFKA=ON` with the vcpkg toolchain. +5. Smoke test must **not** call `IServer::start()` or `SendCommand` without a broker (reconnect / RPC timeout). + +## Suggested PR checklist + +- [x] Source complete (client + server RPC with correlation-id / reply-to) +- [x] CMake FetchContent + vcpkg.json +- [x] Linux configure/build/smoke in this environment +- [ ] Manual GG round-trip with `docker compose up -d` + `./scripts/create-topics.sh` +- [ ] Copy folder contents to `graftcode-extensions/kafka/` (flat drop-in; do not nest an extra sketch directory) diff --git a/kafka/CMakeLists.txt b/kafka/CMakeLists.txt new file mode 100644 index 0000000..a7f50a4 --- /dev/null +++ b/kafka/CMakeLists.txt @@ -0,0 +1,22 @@ +set(CMAKE_MIN 3.22) +cmake_minimum_required(VERSION ${CMAKE_MIN}) +set(CMAKE_POLICY_VERSION_MINIMUM ${CMAKE_MIN}) +cmake_policy(VERSION ${CMAKE_MIN}) + +project("KafkaPlugin" VERSION 1.0.0) + +set(CMAKE_CXX_STANDARD 20) +set(CMAKE_CXX_STANDARD_REQUIRED ON) +set(CMAKE_POSITION_INDEPENDENT_CODE ON) + +add_definitions(-DUNICODE) +enable_testing() + +set(CMAKE_POLICY_DEFAULT_CMP0135 NEW) + +if(MSVC) + add_compile_options(/utf-8) +endif() + +add_subdirectory("KafkaPlugin") +add_subdirectory("KafkaPluginTest") diff --git a/kafka/GraftcodePluginsInterfaces/IServer.h b/kafka/GraftcodePluginsInterfaces/IServer.h new file mode 100644 index 0000000..4574ab8 --- /dev/null +++ b/kafka/GraftcodePluginsInterfaces/IServer.h @@ -0,0 +1,16 @@ +#pragma once +#include + +namespace GraftcodeGateway { +class IServer { +public: + using byte = unsigned char; + using WriteResponseFn = void(*)(void* context, const byte* data, std::size_t size); + using ProcessMessageFn = bool(*)(const byte* requestData, std::size_t requestSize, WriteResponseFn writeResponse, void* writeContext); + + virtual ~IServer() = default; + virtual void configure(const char* jsonConfig, ProcessMessageFn processMessage) = 0; + virtual void start() = 0; + virtual void stop() = 0; +}; +} diff --git a/kafka/GraftcodePluginsInterfaces/ITransport.h b/kafka/GraftcodePluginsInterfaces/ITransport.h new file mode 100644 index 0000000..3ab68b9 --- /dev/null +++ b/kafka/GraftcodePluginsInterfaces/ITransport.h @@ -0,0 +1,12 @@ +#pragma once +#include "_common.h" + +namespace Hypertube::Native::Interfaces { +class ITransport { +public: + virtual ~ITransport() = default; + virtual int Initialize(byte callingRuntimeNumber, byte calledRuntimeNumber, byte calledRuntimeVersion) = 0; + virtual int SendCommand(byte* messageByteArray, int32_t messageByteArrayLen) = 0; + virtual int ReadResponse(byte* responseByteArray, int32_t responseByteArrayLen) = 0; +}; +} diff --git a/kafka/GraftcodePluginsInterfaces/_common.h b/kafka/GraftcodePluginsInterfaces/_common.h new file mode 100644 index 0000000..f878601 --- /dev/null +++ b/kafka/GraftcodePluginsInterfaces/_common.h @@ -0,0 +1,9 @@ +#pragma once +#ifndef byte +typedef unsigned char byte; +#endif +#ifndef int32_t +typedef int int32_t; +#endif +#include +#include diff --git a/kafka/KafkaPlugin/CMakeLists.txt b/kafka/KafkaPlugin/CMakeLists.txt new file mode 100644 index 0000000..a840585 --- /dev/null +++ b/kafka/KafkaPlugin/CMakeLists.txt @@ -0,0 +1,99 @@ +# Graftcode Kafka plugin (request/reply via librdkafka C++ API). +set(target_name KafkaPlugin) + +add_library(${target_name} SHARED + TransportKafka.cpp + KafkaClient.cpp + KafkaServer.cpp +) + +target_include_directories(${target_name} PUBLIC + "${CMAKE_SOURCE_DIR}/GraftcodePluginsInterfaces" +) + +include(FetchContent) + +# --------------------------------------------------------------------------- +# nlohmann/json +# --------------------------------------------------------------------------- +find_package(nlohmann_json CONFIG QUIET) +if(NOT nlohmann_json_FOUND) + FetchContent_Declare( + json + GIT_REPOSITORY https://github.com/nlohmann/json.git + GIT_TAG v3.12.0 + GIT_SHALLOW TRUE + ) + FetchContent_MakeAvailable(json) +endif() + +# --------------------------------------------------------------------------- +# librdkafka (rdkafka++) +# +# Preferred resolution order: +# 1) find_package(RdKafka) — system install or vcpkg toolchain +# 2) FetchContent from confluentinc/librdkafka (edenhill redirects here) +# +# Override the FetchContent tag with -DKAFKA_LIBRDKAFKA_GIT_TAG=vX.Y.Z +# Skip FetchContent with -DKAFKA_USE_SYSTEM_RDKAFKA=ON (requires find_package). +# --------------------------------------------------------------------------- +set(KAFKA_LIBRDKAFKA_GIT_TAG "v2.8.0" CACHE STRING "librdkafka git tag for FetchContent") +option(KAFKA_USE_SYSTEM_RDKAFKA "Require system/vcpkg RdKafka; do not FetchContent" OFF) + +find_package(RdKafka CONFIG QUIET) +set(_kafka_rdkafka_target "") + +if(RdKafka_FOUND) + if(TARGET RdKafka::rdkafka++) + set(_kafka_rdkafka_target RdKafka::rdkafka++) + elseif(TARGET rdkafka++) + set(_kafka_rdkafka_target rdkafka++) + endif() +endif() + +if(NOT _kafka_rdkafka_target) + if(KAFKA_USE_SYSTEM_RDKAFKA) + message(FATAL_ERROR + "KAFKA_USE_SYSTEM_RDKAFKA=ON but find_package(RdKafka) failed. " + "Install librdkafka or configure with the vcpkg toolchain.") + endif() + + message(STATUS "RdKafka not found via find_package; FetchContent ${KAFKA_LIBRDKAFKA_GIT_TAG}") + + # Prefer a static librdkafka so the shared plugin is easier to drop in. + set(RDKAFKA_BUILD_STATIC ON CACHE BOOL "Build librdkafka as static library" FORCE) + set(RDKAFKA_BUILD_EXAMPLES OFF CACHE BOOL "Skip librdkafka examples" FORCE) + set(RDKAFKA_BUILD_TESTS OFF CACHE BOOL "Skip librdkafka tests" FORCE) + set(WITH_SSL ON CACHE BOOL "Enable SSL in librdkafka" FORCE) + + FetchContent_Declare( + librdkafka + GIT_REPOSITORY https://github.com/confluentinc/librdkafka.git + GIT_TAG ${KAFKA_LIBRDKAFKA_GIT_TAG} + GIT_SHALLOW TRUE + ) + FetchContent_MakeAvailable(librdkafka) + + if(TARGET rdkafka++) + set(_kafka_rdkafka_target rdkafka++) + elseif(TARGET RdKafka::rdkafka++) + set(_kafka_rdkafka_target RdKafka::rdkafka++) + else() + message(FATAL_ERROR "librdkafka FetchContent succeeded but rdkafka++ target is missing") + endif() +endif() + +target_link_libraries(${target_name} PUBLIC + nlohmann_json::nlohmann_json + ${_kafka_rdkafka_target} +) + +# Ensure consumers see librdkafka headers when linking a plain target name. +if(TARGET rdkafka) + target_link_libraries(${target_name} PUBLIC rdkafka) +endif() + +if(UNIX AND NOT APPLE) + find_package(Threads REQUIRED) + target_link_libraries(${target_name} PUBLIC Threads::Threads) +endif() diff --git a/kafka/KafkaPlugin/KafkaClient.cpp b/kafka/KafkaPlugin/KafkaClient.cpp new file mode 100644 index 0000000..fd7718b --- /dev/null +++ b/kafka/KafkaPlugin/KafkaClient.cpp @@ -0,0 +1,217 @@ +#include "KafkaClient.h" +#include "KafkaUtil.h" + +#include +#include +#include +#include + +#if __has_include() +#include +#elif __has_include() +#include +#else +#error "librdkafka C++ header not found (expected librdkafka/rdkafkacpp.h or rdkafkacpp.h)" +#endif + +namespace { + +void drainProducer(RdKafka::Producer* producer, int timeoutMs) { + if (!producer) { + return; + } + producer->flush(timeoutMs); + while (producer->outq_len() > 0) { + producer->poll(50); + } +} + +} // namespace + +KafkaClient::KafkaClient(Config cfg) : cfg_(std::move(cfg)) { + instanceId_ = KafkaPluginUtil::makeUuidV4(); + // Unique group so concurrent clients sharing one reply topic each get a full + // copy of replies, then filter by correlation-id (broadcast-per-group model). + consumerGroupId_ = cfg_.groupId.empty() + ? ("graft-client-" + instanceId_) + : (cfg_.groupId + "-" + instanceId_); +} + +KafkaClient::~KafkaClient() { + std::lock_guard lock(callMutex_); + if (consumer_) { + consumer_->close(); + consumer_.reset(); + } + if (producer_) { + drainProducer(producer_.get(), 5000); + producer_.reset(); + } +} + +void KafkaClient::applySecurity(RdKafka::Conf* conf) const { + using KafkaPluginUtil::setConfOrThrow; + if (!cfg_.securityProtocol.empty()) { + setConfOrThrow(conf, "security.protocol", cfg_.securityProtocol); + } + if (!cfg_.saslMechanism.empty()) { + setConfOrThrow(conf, "sasl.mechanism", cfg_.saslMechanism); + } + if (!cfg_.saslUsername.empty()) { + setConfOrThrow(conf, "sasl.username", cfg_.saslUsername); + } + if (!cfg_.saslPassword.empty()) { + setConfOrThrow(conf, "sasl.password", cfg_.saslPassword); + } + if (!cfg_.sslCaLocation.empty()) { + setConfOrThrow(conf, "ssl.ca.location", cfg_.sslCaLocation); + } +} + +void KafkaClient::ensureStarted() { + if (started_) { + return; + } + + using KafkaPluginUtil::setConfOrThrow; + std::string errstr; + + { + std::unique_ptr pconf(RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL)); + setConfOrThrow(pconf.get(), "bootstrap.servers", cfg_.brokers); + setConfOrThrow(pconf.get(), "client.id", "graft-kafka-client-producer-" + instanceId_); + setConfOrThrow(pconf.get(), "message.timeout.ms", std::to_string(cfg_.rpcTimeoutMs)); + applySecurity(pconf.get()); + + RdKafka::Producer* raw = RdKafka::Producer::create(pconf.release(), errstr); + if (!raw) { + throw std::runtime_error("Failed to create Kafka producer: " + errstr); + } + producer_.reset(raw); + } + + { + std::unique_ptr cconf(RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL)); + setConfOrThrow(cconf.get(), "bootstrap.servers", cfg_.brokers); + setConfOrThrow(cconf.get(), "group.id", consumerGroupId_); + setConfOrThrow(cconf.get(), "client.id", "graft-kafka-client-consumer-" + instanceId_); + setConfOrThrow(cconf.get(), "enable.auto.commit", "true"); + setConfOrThrow(cconf.get(), "auto.offset.reset", "latest"); + setConfOrThrow(cconf.get(), "allow.auto.create.topics", "true"); + applySecurity(cconf.get()); + + RdKafka::KafkaConsumer* raw = RdKafka::KafkaConsumer::create(cconf.release(), errstr); + if (!raw) { + throw std::runtime_error("Failed to create Kafka consumer: " + errstr); + } + consumer_.reset(raw); + + const RdKafka::ErrorCode subErr = consumer_->subscribe({cfg_.replyTopic}); + if (subErr != RdKafka::ERR_NO_ERROR) { + throw std::runtime_error("Failed to subscribe to reply topic '" + cfg_.replyTopic + + "': " + RdKafka::err2str(subErr)); + } + + // Brief poll so the consumer joins the group / gets assignment before produce. + // Without this, a fast reply can be published before the client is assigned. + const auto warmDeadline = + std::chrono::steady_clock::now() + std::chrono::milliseconds(1500); + while (std::chrono::steady_clock::now() < warmDeadline) { + std::unique_ptr msg(consumer_->consume(100)); + if (!msg) { + continue; + } + if (msg->err() == RdKafka::ERR_NO_ERROR) { + // Unexpected early message — leave for the next call() filter. + break; + } + if (msg->err() != RdKafka::ERR__TIMED_OUT && + msg->err() != RdKafka::ERR__PARTITION_EOF) { + // Keep warming on transient errors. + } + } + } + + started_ = true; +} + +std::vector KafkaClient::call(const unsigned char* data, std::size_t len) { + if (data == nullptr && len > 0) { + throw std::invalid_argument("KafkaClient::call: null data with non-zero length"); + } + + std::lock_guard lock(callMutex_); + ensureStarted(); + + const std::string correlationId = KafkaPluginUtil::makeUuidV4(); + RdKafka::Headers* headers = + KafkaPluginUtil::makeRpcHeaders(correlationId, cfg_.replyTopic); + + const RdKafka::ErrorCode produceErr = producer_->produce( + cfg_.requestTopic, + RdKafka::Topic::PARTITION_UA, + RdKafka::Producer::RK_MSG_COPY, + /*payload=*/const_cast(data ? data : reinterpret_cast("")), + /*len=*/len, + /*key=*/nullptr, + /*key_len=*/0, + /*timestamp=*/0, + headers, + /*msg_opaque=*/nullptr); + + if (produceErr != RdKafka::ERR_NO_ERROR) { + // produce() takes ownership of headers only on success. + delete headers; + throw std::runtime_error("Kafka produce to '" + cfg_.requestTopic + + "' failed: " + RdKafka::err2str(produceErr)); + } + + producer_->poll(0); + // Wait until the request is handed off to the broker (best-effort). + if (producer_->flush(std::min(cfg_.rpcTimeoutMs, 10000)) != RdKafka::ERR_NO_ERROR) { + throw std::runtime_error("Kafka produce flush timed out for correlation-id=" + + correlationId); + } + + const auto deadline = + std::chrono::steady_clock::now() + std::chrono::milliseconds(cfg_.rpcTimeoutMs); + + while (std::chrono::steady_clock::now() < deadline) { + const auto remaining = std::chrono::duration_cast( + deadline - std::chrono::steady_clock::now()) + .count(); + const int pollMs = static_cast(std::min(remaining, 200)); + + std::unique_ptr msg(consumer_->consume(pollMs)); + if (!msg) { + continue; + } + + if (msg->err() == RdKafka::ERR__TIMED_OUT || + msg->err() == RdKafka::ERR__PARTITION_EOF) { + continue; + } + if (msg->err() != RdKafka::ERR_NO_ERROR) { + throw std::runtime_error("Kafka consume error while waiting for reply: " + + msg->errstr()); + } + + const std::string gotId = + KafkaPluginUtil::headerValue(msg->headers(), "correlation-id"); + if (gotId != correlationId) { + // Shared reply topic / other clients — skip. + continue; + } + + const void* payload = msg->payload(); + const std::size_t payloadLen = static_cast(msg->len()); + if (!payload || payloadLen == 0) { + return {}; + } + const auto* bytes = static_cast(payload); + return std::vector(bytes, bytes + payloadLen); + } + + throw std::runtime_error("Kafka RPC timed out after " + std::to_string(cfg_.rpcTimeoutMs) + + " ms waiting for correlation-id=" + correlationId); +} diff --git a/kafka/KafkaPlugin/KafkaClient.h b/kafka/KafkaPlugin/KafkaClient.h new file mode 100644 index 0000000..8c3cc7f --- /dev/null +++ b/kafka/KafkaPlugin/KafkaClient.h @@ -0,0 +1,52 @@ +#pragma once + +#include +#include +#include +#include +#include + +namespace RdKafka { +class Conf; +class Producer; +class KafkaConsumer; +} // namespace RdKafka + +// Client-side RPC over Kafka: produce to request topic, consume correlated reply. +class KafkaClient { +public: + struct Config { + std::string brokers = "localhost:9092"; + std::string requestTopic = "graft.requests"; + std::string replyTopic = "graft.replies"; + std::string groupId = "graft-client"; + int rpcTimeoutMs = 30000; + // Optional SASL/SSL pass-through (empty = leave librdkafka defaults). + std::string securityProtocol; // e.g. "SASL_SSL", "SSL", "PLAINTEXT" + std::string saslMechanism; // e.g. "PLAIN", "SCRAM-SHA-512" + std::string saslUsername; + std::string saslPassword; + std::string sslCaLocation; + }; + + explicit KafkaClient(Config cfg); + ~KafkaClient(); + + KafkaClient(const KafkaClient&) = delete; + KafkaClient& operator=(const KafkaClient&) = delete; + + // Returns response bytes; throws on timeout or produce/consume failure. + std::vector call(const unsigned char* data, std::size_t len); + +private: + void applySecurity(RdKafka::Conf* conf) const; + void ensureStarted(); + + Config cfg_; + std::string instanceId_; + std::string consumerGroupId_; + std::unique_ptr producer_; + std::unique_ptr consumer_; + std::mutex callMutex_; + bool started_ = false; +}; diff --git a/kafka/KafkaPlugin/KafkaServer.cpp b/kafka/KafkaPlugin/KafkaServer.cpp new file mode 100644 index 0000000..dd22d45 --- /dev/null +++ b/kafka/KafkaPlugin/KafkaServer.cpp @@ -0,0 +1,308 @@ +#include "KafkaServer.h" +#include "KafkaUtil.h" + +#include + +#include +#include +#include +#include +#include + +#if __has_include() +#include +#elif __has_include() +#include +#else +#error "librdkafka C++ header not found (expected librdkafka/rdkafkacpp.h or rdkafkacpp.h)" +#endif + +namespace { + +void logInfo(const std::string& msg) { + std::cout << "[KafkaServer][INFO] " << msg << std::endl; +} + +void logWarn(const std::string& msg) { + std::cout << "[KafkaServer][WARN] " << msg << std::endl; +} + +} // namespace + +KafkaServer::~KafkaServer() { stop(); } + +void KafkaServer::configure(const char* jsonConfig, ProcessMessageFn processMessage) { + { + std::lock_guard lock(processMutex_); + process_ = processMessage; + } + if (!jsonConfig) { + return; + } + auto j = nlohmann::json::parse(jsonConfig, nullptr, false); + if (j.is_discarded() || !j.is_object()) { + return; + } + if (j.contains("brokers") && j["brokers"].is_string()) { + brokers_ = j["brokers"].get(); + } else if (j.contains("host") && j["host"].is_string()) { + brokers_ = j["host"].get(); // compatibility with other plugins + } + if (j.contains("requestTopic") && j["requestTopic"].is_string()) { + requestTopic_ = j["requestTopic"].get(); + } else if (j.contains("queue") && j["queue"].is_string()) { + requestTopic_ = j["queue"].get(); + } + if (j.contains("replyTopic") && j["replyTopic"].is_string()) { + replyTopic_ = j["replyTopic"].get(); + } else if (j.contains("replyQueue") && j["replyQueue"].is_string()) { + replyTopic_ = j["replyQueue"].get(); + } + if (j.contains("groupId") && j["groupId"].is_string()) { + groupId_ = j["groupId"].get(); + } + if (j.contains("securityProtocol") && j["securityProtocol"].is_string()) { + securityProtocol_ = j["securityProtocol"].get(); + } + if (j.contains("saslMechanism") && j["saslMechanism"].is_string()) { + saslMechanism_ = j["saslMechanism"].get(); + } + if (j.contains("saslUsername") && j["saslUsername"].is_string()) { + saslUsername_ = j["saslUsername"].get(); + } + if (j.contains("saslPassword") && j["saslPassword"].is_string()) { + saslPassword_ = j["saslPassword"].get(); + } + if (j.contains("sslCaLocation") && j["sslCaLocation"].is_string()) { + sslCaLocation_ = j["sslCaLocation"].get(); + } +} + +void KafkaServer::applySecurity(RdKafka::Conf* conf) const { + using KafkaPluginUtil::setConfOrThrow; + if (!securityProtocol_.empty()) { + setConfOrThrow(conf, "security.protocol", securityProtocol_); + } + if (!saslMechanism_.empty()) { + setConfOrThrow(conf, "sasl.mechanism", saslMechanism_); + } + if (!saslUsername_.empty()) { + setConfOrThrow(conf, "sasl.username", saslUsername_); + } + if (!saslPassword_.empty()) { + setConfOrThrow(conf, "sasl.password", saslPassword_); + } + if (!sslCaLocation_.empty()) { + setConfOrThrow(conf, "ssl.ca.location", sslCaLocation_); + } +} + +void KafkaServer::start() { + if (running_.exchange(true)) { + return; + } + worker_ = std::thread([this] { loop(); }); +} + +void KafkaServer::stop() { + if (!running_.exchange(false)) { + if (worker_.joinable()) { + worker_.join(); + } + return; + } + if (worker_.joinable()) { + worker_.join(); + } +} + +void KafkaServer::publishReply(RdKafka::Producer* producer, + const std::string& replyTopic, + const std::string& correlationId, + const std::vector& response) { + RdKafka::Headers* headers = RdKafka::Headers::create(); + if (!correlationId.empty()) { + headers->add("correlation-id", correlationId); + } + + void* payload = nullptr; + std::size_t len = 0; + if (!response.empty()) { + payload = const_cast(response.data()); + len = response.size(); + } + + const RdKafka::ErrorCode err = producer->produce( + replyTopic, + RdKafka::Topic::PARTITION_UA, + RdKafka::Producer::RK_MSG_COPY, + payload, + len, + nullptr, + 0, + 0, + headers, + nullptr); + + if (err != RdKafka::ERR_NO_ERROR) { + delete headers; + throw std::runtime_error("Failed to publish reply to '" + replyTopic + + "': " + RdKafka::err2str(err)); + } + producer->poll(0); +} + +void KafkaServer::loop() { + logInfo("starting consumer on topic='" + requestTopic_ + "' group='" + groupId_ + + "' brokers='" + brokers_ + "'"); + + while (running_) { + std::unique_ptr producer; + std::unique_ptr consumer; + + try { + using KafkaPluginUtil::setConfOrThrow; + std::string errstr; + + { + std::unique_ptr pconf( + RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL)); + setConfOrThrow(pconf.get(), "bootstrap.servers", brokers_); + setConfOrThrow(pconf.get(), "client.id", "graft-kafka-server-producer"); + applySecurity(pconf.get()); + RdKafka::Producer* raw = RdKafka::Producer::create(pconf.release(), errstr); + if (!raw) { + throw std::runtime_error("Failed to create reply producer: " + errstr); + } + producer.reset(raw); + } + + { + std::unique_ptr cconf( + RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL)); + setConfOrThrow(cconf.get(), "bootstrap.servers", brokers_); + setConfOrThrow(cconf.get(), "group.id", groupId_); + setConfOrThrow(cconf.get(), "client.id", "graft-kafka-server-consumer"); + setConfOrThrow(cconf.get(), "enable.auto.commit", "false"); + setConfOrThrow(cconf.get(), "auto.offset.reset", "earliest"); + setConfOrThrow(cconf.get(), "allow.auto.create.topics", "true"); + applySecurity(cconf.get()); + RdKafka::KafkaConsumer* raw = + RdKafka::KafkaConsumer::create(cconf.release(), errstr); + if (!raw) { + throw std::runtime_error("Failed to create request consumer: " + errstr); + } + consumer.reset(raw); + + const RdKafka::ErrorCode subErr = consumer->subscribe({requestTopic_}); + if (subErr != RdKafka::ERR_NO_ERROR) { + throw std::runtime_error("Failed to subscribe to '" + requestTopic_ + + "': " + RdKafka::err2str(subErr)); + } + } + + logInfo("Kafka server loop connected"); + + while (running_) { + std::unique_ptr msg(consumer->consume(200)); + if (!msg) { + continue; + } + if (msg->err() == RdKafka::ERR__TIMED_OUT || + msg->err() == RdKafka::ERR__PARTITION_EOF) { + producer->poll(0); + continue; + } + if (msg->err() != RdKafka::ERR_NO_ERROR) { + logWarn("consume error: " + msg->errstr()); + continue; + } + + ProcessMessageFn process = nullptr; + { + std::lock_guard lock(processMutex_); + process = process_; + } + if (!process) { + logWarn("processMessage callback is not configured; skipping message"); + consumer->commitSync(msg.get()); + continue; + } + + const std::string correlationId = + KafkaPluginUtil::headerValue(msg->headers(), "correlation-id"); + std::string replyTo = + KafkaPluginUtil::headerValue(msg->headers(), "reply-to"); + if (replyTo.empty()) { + replyTo = replyTopic_; + } + + std::vector request; + if (msg->payload() && msg->len() > 0) { + const auto* bytes = static_cast(msg->payload()); + request.assign(bytes, bytes + msg->len()); + } + + std::vector response; + auto writeResponse = [](void* context, const byte* data, std::size_t size) { + auto* out = static_cast*>(context); + if (!out) { + return; + } + if (!data || size == 0) { + out->clear(); + return; + } + out->assign(data, data + size); + }; + + const bool ok = + process(request.data(), request.size(), writeResponse, &response); + if (!ok) { + logWarn("processMessage returned false; not publishing reply"); + // Still commit to avoid poison-pill loops; gateway owns retry policy. + consumer->commitSync(msg.get()); + continue; + } + + if (!replyTo.empty()) { + publishReply(producer.get(), replyTo, correlationId, response); + const RdKafka::ErrorCode flushErr = producer->flush(10000); + if (flushErr != RdKafka::ERR_NO_ERROR) { + logWarn("reply flush incomplete: " + RdKafka::err2str(flushErr)); + } + } else { + logWarn("no reply-to / replyTopic configured; dropping response"); + } + + consumer->commitSync(msg.get()); + } + } catch (const std::exception& ex) { + logWarn(std::string("loop error: ") + ex.what()); + } catch (...) { + logWarn("loop error: unknown"); + } + + if (consumer) { + try { + consumer->close(); + } catch (...) { + } + consumer.reset(); + } + if (producer) { + try { + producer->flush(2000); + } catch (...) { + } + producer.reset(); + } + + if (running_) { + logWarn("reconnecting in 2s..."); + std::this_thread::sleep_for(std::chrono::seconds(2)); + } + } + + logInfo("stopped"); +} diff --git a/kafka/KafkaPlugin/KafkaServer.h b/kafka/KafkaPlugin/KafkaServer.h new file mode 100644 index 0000000..487f44f --- /dev/null +++ b/kafka/KafkaPlugin/KafkaServer.h @@ -0,0 +1,50 @@ +#pragma once + +#include "IServer.h" + +#include +#include +#include +#include +#include +#include + +namespace RdKafka { +class Producer; +class KafkaConsumer; +class Conf; +} // namespace RdKafka + +class KafkaServer : public GraftcodeGateway::IServer { +public: + KafkaServer() = default; + ~KafkaServer() override; + + void configure(const char* jsonConfig, ProcessMessageFn processMessage) override; + void start() override; + void stop() override; + +private: + void loop(); + void applySecurity(RdKafka::Conf* conf) const; + void publishReply(RdKafka::Producer* producer, + const std::string& replyTopic, + const std::string& correlationId, + const std::vector& response); + + ProcessMessageFn process_{nullptr}; + std::mutex processMutex_; + + std::string brokers_ = "localhost:9092"; + std::string requestTopic_ = "graft.requests"; + std::string replyTopic_ = "graft.replies"; + std::string groupId_ = "graft-gateway"; + std::string securityProtocol_; + std::string saslMechanism_; + std::string saslUsername_; + std::string saslPassword_; + std::string sslCaLocation_; + + std::atomic running_{false}; + std::thread worker_; +}; diff --git a/kafka/KafkaPlugin/KafkaUtil.h b/kafka/KafkaPlugin/KafkaUtil.h new file mode 100644 index 0000000..e118e4f --- /dev/null +++ b/kafka/KafkaPlugin/KafkaUtil.h @@ -0,0 +1,74 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include + +#if __has_include() +#include +#elif __has_include() +#include +#else +#error "librdkafka C++ header not found (expected librdkafka/rdkafkacpp.h or rdkafkacpp.h)" +#endif + +namespace KafkaPluginUtil { + +inline std::string makeUuidV4() { + std::random_device rd; + std::mt19937_64 gen(rd()); + std::uniform_int_distribution dist; + + const std::uint64_t a = dist(gen); + const std::uint64_t b = dist(gen); + + // RFC 4122 variant 1, version 4 + const std::uint64_t time_low = a & 0xFFFFFFFFULL; + const std::uint64_t time_mid = (a >> 32) & 0xFFFFULL; + const std::uint64_t time_hi = ((a >> 48) & 0x0FFFULL) | 0x4000ULL; + const std::uint64_t clock_seq = ((b >> 48) & 0x3FFFULL) | 0x8000ULL; + const std::uint64_t node = b & 0xFFFFFFFFFFFFULL; + + std::ostringstream oss; + oss << std::hex << std::setfill('0') + << std::setw(8) << time_low << '-' + << std::setw(4) << time_mid << '-' + << std::setw(4) << time_hi << '-' + << std::setw(4) << clock_seq << '-' + << std::setw(12) << node; + return oss.str(); +} + +inline std::string headerValue(const RdKafka::Headers* headers, const std::string& key) { + if (!headers) { + return {}; + } + const auto result = headers->get(key); + if (result.empty() || !result[0].value() || result[0].value_size() == 0) { + return {}; + } + return std::string(static_cast(result[0].value()), result[0].value_size()); +} + +inline RdKafka::Headers* makeRpcHeaders(const std::string& correlationId, + const std::string& replyTo) { + RdKafka::Headers* headers = RdKafka::Headers::create(); + headers->add("correlation-id", correlationId); + if (!replyTo.empty()) { + headers->add("reply-to", replyTo); + } + return headers; +} + +inline void setConfOrThrow(RdKafka::Conf* conf, const std::string& key, const std::string& value) { + std::string errstr; + if (conf->set(key, value, errstr) != RdKafka::Conf::CONF_OK) { + throw std::runtime_error("librdkafka conf set '" + key + "' failed: " + errstr); + } +} + +} // namespace KafkaPluginUtil diff --git a/kafka/KafkaPlugin/TransportKafka.cpp b/kafka/KafkaPlugin/TransportKafka.cpp new file mode 100644 index 0000000..a25129c --- /dev/null +++ b/kafka/KafkaPlugin/TransportKafka.cpp @@ -0,0 +1,97 @@ +#include "TransportKafka.h" +#include "KafkaServer.h" + +#include + +#include + +#if defined(_WIN32) +#define KAFKA_PLUGIN_EXPORT extern "C" __declspec(dllexport) +#else +#define KAFKA_PLUGIN_EXPORT extern "C" +#endif + +static KafkaClient::Config parseClientConfig(const char* configJson) { + KafkaClient::Config c; + if (!configJson) { + return c; + } + auto j = nlohmann::json::parse(configJson, nullptr, false); + if (j.is_discarded() || !j.is_object()) { + return c; + } + if (j.contains("brokers") && j["brokers"].is_string()) { + c.brokers = j["brokers"].get(); + } else if (j.contains("host") && j["host"].is_string()) { + c.brokers = j["host"].get(); + } + if (j.contains("requestTopic") && j["requestTopic"].is_string()) { + c.requestTopic = j["requestTopic"].get(); + } else if (j.contains("queue") && j["queue"].is_string()) { + c.requestTopic = j["queue"].get(); + } + if (j.contains("replyTopic") && j["replyTopic"].is_string()) { + c.replyTopic = j["replyTopic"].get(); + } else if (j.contains("replyQueue") && j["replyQueue"].is_string()) { + c.replyTopic = j["replyQueue"].get(); + } + if (j.contains("groupId") && j["groupId"].is_string()) { + c.groupId = j["groupId"].get(); + } + if (j.contains("rpcTimeoutMs") && j["rpcTimeoutMs"].is_number_integer()) { + c.rpcTimeoutMs = j["rpcTimeoutMs"].get(); + } + if (j.contains("securityProtocol") && j["securityProtocol"].is_string()) { + c.securityProtocol = j["securityProtocol"].get(); + } + if (j.contains("saslMechanism") && j["saslMechanism"].is_string()) { + c.saslMechanism = j["saslMechanism"].get(); + } + if (j.contains("saslUsername") && j["saslUsername"].is_string()) { + c.saslUsername = j["saslUsername"].get(); + } + if (j.contains("saslPassword") && j["saslPassword"].is_string()) { + c.saslPassword = j["saslPassword"].get(); + } + if (j.contains("sslCaLocation") && j["sslCaLocation"].is_string()) { + c.sslCaLocation = j["sslCaLocation"].get(); + } + return c; +} + +TransportKafka::TransportKafka(const char*, unsigned short, const char* configJson) + : client_(std::make_unique(parseClientConfig(configJson))) {} + +int TransportKafka::Initialize(byte, byte, byte) { return 0; } + +int TransportKafka::SendCommand(byte* messageByteArray, int32_t messageByteArrayLen) { + lastResponse_ = client_->call(messageByteArray, static_cast(messageByteArrayLen)); + return static_cast(lastResponse_.size()); +} + +int TransportKafka::ReadResponse(byte* responseByteArray, int32_t responseByteArrayLen) { + if (responseByteArrayLen < static_cast(lastResponse_.size())) { + return -1; + } + if (!lastResponse_.empty()) { + std::memcpy(responseByteArray, lastResponse_.data(), lastResponse_.size()); + } + return 0; +} + +KAFKA_PLUGIN_EXPORT Hypertube::Native::Interfaces::ITransport* +CreateTransportChannel(const char* ipAddress, const unsigned short port, const char* configSource) { + return new TransportKafka(ipAddress, port, configSource); +} + +KAFKA_PLUGIN_EXPORT void DestroyTransportChannel(Hypertube::Native::Interfaces::ITransport* transport) { + delete transport; +} + +KAFKA_PLUGIN_EXPORT GraftcodeGateway::IServer* CreateServer() { + return new KafkaServer(); +} + +KAFKA_PLUGIN_EXPORT void DestroyServer(GraftcodeGateway::IServer* server) { + delete server; +} diff --git a/kafka/KafkaPlugin/TransportKafka.h b/kafka/KafkaPlugin/TransportKafka.h new file mode 100644 index 0000000..e8dcbdc --- /dev/null +++ b/kafka/KafkaPlugin/TransportKafka.h @@ -0,0 +1,17 @@ +#pragma once +#include "ITransport.h" +#include "KafkaClient.h" +#include +#include + +class TransportKafka : public Hypertube::Native::Interfaces::ITransport { +public: + TransportKafka(const char* /*ip*/, unsigned short /*port*/, const char* configJson); + int Initialize(byte callingRuntimeNumber, byte calledRuntimeNumber, byte calledRuntimeVersion) override; + int SendCommand(byte* messageByteArray, int32_t messageByteArrayLen) override; + int ReadResponse(byte* responseByteArray, int32_t responseByteArrayLen) override; + +private: + std::unique_ptr client_; + std::vector lastResponse_; +}; diff --git a/kafka/KafkaPluginTest/CMakeLists.txt b/kafka/KafkaPluginTest/CMakeLists.txt new file mode 100644 index 0000000..d65dd1b --- /dev/null +++ b/kafka/KafkaPluginTest/CMakeLists.txt @@ -0,0 +1,6 @@ +add_executable(KafkaPluginTest smoke.cpp) +target_link_libraries(KafkaPluginTest PRIVATE KafkaPlugin) +target_include_directories(KafkaPluginTest PRIVATE + "${CMAKE_SOURCE_DIR}/GraftcodePluginsInterfaces" +) +add_test(NAME KafkaPluginSmoke COMMAND KafkaPluginTest) diff --git a/kafka/KafkaPluginTest/smoke.cpp b/kafka/KafkaPluginTest/smoke.cpp new file mode 100644 index 0000000..b68b717 --- /dev/null +++ b/kafka/KafkaPluginTest/smoke.cpp @@ -0,0 +1,62 @@ +#include "IServer.h" +#include "ITransport.h" + +#include +#include + +#if defined(_WIN32) +#define KAFKA_PLUGIN_IMPORT extern "C" __declspec(dllimport) +#else +#define KAFKA_PLUGIN_IMPORT extern "C" +#endif + +KAFKA_PLUGIN_IMPORT GraftcodeGateway::IServer* CreateServer(); +KAFKA_PLUGIN_IMPORT void DestroyServer(GraftcodeGateway::IServer* server); +KAFKA_PLUGIN_IMPORT Hypertube::Native::Interfaces::ITransport* CreateTransportChannel( + const char* ipAddress, unsigned short port, const char* configSource); +KAFKA_PLUGIN_IMPORT void DestroyTransportChannel(Hypertube::Native::Interfaces::ITransport* transport); + +static bool g_processCalled = false; + +static bool smokeProcess(const GraftcodeGateway::IServer::byte* /*requestData*/, + std::size_t /*requestSize*/, + GraftcodeGateway::IServer::WriteResponseFn writeResponse, + void* writeContext) { + g_processCalled = true; + static const unsigned char kPayload[] = {'o', 'k'}; + if (writeResponse) { + writeResponse(writeContext, kPayload, sizeof(kPayload)); + } + return true; +} + +int main() { + // Link / symbol smoke — no broker required. + GraftcodeGateway::IServer* server = CreateServer(); + if (!server) { + std::puts("FAIL: CreateServer returned null"); + return 1; + } + + const char* cfg = + R"({"brokers":"127.0.0.1:1","requestTopic":"graft.requests","replyTopic":"graft.replies","groupId":"graft-smoke"})"; + server->configure(cfg, smokeProcess); + // Do not start() — that would spin a reconnect loop against a missing broker. + DestroyServer(server); + + Hypertube::Native::Interfaces::ITransport* transport = + CreateTransportChannel("127.0.0.1", 0, cfg); + if (!transport) { + std::puts("FAIL: CreateTransportChannel returned null"); + return 1; + } + if (transport->Initialize(1, 2, 1) != 0) { + std::puts("FAIL: Initialize returned non-zero"); + DestroyTransportChannel(transport); + return 1; + } + DestroyTransportChannel(transport); + + std::puts("KafkaPlugin smoke OK — factories linked; run against local Kafka to exercise RPC"); + return 0; +} diff --git a/kafka/Readme.md b/kafka/Readme.md new file mode 100644 index 0000000..722d44d --- /dev/null +++ b/kafka/Readme.md @@ -0,0 +1,210 @@ +# Kafka Plugin Build (CMake) + +This plugin is the Apache Kafka counterpart of the RabbitMQ / Azure Service Bus +plugins. It implements the same Graftcode plugin interfaces +(`Hypertube::Native::Interfaces::ITransport` for the calling runtime and +`GraftcodeGateway::IServer` for the gateway) and exposes the same exported +factory symbols (`CreateTransportChannel` / `DestroyTransportChannel` and +`CreateServer` / `DestroyServer`). + +It is written in C++ and talks to Kafka using **librdkafka** (`rdkafka++`), +acquired either through CMake `FetchContent` (default) or vcpkg / a system +install. + +## RPC model + +Kafka has no native request/reply. This plugin mirrors the Service Bus / AMQP +pattern with message headers: + +1. **Client** produces to `requestTopic` with headers: + - `correlation-id` — UUID per call + - `reply-to` — reply topic the server should use +2. **Server** (GG plugin) consumes `requestTopic`, runs `processMessage`, then + produces to the `reply-to` topic (falling back to configured `replyTopic`) + echoing the same `correlation-id`. +3. **Client** consumes `replyTopic` until a message with a matching + `correlation-id` arrives, or `rpcTimeoutMs` elapses. + +Each client instance appends a unique suffix to `groupId` so concurrent clients +sharing one reply topic each receive a copy of replies and filter by +`correlation-id`. For higher fan-out efficiency you can give each client its own +`replyTopic` (and set it in config); the server always honors `reply-to`. + +## 1) Place this folder in graftcode-extensions + +Copy the contents of this directory to `kafka/` inside +[graftcode-extensions](https://github.com/grft-dev/graftcode-extensions) +(alongside `rabbitmq/` and `servicebus/`). + +```bash +cd graftcode-extensions/kafka +``` + +## 2) Configure with CMake (FetchContent — default) + +```bash +cmake -S . -B build -DCMAKE_BUILD_TYPE=Release +cmake --build build --config Release +``` + +This downloads **nlohmann/json** and **librdkafka** (`KAFKA_LIBRDKAFKA_GIT_TAG`, +default `v2.8.0`) via FetchContent and links a static `rdkafka++` into the +shared plugin. + +Optional flags: + +| Flag | Meaning | +|------|---------| +| `-DKAFKA_LIBRDKAFKA_GIT_TAG=v2.15.0` | Pin a different librdkafka release | +| `-DKAFKA_USE_SYSTEM_RDKAFKA=ON` | Do not FetchContent; require `find_package(RdKafka)` | + +### Alternative: vcpkg + +```bash +git clone https://github.com/microsoft/vcpkg.git +./vcpkg/bootstrap-vcpkg.sh # Windows: .\vcpkg\bootstrap-vcpkg.bat + +cmake -S . -B build -DCMAKE_BUILD_TYPE=Release \ + -DCMAKE_TOOLCHAIN_FILE=./vcpkg/scripts/buildsystems/vcpkg.cmake \ + -DKAFKA_USE_SYSTEM_RDKAFKA=ON +cmake --build build --config Release +``` + +Manifest dependencies are declared in `vcpkg.json` (`librdkafka`, `nlohmann-json`). + +## 3) Build output + +- `kafka/build/KafkaPlugin/KafkaPlugin.dll` (Windows) +- `kafka/build/KafkaPlugin/libKafkaPlugin.so` / `.dylib` (Linux / macOS) + +If the generated library is `libKafkaPlugin.*`, use plugin name +`libKafkaPlugin` in config. + +## 4) Download GG + +Download `gg` from: +https://github.com/grft-dev/graftcode-gateway/releases/ + +## 5) Run a local broker + +### Apache Kafka (KRaft) + +```bash +docker compose up -d +./scripts/create-topics.sh +# or: KAFKA_CONTAINER= ./scripts/create-topics.sh +``` + +Brokers: `localhost:9092`. + +### Redpanda + +```bash +docker compose -f docker-compose.redpanda.yml up -d +# topics (example): +docker exec -it rpk topic create graft.requests graft.replies +``` + +Brokers: `localhost:19092` — set `"brokers": "localhost:19092"` in config. + +## 6) Run GG with a sample library + +Create `pluginConfig.json` next to your module (a ready sample lives in this +folder): + +```json +{ + "name": "KafkaPlugin", + "brokers": "localhost:9092", + "requestTopic": "graft.requests", + "replyTopic": "graft.replies", + "groupId": "graft-gateway", + "rpcTimeoutMs": 30000 +} +``` + +If you built `libKafkaPlugin.*`, set `"name": "libKafkaPlugin"`. + +```bash +./gg ./YourLibrary.dll --config ./pluginConfig.json +``` + +## 7) Get installation command + +Visit `http://localhost:81/GV`, select your package manager, and copy the +generated installation command. + +## 8) Configure Graft after installation + +```csharp +string configSource = +""" +{ + "configurations": { + "graft.nuget.PhysicsCalculator": { + "runtime": "netcore", + "stateless": true, + "plugin": { + "name": "KafkaPlugin", + "brokers": "localhost:9092", + "requestTopic": "graft.requests", + "replyTopic": "graft.replies", + "groupId": "graft-client", + "rpcTimeoutMs": 30000 + } + } + } +} +"""; + +graft.nuget.PhysicsCalculator.GraftConfig.SetConfig(configSource); +``` + +## Configuration reference + +| Field | Required | Description | +|-------|----------|-------------| +| `brokers` | yes* | Kafka bootstrap servers (`host:port[,host:port…]`). | +| `host` | * | Alias for `brokers` (compatibility with other plugins). | +| `requestTopic` | yes | Topic requests are produced to / consumed from. | +| `queue` | | Alias for `requestTopic`. | +| `replyTopic` | yes (client) | Topic replies are consumed from / default produce target. | +| `replyQueue` | | Alias for `replyTopic`. | +| `groupId` | no | Consumer group. Server default `graft-gateway`. Client default `graft-client` (a unique instance suffix is always appended on the client). | +| `rpcTimeoutMs` | no | Client request/response timeout in milliseconds (default `30000`). | +| `securityProtocol` | no | librdkafka `security.protocol` (e.g. `SASL_SSL`, `SSL`, `PLAINTEXT`). | +| `saslMechanism` | no | e.g. `PLAIN`, `SCRAM-SHA-256`, `SCRAM-SHA-512`. | +| `saslUsername` | no | SASL username. | +| `saslPassword` | no | SASL password. | +| `sslCaLocation` | no | Path to CA PEM for SSL verification. | + +\* Provide `brokers` or `host`. + +## Smoke test + +```bash +cmake -S . -B build -DCMAKE_BUILD_TYPE=Release +cmake --build build --config Release +ctest --test-dir build --output-on-failure +``` + +`KafkaPluginTest` links the shared library and exercises +`CreateServer` / `CreateTransportChannel` without requiring a broker. +End-to-end RPC still needs a running Kafka/Redpanda and the topics above. + +## Layout (drop-in as `kafka/`) + +``` +kafka/ + CMakeLists.txt + vcpkg.json + pluginConfig.json + docker-compose.yml + docker-compose.redpanda.yml + Readme.md + BUILD_NOTES.md + GraftcodePluginsInterfaces/ + KafkaPlugin/ + KafkaPluginTest/ + scripts/create-topics.sh +``` diff --git a/kafka/docker-compose.redpanda.yml b/kafka/docker-compose.redpanda.yml new file mode 100644 index 0000000..eff7477 --- /dev/null +++ b/kafka/docker-compose.redpanda.yml @@ -0,0 +1,10 @@ +services: + redpanda: + image: redpandadata/redpanda:v24.2.4 + command: + - redpanda start + - --kafka-addr internal://0.0.0.0:9092,external://0.0.0.0:19092 + - --advertise-kafka-addr internal://redpanda:9092,external://localhost:19092 + - --mode dev-container + ports: + - "19092:19092" diff --git a/kafka/docker-compose.yml b/kafka/docker-compose.yml new file mode 100644 index 0000000..d892980 --- /dev/null +++ b/kafka/docker-compose.yml @@ -0,0 +1,19 @@ +# Lokalny Kafka (KRaft, bez ZooKeeper) — wystarczy do testów pluginu. +services: + kafka: + image: apache/kafka:3.8.1 + ports: + - "9092:9092" + environment: + KAFKA_NODE_ID: 1 + KAFKA_PROCESS_ROLES: broker,controller + KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093 + KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092 + KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER + KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT + KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093 + KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 + KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 + KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 + KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0 + CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk diff --git a/kafka/pluginConfig.json b/kafka/pluginConfig.json new file mode 100644 index 0000000..e607326 --- /dev/null +++ b/kafka/pluginConfig.json @@ -0,0 +1,8 @@ +{ + "name": "KafkaPlugin", + "brokers": "localhost:9092", + "requestTopic": "graft.requests", + "replyTopic": "graft.replies", + "groupId": "graft-gateway", + "rpcTimeoutMs": 30000 +} diff --git a/kafka/scripts/create-topics.sh b/kafka/scripts/create-topics.sh new file mode 100644 index 0000000..efbb103 --- /dev/null +++ b/kafka/scripts/create-topics.sh @@ -0,0 +1,53 @@ +#!/usr/bin/env bash +# Create the default Graftcode Kafka request/reply topics. +set -euo pipefail + +BROKERS="${BROKERS:-localhost:9092}" +REQUEST_TOPIC="${REQUEST_TOPIC:-graft.requests}" +REPLY_TOPIC="${REPLY_TOPIC:-graft.replies}" +CONTAINER="${KAFKA_CONTAINER:-}" + +create_with_kafka_topics() { + local bin="$1" + "$bin" --bootstrap-server "$BROKERS" --create --if-not-exists --topic "$REQUEST_TOPIC" --partitions 1 --replication-factor 1 + "$bin" --bootstrap-server "$BROKERS" --create --if-not-exists --topic "$REPLY_TOPIC" --partitions 1 --replication-factor 1 +} + +if [[ -n "$CONTAINER" ]]; then + echo "Creating topics via docker exec on container '$CONTAINER' (bootstrap $BROKERS)..." + # apache/kafka image path + if docker exec "$CONTAINER" test -x /opt/kafka/bin/kafka-topics.sh; then + docker exec "$CONTAINER" /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 \ + --create --if-not-exists --topic "$REQUEST_TOPIC" --partitions 1 --replication-factor 1 + docker exec "$CONTAINER" /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 \ + --create --if-not-exists --topic "$REPLY_TOPIC" --partitions 1 --replication-factor 1 + # bitnami / other layouts + elif docker exec "$CONTAINER" sh -c 'command -v kafka-topics.sh' >/dev/null 2>&1; then + docker exec "$CONTAINER" kafka-topics.sh --bootstrap-server localhost:9092 \ + --create --if-not-exists --topic "$REQUEST_TOPIC" --partitions 1 --replication-factor 1 + docker exec "$CONTAINER" kafka-topics.sh --bootstrap-server localhost:9092 \ + --create --if-not-exists --topic "$REPLY_TOPIC" --partitions 1 --replication-factor 1 + else + echo "Could not find kafka-topics.sh inside $CONTAINER" >&2 + exit 1 + fi +elif command -v kafka-topics.sh >/dev/null 2>&1; then + create_with_kafka_topics kafka-topics.sh +elif command -v rpk >/dev/null 2>&1; then + echo "Creating topics via rpk (brokers $BROKERS)..." + rpk topic create "$REQUEST_TOPIC" -X brokers="$BROKERS" || true + rpk topic create "$REPLY_TOPIC" -X brokers="$BROKERS" || true +else + # Auto-detect a running compose kafka container + CONTAINER="$(docker ps --format '{{.Names}}' | grep -E 'kafka|redpanda' | head -n1 || true)" + if [[ -z "$CONTAINER" ]]; then + echo "No kafka-topics.sh/rpk on PATH and no kafka/redpanda container running." >&2 + echo "Start the broker first: docker compose up -d" >&2 + echo "Or set KAFKA_CONTAINER=." >&2 + exit 1 + fi + KAFKA_CONTAINER="$CONTAINER" "$0" + exit $? +fi + +echo "Topics ready: $REQUEST_TOPIC, $REPLY_TOPIC" diff --git a/kafka/vcpkg.json b/kafka/vcpkg.json new file mode 100644 index 0000000..9aadb2b --- /dev/null +++ b/kafka/vcpkg.json @@ -0,0 +1,9 @@ +{ + "name": "graftcode-kafka-plugin", + "version": "1.0.0", + "description": "Graftcode Kafka transport/server plugin built on librdkafka (rdkafka++).", + "dependencies": [ + "librdkafka", + "nlohmann-json" + ] +} From 526ab916c4ba5147bad7fcd8b90dce77b788eaeb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Komor?= Date: Wed, 16 Sep 2026 09:03:30 +0200 Subject: [PATCH 02/10] update gitignore --- .gitignore | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/.gitignore b/.gitignore index d6e96c3..a8ffd37 100644 --- a/.gitignore +++ b/.gitignore @@ -3,4 +3,7 @@ .cursor out/ -build/ \ No newline at end of file +build/ +bin/ +obj/ +Binaries/ From 5ec06ce62ff7da2a04f9459d45542d372e3d2009 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Komor?= Date: Wed, 16 Sep 2026 09:10:28 +0200 Subject: [PATCH 03/10] Drop vcpkg; fetch plugin deps via CMake FetchContent Align kafka and servicebus with rabbitmq: nlohmann/json and transport libraries come from FetchContent. Remove vcpkg.json manifests and toolchain-based docs. --- kafka/BUILD_NOTES.md | 36 +---- kafka/KafkaPlugin/CMakeLists.txt | 91 ++++--------- kafka/Readme.md | 146 +++------------------ kafka/vcpkg.json | 9 -- servicebus/Readme.md | 40 +++--- servicebus/ServiceBusPlugin/CMakeLists.txt | 58 ++++++-- servicebus/vcpkg.json | 9 -- 7 files changed, 111 insertions(+), 278 deletions(-) delete mode 100644 kafka/vcpkg.json delete mode 100644 servicebus/vcpkg.json diff --git a/kafka/BUILD_NOTES.md b/kafka/BUILD_NOTES.md index 5585967..7524eb5 100644 --- a/kafka/BUILD_NOTES.md +++ b/kafka/BUILD_NOTES.md @@ -1,33 +1,5 @@ -# Build notes +# Kafka plugin build notes -## Verified in this workspace (2026-09-16 Europe/Warsaw) - -| Step | Result | -|------|--------| -| `cmake -S . -B build -DCMAKE_BUILD_TYPE=Release` | **Succeeded** (FetchContent nlohmann/json + librdkafka `v2.8.0`) | -| `cmake --build build -j$(nproc)` | **Succeeded** → `build/KafkaPlugin/libKafkaPlugin.so` | -| `ctest --test-dir build` | **Passed** (`KafkaPluginSmoke`) | -| Exported symbols | `CreateServer`, `DestroyServer`, `CreateTransportChannel`, `DestroyTransportChannel` | - -Toolchain used: GCC 14, CMake 3.31, OpenSSL 3.5, libsasl2, zlib, zstd, libcurl. - -## Include path note - -FetchContent builds expose ``; packaged/vcpkg installs typically use -``. Sources use `__has_include` to accept either. - -## Possible blockers elsewhere - -1. **Network / git** required on first configure for FetchContent. -2. **librdkafka compile time** is several minutes; if the build is OOM-killed, use `cmake --build build -j2`. -3. Missing SSL/SASL packages: install `libssl-dev`, `libsasl2-dev`, `zlib1g-dev`, `libzstd-dev` (and optionally `libcurl4-openssl-dev`). -4. **Windows**: prefer vcpkg + `-DKAFKA_USE_SYSTEM_RDKAFKA=ON` with the vcpkg toolchain. -5. Smoke test must **not** call `IServer::start()` or `SendCommand` without a broker (reconnect / RPC timeout). - -## Suggested PR checklist - -- [x] Source complete (client + server RPC with correlation-id / reply-to) -- [x] CMake FetchContent + vcpkg.json -- [x] Linux configure/build/smoke in this environment -- [ ] Manual GG round-trip with `docker compose up -d` + `./scripts/create-topics.sh` -- [ ] Copy folder contents to `graftcode-extensions/kafka/` (flat drop-in; do not nest an extra sketch directory) +- Dependencies: CMake `FetchContent` only (nlohmann/json + librdkafka). No vcpkg. +- Smoke test links factory exports; full RPC needs a live broker (see docker-compose). +- First configure needs network/git for FetchContent downloads. diff --git a/kafka/KafkaPlugin/CMakeLists.txt b/kafka/KafkaPlugin/CMakeLists.txt index a840585..58671ea 100644 --- a/kafka/KafkaPlugin/CMakeLists.txt +++ b/kafka/KafkaPlugin/CMakeLists.txt @@ -13,86 +13,43 @@ target_include_directories(${target_name} PUBLIC include(FetchContent) -# --------------------------------------------------------------------------- -# nlohmann/json -# --------------------------------------------------------------------------- -find_package(nlohmann_json CONFIG QUIET) -if(NOT nlohmann_json_FOUND) - FetchContent_Declare( - json - GIT_REPOSITORY https://github.com/nlohmann/json.git - GIT_TAG v3.12.0 - GIT_SHALLOW TRUE - ) - FetchContent_MakeAvailable(json) -endif() - -# --------------------------------------------------------------------------- -# librdkafka (rdkafka++) -# -# Preferred resolution order: -# 1) find_package(RdKafka) — system install or vcpkg toolchain -# 2) FetchContent from confluentinc/librdkafka (edenhill redirects here) -# -# Override the FetchContent tag with -DKAFKA_LIBRDKAFKA_GIT_TAG=vX.Y.Z -# Skip FetchContent with -DKAFKA_USE_SYSTEM_RDKAFKA=ON (requires find_package). -# --------------------------------------------------------------------------- -set(KAFKA_LIBRDKAFKA_GIT_TAG "v2.8.0" CACHE STRING "librdkafka git tag for FetchContent") -option(KAFKA_USE_SYSTEM_RDKAFKA "Require system/vcpkg RdKafka; do not FetchContent" OFF) - -find_package(RdKafka CONFIG QUIET) -set(_kafka_rdkafka_target "") - -if(RdKafka_FOUND) - if(TARGET RdKafka::rdkafka++) - set(_kafka_rdkafka_target RdKafka::rdkafka++) - elseif(TARGET rdkafka++) - set(_kafka_rdkafka_target rdkafka++) - endif() -endif() - -if(NOT _kafka_rdkafka_target) - if(KAFKA_USE_SYSTEM_RDKAFKA) - message(FATAL_ERROR - "KAFKA_USE_SYSTEM_RDKAFKA=ON but find_package(RdKafka) failed. " - "Install librdkafka or configure with the vcpkg toolchain.") - endif() - - message(STATUS "RdKafka not found via find_package; FetchContent ${KAFKA_LIBRDKAFKA_GIT_TAG}") +FetchContent_Declare( + json + GIT_REPOSITORY https://github.com/nlohmann/json.git + GIT_TAG v3.12.0 + GIT_SHALLOW TRUE +) +FetchContent_MakeAvailable(json) - # Prefer a static librdkafka so the shared plugin is easier to drop in. - set(RDKAFKA_BUILD_STATIC ON CACHE BOOL "Build librdkafka as static library" FORCE) - set(RDKAFKA_BUILD_EXAMPLES OFF CACHE BOOL "Skip librdkafka examples" FORCE) - set(RDKAFKA_BUILD_TESTS OFF CACHE BOOL "Skip librdkafka tests" FORCE) - set(WITH_SSL ON CACHE BOOL "Enable SSL in librdkafka" FORCE) +set(RDKAFKA_BUILD_STATIC ON CACHE BOOL "Build librdkafka as static library" FORCE) +set(RDKAFKA_BUILD_EXAMPLES OFF CACHE BOOL "Skip librdkafka examples" FORCE) +set(RDKAFKA_BUILD_TESTS OFF CACHE BOOL "Skip librdkafka tests" FORCE) - FetchContent_Declare( - librdkafka - GIT_REPOSITORY https://github.com/confluentinc/librdkafka.git - GIT_TAG ${KAFKA_LIBRDKAFKA_GIT_TAG} - GIT_SHALLOW TRUE - ) - FetchContent_MakeAvailable(librdkafka) +FetchContent_Declare( + librdkafka + GIT_REPOSITORY https://github.com/confluentinc/librdkafka.git + GIT_TAG v2.8.0 + GIT_SHALLOW TRUE +) +FetchContent_MakeAvailable(librdkafka) - if(TARGET rdkafka++) - set(_kafka_rdkafka_target rdkafka++) - elseif(TARGET RdKafka::rdkafka++) - set(_kafka_rdkafka_target RdKafka::rdkafka++) - else() - message(FATAL_ERROR "librdkafka FetchContent succeeded but rdkafka++ target is missing") - endif() +if(NOT TARGET rdkafka++) + message(FATAL_ERROR "librdkafka FetchContent succeeded but rdkafka++ target was not found") endif() target_link_libraries(${target_name} PUBLIC nlohmann_json::nlohmann_json - ${_kafka_rdkafka_target} + rdkafka++ ) -# Ensure consumers see librdkafka headers when linking a plain target name. if(TARGET rdkafka) target_link_libraries(${target_name} PUBLIC rdkafka) endif() +if(WIN32 OR CMAKE_SYSTEM_NAME STREQUAL "Windows") + target_link_libraries(${target_name} PUBLIC ws2_32) +endif() + if(UNIX AND NOT APPLE) find_package(Threads REQUIRED) target_link_libraries(${target_name} PUBLIC Threads::Threads) diff --git a/kafka/Readme.md b/kafka/Readme.md index 722d44d..a8714c7 100644 --- a/kafka/Readme.md +++ b/kafka/Readme.md @@ -8,8 +8,7 @@ factory symbols (`CreateTransportChannel` / `DestroyTransportChannel` and `CreateServer` / `DestroyServer`). It is written in C++ and talks to Kafka using **librdkafka** (`rdkafka++`), -acquired either through CMake `FetchContent` (default) or vcpkg / a system -install. +acquired through CMake `FetchContent` (same pattern as the RabbitMQ plugin). ## RPC model @@ -25,55 +24,29 @@ pattern with message headers: 3. **Client** consumes `replyTopic` until a message with a matching `correlation-id` arrives, or `rpcTimeoutMs` elapses. -Each client instance appends a unique suffix to `groupId` so concurrent clients -sharing one reply topic each receive a copy of replies and filter by -`correlation-id`. For higher fan-out efficiency you can give each client its own -`replyTopic` (and set it in config); the server always honors `reply-to`. - -## 1) Place this folder in graftcode-extensions - -Copy the contents of this directory to `kafka/` inside -[graftcode-extensions](https://github.com/grft-dev/graftcode-extensions) -(alongside `rabbitmq/` and `servicebus/`). +## 1) Clone repository ```bash +git clone https://github.com/grft-dev/graftcode-extensions.git cd graftcode-extensions/kafka ``` -## 2) Configure with CMake (FetchContent — default) +## 2) Configure with CMake ```bash cmake -S . -B build -DCMAKE_BUILD_TYPE=Release -cmake --build build --config Release ``` -This downloads **nlohmann/json** and **librdkafka** (`KAFKA_LIBRDKAFKA_GIT_TAG`, -default `v2.8.0`) via FetchContent and links a static `rdkafka++` into the -shared plugin. - -Optional flags: - -| Flag | Meaning | -|------|---------| -| `-DKAFKA_LIBRDKAFKA_GIT_TAG=v2.15.0` | Pin a different librdkafka release | -| `-DKAFKA_USE_SYSTEM_RDKAFKA=ON` | Do not FetchContent; require `find_package(RdKafka)` | - -### Alternative: vcpkg +## 3) Build ```bash -git clone https://github.com/microsoft/vcpkg.git -./vcpkg/bootstrap-vcpkg.sh # Windows: .\vcpkg\bootstrap-vcpkg.bat - -cmake -S . -B build -DCMAKE_BUILD_TYPE=Release \ - -DCMAKE_TOOLCHAIN_FILE=./vcpkg/scripts/buildsystems/vcpkg.cmake \ - -DKAFKA_USE_SYSTEM_RDKAFKA=ON cmake --build build --config Release ``` -Manifest dependencies are declared in `vcpkg.json` (`librdkafka`, `nlohmann-json`). - -## 3) Build output +CMake downloads **nlohmann/json** and **librdkafka** (`v2.8.0`) via FetchContent +and links a static `rdkafka++` into the shared plugin. +As a result, you will receive: - `kafka/build/KafkaPlugin/KafkaPlugin.dll` (Windows) - `kafka/build/KafkaPlugin/libKafkaPlugin.so` / `.dylib` (Linux / macOS) @@ -90,27 +63,20 @@ https://github.com/grft-dev/graftcode-gateway/releases/ ### Apache Kafka (KRaft) ```bash -docker compose up -d +docker compose -f docker-compose.yml up -d ./scripts/create-topics.sh -# or: KAFKA_CONTAINER= ./scripts/create-topics.sh ``` -Brokers: `localhost:9092`. - -### Redpanda +### Redpanda (lighter alternative) ```bash docker compose -f docker-compose.redpanda.yml up -d -# topics (example): -docker exec -it rpk topic create graft.requests graft.replies +# brokers: localhost:19092 ``` -Brokers: `localhost:19092` — set `"brokers": "localhost:19092"` in config. - -## 6) Run GG with a sample library +## 6) Run GG with sample library -Create `pluginConfig.json` next to your module (a ready sample lives in this -folder): +Create `pluginConfig.json` (see also the example in this folder): ```json { @@ -123,88 +89,18 @@ folder): } ``` -If you built `libKafkaPlugin.*`, set `"name": "libKafkaPlugin"`. +Then run: -```bash -./gg ./YourLibrary.dll --config ./pluginConfig.json -``` - -## 7) Get installation command - -Visit `http://localhost:81/GV`, select your package manager, and copy the -generated installation command. - -## 8) Configure Graft after installation - -```csharp -string configSource = -""" -{ - "configurations": { - "graft.nuget.PhysicsCalculator": { - "runtime": "netcore", - "stateless": true, - "plugin": { - "name": "KafkaPlugin", - "brokers": "localhost:9092", - "requestTopic": "graft.requests", - "replyTopic": "graft.replies", - "groupId": "graft-client", - "rpcTimeoutMs": 30000 - } - } - } -} -"""; - -graft.nuget.PhysicsCalculator.GraftConfig.SetConfig(configSource); +```powershell +./gg .\PhysicsCalculator.dll --config .\pluginConfig.json ``` ## Configuration reference | Field | Required | Description | |-------|----------|-------------| -| `brokers` | yes* | Kafka bootstrap servers (`host:port[,host:port…]`). | -| `host` | * | Alias for `brokers` (compatibility with other plugins). | -| `requestTopic` | yes | Topic requests are produced to / consumed from. | -| `queue` | | Alias for `requestTopic`. | -| `replyTopic` | yes (client) | Topic replies are consumed from / default produce target. | -| `replyQueue` | | Alias for `replyTopic`. | -| `groupId` | no | Consumer group. Server default `graft-gateway`. Client default `graft-client` (a unique instance suffix is always appended on the client). | -| `rpcTimeoutMs` | no | Client request/response timeout in milliseconds (default `30000`). | -| `securityProtocol` | no | librdkafka `security.protocol` (e.g. `SASL_SSL`, `SSL`, `PLAINTEXT`). | -| `saslMechanism` | no | e.g. `PLAIN`, `SCRAM-SHA-256`, `SCRAM-SHA-512`. | -| `saslUsername` | no | SASL username. | -| `saslPassword` | no | SASL password. | -| `sslCaLocation` | no | Path to CA PEM for SSL verification. | - -\* Provide `brokers` or `host`. - -## Smoke test - -```bash -cmake -S . -B build -DCMAKE_BUILD_TYPE=Release -cmake --build build --config Release -ctest --test-dir build --output-on-failure -``` - -`KafkaPluginTest` links the shared library and exercises -`CreateServer` / `CreateTransportChannel` without requiring a broker. -End-to-end RPC still needs a running Kafka/Redpanda and the topics above. - -## Layout (drop-in as `kafka/`) - -``` -kafka/ - CMakeLists.txt - vcpkg.json - pluginConfig.json - docker-compose.yml - docker-compose.redpanda.yml - Readme.md - BUILD_NOTES.md - GraftcodePluginsInterfaces/ - KafkaPlugin/ - KafkaPluginTest/ - scripts/create-topics.sh -``` +| `brokers` / `host` | yes | Kafka bootstrap servers | +| `requestTopic` / `queue` | yes | Topic for requests | +| `replyTopic` / `replyQueue` | yes (RPC client) | Topic for replies | +| `groupId` | no | Consumer group base name | +| `rpcTimeoutMs` | no | Request/response timeout (default 30000) | diff --git a/kafka/vcpkg.json b/kafka/vcpkg.json deleted file mode 100644 index 9aadb2b..0000000 --- a/kafka/vcpkg.json +++ /dev/null @@ -1,9 +0,0 @@ -{ - "name": "graftcode-kafka-plugin", - "version": "1.0.0", - "description": "Graftcode Kafka transport/server plugin built on librdkafka (rdkafka++).", - "dependencies": [ - "librdkafka", - "nlohmann-json" - ] -} diff --git a/servicebus/Readme.md b/servicebus/Readme.md index 2b2311d..e09d69d 100644 --- a/servicebus/Readme.md +++ b/servicebus/Readme.md @@ -7,35 +7,27 @@ factory symbols (`CreateTransportChannel` / `DestroyTransportChannel` and `CreateServer` / `DestroyServer`). It is written in C++ and talks to Azure Service Bus over its native AMQP 1.0 protocol using -the Azure SDK for C++ AMQP library (`azure-core-amqp`), acquired through vcpkg. +the Azure SDK for C++ AMQP library (`azure-core-amqp`), acquired through CMake `FetchContent` +(same pattern as the RabbitMQ plugin). ## 1) Clone repository ```bash -git clone https://github.com/grft-dev/graftcode-plugins.git -cd graftcode-plugins/servicebus +git clone https://github.com/grft-dev/graftcode-extensions.git +cd graftcode-extensions/servicebus ``` -## 2) Get vcpkg - -The plugin depends on `azure-core-amqp-cpp` and `nlohmann-json`, declared in `vcpkg.json`. +## 2) Configure with CMake ```bash -git clone https://github.com/microsoft/vcpkg.git -./vcpkg/bootstrap-vcpkg.sh # on Windows: .\vcpkg\bootstrap-vcpkg.bat +cmake -S . -B build -DCMAKE_BUILD_TYPE=Release ``` -## 3) Configure with CMake +CMake downloads **nlohmann/json** and **Azure SDK for C++** (`azure-core` + +`azure-core-amqp`) via `FetchContent`. No vcpkg toolchain is required. First +configure needs network access for those downloads. -Point CMake at the vcpkg toolchain so the dependencies are installed and discovered -automatically (manifest mode): - -```bash -cmake -S . -B build -DCMAKE_BUILD_TYPE=Release \ - -DCMAKE_TOOLCHAIN_FILE=./vcpkg/scripts/buildsystems/vcpkg.cmake -``` - -## 4) Build +## 3) Build ```bash cmake --build build --config Release @@ -47,12 +39,11 @@ As a result, you will receive: If the generated library is `libServiceBusPlugin.*`, use plugin name: `libServiceBusPlugin`. -## 5) Download GG +## 4) Download GG Download `gg` from: - https://github.com/grft-dev/graftcode-gateway/releases/ - -## 6) Create the queues +## 5) Create the queues In the Azure portal (or via Azure CLI) create two queues in your Service Bus namespace, for example `myqueue` (requests) and `myqueue.reply` (responses). The **reply queue must be @@ -84,7 +75,7 @@ connection string (or set `"useDevelopmentEmulator": true` in config) and point emulator endpoint. See the official emulator: https://learn.microsoft.com/azure/service-bus-messaging/test-locally-with-service-bus-emulator -## 7) Run GG with sample library +## 6) Run GG with sample library In your sample folder, create `pluginConfig.json`: @@ -110,7 +101,7 @@ Then run: ./gg .\PhysicsCalculator.dll --config .\pluginConfig.json ``` -## 8) Get installation command +## 7) Get installation command Visit `http://localhost:81/GV`, select your package manager, and copy the generated installation command. @@ -122,7 +113,7 @@ dotnet new console dotnet add package -s https://grft.dev/019cf6aa-e2e0-74e7-a2b0-be30db97ccb5__graftcode graft.nuget.physicscalculator --version 1.0.0 ``` -## 9) Configure Graft after installation +## 8) Configure Graft after installation Use this configuration: @@ -216,3 +207,4 @@ Create the topic and subscription (the subscription does **not** need sessions): az servicebus topic create --resource-group --namespace-name --name mytopic az servicebus topic subscription create --resource-group --namespace-name --topic-name mytopic --name mysubscription ``` + diff --git a/servicebus/ServiceBusPlugin/CMakeLists.txt b/servicebus/ServiceBusPlugin/CMakeLists.txt index b6be69d..3482879 100644 --- a/servicebus/ServiceBusPlugin/CMakeLists.txt +++ b/servicebus/ServiceBusPlugin/CMakeLists.txt @@ -1,25 +1,59 @@ -# Graftcode Azure Service Bus plugin (AMQP 1.0 via the Azure SDK for C++). +# Graftcode Azure Service Bus plugin (AMQP 1.0 via Azure SDK for C++). set(target_name ServiceBusPlugin) add_library(${target_name} SHARED - "TransportServiceBus.cpp" - "ServiceBusClient.cpp" - "ServiceBusServer.cpp" + "TransportServiceBus.cpp" + "ServiceBusClient.cpp" + "ServiceBusServer.cpp" ) target_include_directories(${target_name} PUBLIC "${CMAKE_SOURCE_DIR}/GraftcodePluginsInterfaces" ) -# Both dependencies are provided by vcpkg (see vcpkg.json). Configure with the vcpkg -# toolchain file so find_package can locate them, e.g.: -# cmake -S . -B build -DCMAKE_TOOLCHAIN_FILE=/scripts/buildsystems/vcpkg.cmake -find_package(azure-core-amqp-cpp CONFIG REQUIRED) -find_package(nlohmann_json CONFIG REQUIRED) +include(FetchContent) -# The vcpkg azure-core-amqp-cpp port is built with the uAMQP backend but does not propagate -# the backend selection macro to consumers, so the public uAMQP API (CreateMessageSender, -# MessageSender::Send, etc.) is compiled out unless we define it here. +FetchContent_Declare( + json + GIT_REPOSITORY https://github.com/nlohmann/json.git + GIT_TAG v3.12.0 + GIT_SHALLOW TRUE +) +FetchContent_MakeAvailable(json) + +# Pull Azure SDK sources via CMake (no vcpkg). FETCH_SOURCE_DEPS makes the +# SDK fetch its own C deps; AZURE_SDK_DISABLE_AUTO_VCPKG blocks vcpkg bootstrap. +set(ENV{AZURE_SDK_DISABLE_AUTO_VCPKG} ON) +set(AZURE_SDK_DISABLE_AUTO_VCPKG ON CACHE BOOL "" FORCE) +set(FETCH_SOURCE_DEPS ON CACHE BOOL "" FORCE) +set(BUILD_TESTING OFF CACHE BOOL "" FORCE) +set(BUILD_SAMPLES OFF CACHE BOOL "" FORCE) +set(BUILD_TRANSPORT_CURL ON CACHE BOOL "" FORCE) +set(WARNINGS_AS_ERRORS OFF CACHE BOOL "" FORCE) +set(DISABLE_AMQP OFF CACHE BOOL "" FORCE) + +FetchContent_Declare( + azure-sdk-for-cpp + GIT_REPOSITORY https://github.com/Azure/azure-sdk-for-cpp.git + GIT_TAG azure-core-amqp_1.0.0-beta.11 + GIT_SHALLOW TRUE +) +FetchContent_GetProperties(azure-sdk-for-cpp) +if(NOT azure-sdk-for-cpp_POPULATED) + FetchContent_Populate(azure-sdk-for-cpp) + add_subdirectory( + ${azure-sdk-for-cpp_SOURCE_DIR}/sdk/core/azure-core + ${azure-sdk-for-cpp_BINARY_DIR}/azure-core + EXCLUDE_FROM_ALL + ) + add_subdirectory( + ${azure-sdk-for-cpp_SOURCE_DIR}/sdk/core/azure-core-amqp + ${azure-sdk-for-cpp_BINARY_DIR}/azure-core-amqp + EXCLUDE_FROM_ALL + ) +endif() + +# uAMQP backend APIs used by this plugin (CreateMessageSender, MessageSender::Send, ...) target_compile_definitions(${target_name} PRIVATE ENABLE_UAMQP=1) target_link_libraries(${target_name} PUBLIC diff --git a/servicebus/vcpkg.json b/servicebus/vcpkg.json deleted file mode 100644 index 7e8cbe2..0000000 --- a/servicebus/vcpkg.json +++ /dev/null @@ -1,9 +0,0 @@ -{ - "name": "graftcode-servicebus-plugin", - "version": "1.0.0", - "description": "Graftcode Azure Service Bus transport/server plugin built on the Azure SDK for C++ AMQP library.", - "dependencies": [ - "azure-core-amqp-cpp", - "nlohmann-json" - ] -} From 01cd0845681ef9070004047c48cb736926889559 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Komor?= Date: Wed, 16 Sep 2026 10:49:42 +0200 Subject: [PATCH 04/10] Add kafka to release workflow and sample configs Include kafka in the shared plugin release matrix (FetchContent build, no vcpkg). Add samples/kafka with docker-compose (broker+UI) and gateway/client plugin config files. --- .github/workflows/release.yml | 65 ++++++++++++-------- samples/kafka/README.md | 29 +++++++++ samples/kafka/docker-compose.redpanda.yml | 10 +++ samples/kafka/docker-compose.yml | 32 ++++++++++ samples/kafka/graftConfig.kafka.example.json | 16 +++++ samples/kafka/pluginConfig.client.json | 8 +++ samples/kafka/pluginConfig.gateway.json | 8 +++ samples/kafka/pluginConfig.json | 8 +++ samples/kafka/scripts/create-topics.sh | 53 ++++++++++++++++ 9 files changed, 205 insertions(+), 24 deletions(-) create mode 100644 samples/kafka/README.md create mode 100644 samples/kafka/docker-compose.redpanda.yml create mode 100644 samples/kafka/docker-compose.yml create mode 100644 samples/kafka/graftConfig.kafka.example.json create mode 100644 samples/kafka/pluginConfig.client.json create mode 100644 samples/kafka/pluginConfig.gateway.json create mode 100644 samples/kafka/pluginConfig.json create mode 100644 samples/kafka/scripts/create-topics.sh diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 96ce8b7..4bac5a8 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -8,12 +8,14 @@ on: paths: - "rabbitmq/**" - "servicebus/**" + - "kafka/**" - ".github/workflows/release.yml" pull_request: branches: [ "main" ] paths: - "rabbitmq/**" - "servicebus/**" + - "kafka/**" - ".github/workflows/release.yml" workflow_dispatch: inputs: @@ -34,7 +36,7 @@ jobs: strategy: fail-fast: false matrix: - plugin: [rabbitmq, servicebus] + plugin: [rabbitmq, servicebus, kafka] platform: [windows-latest, windows-11-arm, ubuntu-22.04, ubuntu-22.04-arm, macos-26-intel, macos-14] steps: @@ -76,29 +78,44 @@ jobs: ;; esac - if [ "${{ matrix.plugin }}" = "servicebus" ]; then - echo "DIRECTORY=servicebus" >> "$GITHUB_ENV" - echo "ARTIFACT_PATH=servicebus/build/ServiceBusPlugin" >> "$GITHUB_ENV" - echo "CONFIGURE_ARGS=-DCMAKE_TOOLCHAIN_FILE=${VCPKG_INSTALLATION_ROOT}/scripts/buildsystems/vcpkg.cmake" >> "$GITHUB_ENV" - if [ "${{ runner.os }}" = "Linux" ]; then - echo "INSTALL_PACKAGES=build-essential cmake git ninja-build pkg-config" >> "$GITHUB_ENV" - elif [ "${{ runner.os }}" = "macOS" ]; then - echo "INSTALL_PACKAGES=cmake ninja pkg-config" >> "$GITHUB_ENV" - else - echo "INSTALL_PACKAGES=cmake" >> "$GITHUB_ENV" - fi - else - echo "DIRECTORY=rabbitmq" >> "$GITHUB_ENV" - echo "ARTIFACT_PATH=rabbitmq/build/RabbitmqPlugin" >> "$GITHUB_ENV" - echo "CONFIGURE_ARGS=" >> "$GITHUB_ENV" - if [ "${{ runner.os }}" = "Linux" ]; then - echo "INSTALL_PACKAGES=build-essential cmake git" >> "$GITHUB_ENV" - elif [ "${{ runner.os }}" = "macOS" ]; then - echo "INSTALL_PACKAGES=cmake" >> "$GITHUB_ENV" - else - echo "INSTALL_PACKAGES=cmake" >> "$GITHUB_ENV" - fi - fi + case "${{ matrix.plugin }}" in + servicebus) + echo "DIRECTORY=servicebus" >> "$GITHUB_ENV" + echo "ARTIFACT_PATH=servicebus/build/ServiceBusPlugin" >> "$GITHUB_ENV" + echo "CONFIGURE_ARGS=" >> "$GITHUB_ENV" + if [ "${{ runner.os }}" = "Linux" ]; then + echo "INSTALL_PACKAGES=build-essential cmake git ninja-build pkg-config libcurl4-openssl-dev libssl-dev" >> "$GITHUB_ENV" + elif [ "${{ runner.os }}" = "macOS" ]; then + echo "INSTALL_PACKAGES=cmake ninja pkg-config" >> "$GITHUB_ENV" + else + echo "INSTALL_PACKAGES=cmake" >> "$GITHUB_ENV" + fi + ;; + kafka) + echo "DIRECTORY=kafka" >> "$GITHUB_ENV" + echo "ARTIFACT_PATH=kafka/build/KafkaPlugin" >> "$GITHUB_ENV" + echo "CONFIGURE_ARGS=" >> "$GITHUB_ENV" + if [ "${{ runner.os }}" = "Linux" ]; then + echo "INSTALL_PACKAGES=build-essential cmake git libssl-dev libsasl2-dev zlib1g-dev" >> "$GITHUB_ENV" + elif [ "${{ runner.os }}" = "macOS" ]; then + echo "INSTALL_PACKAGES=cmake openssl" >> "$GITHUB_ENV" + else + echo "INSTALL_PACKAGES=cmake" >> "$GITHUB_ENV" + fi + ;; + *) + echo "DIRECTORY=rabbitmq" >> "$GITHUB_ENV" + echo "ARTIFACT_PATH=rabbitmq/build/RabbitmqPlugin" >> "$GITHUB_ENV" + echo "CONFIGURE_ARGS=" >> "$GITHUB_ENV" + if [ "${{ runner.os }}" = "Linux" ]; then + echo "INSTALL_PACKAGES=build-essential cmake git" >> "$GITHUB_ENV" + elif [ "${{ runner.os }}" = "macOS" ]; then + echo "INSTALL_PACKAGES=cmake" >> "$GITHUB_ENV" + else + echo "INSTALL_PACKAGES=cmake" >> "$GITHUB_ENV" + fi + ;; + esac echo "PLUGIN_NAME=${{ matrix.plugin }}" >> "$GITHUB_ENV" diff --git a/samples/kafka/README.md b/samples/kafka/README.md new file mode 100644 index 0000000..e7a0106 --- /dev/null +++ b/samples/kafka/README.md @@ -0,0 +1,29 @@ +# Kafka sample configs + +Local broker + UI and Graftcode plugin connection files. + +## Start Kafka + +```bash +cd samples/kafka +docker compose up -d +``` + +- Broker: `localhost:9092` +- UI: http://localhost:8080 + +Create topics (if needed): + +```bash +docker exec graftcode-kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --if-not-exists --topic graft.requests --partitions 1 --replication-factor 1 +docker exec graftcode-kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --if-not-exists --topic graft.replies --partitions 1 --replication-factor 1 +``` + +## Config files + +| File | Use | +|------|-----| +| `pluginConfig.gateway.json` | `gg YourModule.dll --config pluginConfig.gateway.json` | +| `pluginConfig.client.json` | Client-side plugin block | +| `graftConfig.kafka.example.json` | Full GraftConfig example for .NET | +| `pluginConfig.json` | Same as gateway (default name) | diff --git a/samples/kafka/docker-compose.redpanda.yml b/samples/kafka/docker-compose.redpanda.yml new file mode 100644 index 0000000..eff7477 --- /dev/null +++ b/samples/kafka/docker-compose.redpanda.yml @@ -0,0 +1,10 @@ +services: + redpanda: + image: redpandadata/redpanda:v24.2.4 + command: + - redpanda start + - --kafka-addr internal://0.0.0.0:9092,external://0.0.0.0:19092 + - --advertise-kafka-addr internal://redpanda:9092,external://localhost:19092 + - --mode dev-container + ports: + - "19092:19092" diff --git a/samples/kafka/docker-compose.yml b/samples/kafka/docker-compose.yml new file mode 100644 index 0000000..4e4f838 --- /dev/null +++ b/samples/kafka/docker-compose.yml @@ -0,0 +1,32 @@ +# Lokalny Kafka (KRaft) + UI — testy Graftcode KafkaPlugin. +services: + kafka: + image: apache/kafka:3.8.1 + container_name: graftcode-kafka + ports: + - "9092:9092" + environment: + KAFKA_NODE_ID: 1 + KAFKA_PROCESS_ROLES: broker,controller + KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093,PLAINTEXT_INTERNAL://0.0.0.0:9094 + KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092,PLAINTEXT_INTERNAL://kafka:9094 + KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_INTERNAL:PLAINTEXT + KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT_INTERNAL + KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER + KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093 + KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 + KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 + KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 + KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0 + CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk + + kafka-ui: + image: provectuslabs/kafka-ui:v0.7.2 + container_name: graftcode-kafka-ui + depends_on: + - kafka + ports: + - "8080:8080" + environment: + KAFKA_CLUSTERS_0_NAME: local + KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9094 diff --git a/samples/kafka/graftConfig.kafka.example.json b/samples/kafka/graftConfig.kafka.example.json new file mode 100644 index 0000000..120c636 --- /dev/null +++ b/samples/kafka/graftConfig.kafka.example.json @@ -0,0 +1,16 @@ +{ + "configurations": { + "graft.nuget.PhysicsCalculator": { + "runtime": "netcore", + "stateless": true, + "plugin": { + "name": "KafkaPlugin", + "brokers": "localhost:9092", + "requestTopic": "graft.requests", + "replyTopic": "graft.replies", + "groupId": "graft-client", + "rpcTimeoutMs": 30000 + } + } + } +} diff --git a/samples/kafka/pluginConfig.client.json b/samples/kafka/pluginConfig.client.json new file mode 100644 index 0000000..8cac248 --- /dev/null +++ b/samples/kafka/pluginConfig.client.json @@ -0,0 +1,8 @@ +{ + "name": "KafkaPlugin", + "brokers": "localhost:9092", + "requestTopic": "graft.requests", + "replyTopic": "graft.replies", + "groupId": "graft-client", + "rpcTimeoutMs": 30000 +} diff --git a/samples/kafka/pluginConfig.gateway.json b/samples/kafka/pluginConfig.gateway.json new file mode 100644 index 0000000..e607326 --- /dev/null +++ b/samples/kafka/pluginConfig.gateway.json @@ -0,0 +1,8 @@ +{ + "name": "KafkaPlugin", + "brokers": "localhost:9092", + "requestTopic": "graft.requests", + "replyTopic": "graft.replies", + "groupId": "graft-gateway", + "rpcTimeoutMs": 30000 +} diff --git a/samples/kafka/pluginConfig.json b/samples/kafka/pluginConfig.json new file mode 100644 index 0000000..e607326 --- /dev/null +++ b/samples/kafka/pluginConfig.json @@ -0,0 +1,8 @@ +{ + "name": "KafkaPlugin", + "brokers": "localhost:9092", + "requestTopic": "graft.requests", + "replyTopic": "graft.replies", + "groupId": "graft-gateway", + "rpcTimeoutMs": 30000 +} diff --git a/samples/kafka/scripts/create-topics.sh b/samples/kafka/scripts/create-topics.sh new file mode 100644 index 0000000..efbb103 --- /dev/null +++ b/samples/kafka/scripts/create-topics.sh @@ -0,0 +1,53 @@ +#!/usr/bin/env bash +# Create the default Graftcode Kafka request/reply topics. +set -euo pipefail + +BROKERS="${BROKERS:-localhost:9092}" +REQUEST_TOPIC="${REQUEST_TOPIC:-graft.requests}" +REPLY_TOPIC="${REPLY_TOPIC:-graft.replies}" +CONTAINER="${KAFKA_CONTAINER:-}" + +create_with_kafka_topics() { + local bin="$1" + "$bin" --bootstrap-server "$BROKERS" --create --if-not-exists --topic "$REQUEST_TOPIC" --partitions 1 --replication-factor 1 + "$bin" --bootstrap-server "$BROKERS" --create --if-not-exists --topic "$REPLY_TOPIC" --partitions 1 --replication-factor 1 +} + +if [[ -n "$CONTAINER" ]]; then + echo "Creating topics via docker exec on container '$CONTAINER' (bootstrap $BROKERS)..." + # apache/kafka image path + if docker exec "$CONTAINER" test -x /opt/kafka/bin/kafka-topics.sh; then + docker exec "$CONTAINER" /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 \ + --create --if-not-exists --topic "$REQUEST_TOPIC" --partitions 1 --replication-factor 1 + docker exec "$CONTAINER" /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 \ + --create --if-not-exists --topic "$REPLY_TOPIC" --partitions 1 --replication-factor 1 + # bitnami / other layouts + elif docker exec "$CONTAINER" sh -c 'command -v kafka-topics.sh' >/dev/null 2>&1; then + docker exec "$CONTAINER" kafka-topics.sh --bootstrap-server localhost:9092 \ + --create --if-not-exists --topic "$REQUEST_TOPIC" --partitions 1 --replication-factor 1 + docker exec "$CONTAINER" kafka-topics.sh --bootstrap-server localhost:9092 \ + --create --if-not-exists --topic "$REPLY_TOPIC" --partitions 1 --replication-factor 1 + else + echo "Could not find kafka-topics.sh inside $CONTAINER" >&2 + exit 1 + fi +elif command -v kafka-topics.sh >/dev/null 2>&1; then + create_with_kafka_topics kafka-topics.sh +elif command -v rpk >/dev/null 2>&1; then + echo "Creating topics via rpk (brokers $BROKERS)..." + rpk topic create "$REQUEST_TOPIC" -X brokers="$BROKERS" || true + rpk topic create "$REPLY_TOPIC" -X brokers="$BROKERS" || true +else + # Auto-detect a running compose kafka container + CONTAINER="$(docker ps --format '{{.Names}}' | grep -E 'kafka|redpanda' | head -n1 || true)" + if [[ -z "$CONTAINER" ]]; then + echo "No kafka-topics.sh/rpk on PATH and no kafka/redpanda container running." >&2 + echo "Start the broker first: docker compose up -d" >&2 + echo "Or set KAFKA_CONTAINER=." >&2 + exit 1 + fi + KAFKA_CONTAINER="$CONTAINER" "$0" + exit $? +fi + +echo "Topics ready: $REQUEST_TOPIC, $REPLY_TOPIC" From 0eb1df23393bad3536fa8166dbeadecdb9285816 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Komor?= Date: Wed, 16 Sep 2026 10:53:24 +0200 Subject: [PATCH 05/10] Enhance Kafka setup; add Kafka UI and config examples Updated docker-compose.yml to improve local Kafka setup: added Kafka UI (provectuslabs/kafka-ui) on port 8080, improved listener configuration, updated environment variables, and renamed Kafka container to graftcode-kafka. Added three example JSON config files for KafkaPlugin usage in .NET Core plugins, client, and gateway components. --- kafka/docker-compose.yml | 23 ++++++++++++++++++----- kafka/graftConfig.kafka.example.json | 16 ++++++++++++++++ kafka/pluginConfig.client.json | 8 ++++++++ kafka/pluginConfig.gateway.json | 8 ++++++++ 4 files changed, 50 insertions(+), 5 deletions(-) create mode 100644 kafka/graftConfig.kafka.example.json create mode 100644 kafka/pluginConfig.client.json create mode 100644 kafka/pluginConfig.gateway.json diff --git a/kafka/docker-compose.yml b/kafka/docker-compose.yml index d892980..4e4f838 100644 --- a/kafka/docker-compose.yml +++ b/kafka/docker-compose.yml @@ -1,19 +1,32 @@ -# Lokalny Kafka (KRaft, bez ZooKeeper) — wystarczy do testów pluginu. +# Lokalny Kafka (KRaft) + UI — testy Graftcode KafkaPlugin. services: kafka: image: apache/kafka:3.8.1 + container_name: graftcode-kafka ports: - "9092:9092" environment: KAFKA_NODE_ID: 1 KAFKA_PROCESS_ROLES: broker,controller - KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093 - KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092 + KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093,PLAINTEXT_INTERNAL://0.0.0.0:9094 + KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092,PLAINTEXT_INTERNAL://kafka:9094 + KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_INTERNAL:PLAINTEXT + KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT_INTERNAL KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT - KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093 + KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093 KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0 CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk + + kafka-ui: + image: provectuslabs/kafka-ui:v0.7.2 + container_name: graftcode-kafka-ui + depends_on: + - kafka + ports: + - "8080:8080" + environment: + KAFKA_CLUSTERS_0_NAME: local + KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9094 diff --git a/kafka/graftConfig.kafka.example.json b/kafka/graftConfig.kafka.example.json new file mode 100644 index 0000000..120c636 --- /dev/null +++ b/kafka/graftConfig.kafka.example.json @@ -0,0 +1,16 @@ +{ + "configurations": { + "graft.nuget.PhysicsCalculator": { + "runtime": "netcore", + "stateless": true, + "plugin": { + "name": "KafkaPlugin", + "brokers": "localhost:9092", + "requestTopic": "graft.requests", + "replyTopic": "graft.replies", + "groupId": "graft-client", + "rpcTimeoutMs": 30000 + } + } + } +} diff --git a/kafka/pluginConfig.client.json b/kafka/pluginConfig.client.json new file mode 100644 index 0000000..8cac248 --- /dev/null +++ b/kafka/pluginConfig.client.json @@ -0,0 +1,8 @@ +{ + "name": "KafkaPlugin", + "brokers": "localhost:9092", + "requestTopic": "graft.requests", + "replyTopic": "graft.replies", + "groupId": "graft-client", + "rpcTimeoutMs": 30000 +} diff --git a/kafka/pluginConfig.gateway.json b/kafka/pluginConfig.gateway.json new file mode 100644 index 0000000..e607326 --- /dev/null +++ b/kafka/pluginConfig.gateway.json @@ -0,0 +1,8 @@ +{ + "name": "KafkaPlugin", + "brokers": "localhost:9092", + "requestTopic": "graft.requests", + "replyTopic": "graft.replies", + "groupId": "graft-gateway", + "rpcTimeoutMs": 30000 +} From 49f7c9c606045f79d39a2375539a87532e5999db Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Komor?= Date: Wed, 16 Sep 2026 10:54:31 +0200 Subject: [PATCH 06/10] fix --- kafka/docker-compose.redpanda.yml | 10 ------ kafka/docker-compose.yml | 32 ----------------- kafka/graftConfig.kafka.example.json | 16 --------- kafka/pluginConfig.client.json | 8 ----- kafka/pluginConfig.gateway.json | 8 ----- kafka/pluginConfig.json | 8 ----- kafka/scripts/create-topics.sh | 53 ---------------------------- 7 files changed, 135 deletions(-) delete mode 100644 kafka/docker-compose.redpanda.yml delete mode 100644 kafka/docker-compose.yml delete mode 100644 kafka/graftConfig.kafka.example.json delete mode 100644 kafka/pluginConfig.client.json delete mode 100644 kafka/pluginConfig.gateway.json delete mode 100644 kafka/pluginConfig.json delete mode 100644 kafka/scripts/create-topics.sh diff --git a/kafka/docker-compose.redpanda.yml b/kafka/docker-compose.redpanda.yml deleted file mode 100644 index eff7477..0000000 --- a/kafka/docker-compose.redpanda.yml +++ /dev/null @@ -1,10 +0,0 @@ -services: - redpanda: - image: redpandadata/redpanda:v24.2.4 - command: - - redpanda start - - --kafka-addr internal://0.0.0.0:9092,external://0.0.0.0:19092 - - --advertise-kafka-addr internal://redpanda:9092,external://localhost:19092 - - --mode dev-container - ports: - - "19092:19092" diff --git a/kafka/docker-compose.yml b/kafka/docker-compose.yml deleted file mode 100644 index 4e4f838..0000000 --- a/kafka/docker-compose.yml +++ /dev/null @@ -1,32 +0,0 @@ -# Lokalny Kafka (KRaft) + UI — testy Graftcode KafkaPlugin. -services: - kafka: - image: apache/kafka:3.8.1 - container_name: graftcode-kafka - ports: - - "9092:9092" - environment: - KAFKA_NODE_ID: 1 - KAFKA_PROCESS_ROLES: broker,controller - KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093,PLAINTEXT_INTERNAL://0.0.0.0:9094 - KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092,PLAINTEXT_INTERNAL://kafka:9094 - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_INTERNAL:PLAINTEXT - KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT_INTERNAL - KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER - KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093 - KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 - KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 - KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 - KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0 - CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk - - kafka-ui: - image: provectuslabs/kafka-ui:v0.7.2 - container_name: graftcode-kafka-ui - depends_on: - - kafka - ports: - - "8080:8080" - environment: - KAFKA_CLUSTERS_0_NAME: local - KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9094 diff --git a/kafka/graftConfig.kafka.example.json b/kafka/graftConfig.kafka.example.json deleted file mode 100644 index 120c636..0000000 --- a/kafka/graftConfig.kafka.example.json +++ /dev/null @@ -1,16 +0,0 @@ -{ - "configurations": { - "graft.nuget.PhysicsCalculator": { - "runtime": "netcore", - "stateless": true, - "plugin": { - "name": "KafkaPlugin", - "brokers": "localhost:9092", - "requestTopic": "graft.requests", - "replyTopic": "graft.replies", - "groupId": "graft-client", - "rpcTimeoutMs": 30000 - } - } - } -} diff --git a/kafka/pluginConfig.client.json b/kafka/pluginConfig.client.json deleted file mode 100644 index 8cac248..0000000 --- a/kafka/pluginConfig.client.json +++ /dev/null @@ -1,8 +0,0 @@ -{ - "name": "KafkaPlugin", - "brokers": "localhost:9092", - "requestTopic": "graft.requests", - "replyTopic": "graft.replies", - "groupId": "graft-client", - "rpcTimeoutMs": 30000 -} diff --git a/kafka/pluginConfig.gateway.json b/kafka/pluginConfig.gateway.json deleted file mode 100644 index e607326..0000000 --- a/kafka/pluginConfig.gateway.json +++ /dev/null @@ -1,8 +0,0 @@ -{ - "name": "KafkaPlugin", - "brokers": "localhost:9092", - "requestTopic": "graft.requests", - "replyTopic": "graft.replies", - "groupId": "graft-gateway", - "rpcTimeoutMs": 30000 -} diff --git a/kafka/pluginConfig.json b/kafka/pluginConfig.json deleted file mode 100644 index e607326..0000000 --- a/kafka/pluginConfig.json +++ /dev/null @@ -1,8 +0,0 @@ -{ - "name": "KafkaPlugin", - "brokers": "localhost:9092", - "requestTopic": "graft.requests", - "replyTopic": "graft.replies", - "groupId": "graft-gateway", - "rpcTimeoutMs": 30000 -} diff --git a/kafka/scripts/create-topics.sh b/kafka/scripts/create-topics.sh deleted file mode 100644 index efbb103..0000000 --- a/kafka/scripts/create-topics.sh +++ /dev/null @@ -1,53 +0,0 @@ -#!/usr/bin/env bash -# Create the default Graftcode Kafka request/reply topics. -set -euo pipefail - -BROKERS="${BROKERS:-localhost:9092}" -REQUEST_TOPIC="${REQUEST_TOPIC:-graft.requests}" -REPLY_TOPIC="${REPLY_TOPIC:-graft.replies}" -CONTAINER="${KAFKA_CONTAINER:-}" - -create_with_kafka_topics() { - local bin="$1" - "$bin" --bootstrap-server "$BROKERS" --create --if-not-exists --topic "$REQUEST_TOPIC" --partitions 1 --replication-factor 1 - "$bin" --bootstrap-server "$BROKERS" --create --if-not-exists --topic "$REPLY_TOPIC" --partitions 1 --replication-factor 1 -} - -if [[ -n "$CONTAINER" ]]; then - echo "Creating topics via docker exec on container '$CONTAINER' (bootstrap $BROKERS)..." - # apache/kafka image path - if docker exec "$CONTAINER" test -x /opt/kafka/bin/kafka-topics.sh; then - docker exec "$CONTAINER" /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 \ - --create --if-not-exists --topic "$REQUEST_TOPIC" --partitions 1 --replication-factor 1 - docker exec "$CONTAINER" /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 \ - --create --if-not-exists --topic "$REPLY_TOPIC" --partitions 1 --replication-factor 1 - # bitnami / other layouts - elif docker exec "$CONTAINER" sh -c 'command -v kafka-topics.sh' >/dev/null 2>&1; then - docker exec "$CONTAINER" kafka-topics.sh --bootstrap-server localhost:9092 \ - --create --if-not-exists --topic "$REQUEST_TOPIC" --partitions 1 --replication-factor 1 - docker exec "$CONTAINER" kafka-topics.sh --bootstrap-server localhost:9092 \ - --create --if-not-exists --topic "$REPLY_TOPIC" --partitions 1 --replication-factor 1 - else - echo "Could not find kafka-topics.sh inside $CONTAINER" >&2 - exit 1 - fi -elif command -v kafka-topics.sh >/dev/null 2>&1; then - create_with_kafka_topics kafka-topics.sh -elif command -v rpk >/dev/null 2>&1; then - echo "Creating topics via rpk (brokers $BROKERS)..." - rpk topic create "$REQUEST_TOPIC" -X brokers="$BROKERS" || true - rpk topic create "$REPLY_TOPIC" -X brokers="$BROKERS" || true -else - # Auto-detect a running compose kafka container - CONTAINER="$(docker ps --format '{{.Names}}' | grep -E 'kafka|redpanda' | head -n1 || true)" - if [[ -z "$CONTAINER" ]]; then - echo "No kafka-topics.sh/rpk on PATH and no kafka/redpanda container running." >&2 - echo "Start the broker first: docker compose up -d" >&2 - echo "Or set KAFKA_CONTAINER=." >&2 - exit 1 - fi - KAFKA_CONTAINER="$CONTAINER" "$0" - exit $? -fi - -echo "Topics ready: $REQUEST_TOPIC, $REPLY_TOPIC" From 82e94f291716b0dc0dd7ddf8c85f7f95f4a17e45 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Komor?= Date: Wed, 16 Sep 2026 10:59:36 +0200 Subject: [PATCH 07/10] fix(servicebus): use WinHTTP on Windows instead of CURL Azure SDK FetchContent build was failing configure with missing CURL. Enable BUILD_TRANSPORT_WINHTTP on Windows and CURL only on Unix. --- servicebus/Readme.md | 2 +- servicebus/ServiceBusPlugin/CMakeLists.txt | 10 +++++++++- 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/servicebus/Readme.md b/servicebus/Readme.md index e09d69d..f6bc823 100644 --- a/servicebus/Readme.md +++ b/servicebus/Readme.md @@ -25,7 +25,7 @@ cmake -S . -B build -DCMAKE_BUILD_TYPE=Release CMake downloads **nlohmann/json** and **Azure SDK for C++** (`azure-core` + `azure-core-amqp`) via `FetchContent`. No vcpkg toolchain is required. First -configure needs network access for those downloads. +configure needs network access for those downloads. On Windows the HTTP transport is WinHTTP (no libcurl); on Linux/macOS it uses libcurl. ## 3) Build diff --git a/servicebus/ServiceBusPlugin/CMakeLists.txt b/servicebus/ServiceBusPlugin/CMakeLists.txt index 3482879..b1d87a1 100644 --- a/servicebus/ServiceBusPlugin/CMakeLists.txt +++ b/servicebus/ServiceBusPlugin/CMakeLists.txt @@ -28,10 +28,18 @@ set(AZURE_SDK_DISABLE_AUTO_VCPKG ON CACHE BOOL "" FORCE) set(FETCH_SOURCE_DEPS ON CACHE BOOL "" FORCE) set(BUILD_TESTING OFF CACHE BOOL "" FORCE) set(BUILD_SAMPLES OFF CACHE BOOL "" FORCE) -set(BUILD_TRANSPORT_CURL ON CACHE BOOL "" FORCE) set(WARNINGS_AS_ERRORS OFF CACHE BOOL "" FORCE) set(DISABLE_AMQP OFF CACHE BOOL "" FORCE) +# Windows: WinHTTP (no libcurl). Unix: libcurl. +if(WIN32) + set(BUILD_TRANSPORT_WINHTTP ON CACHE BOOL "" FORCE) + set(BUILD_TRANSPORT_CURL OFF CACHE BOOL "" FORCE) +else() + set(BUILD_TRANSPORT_CURL ON CACHE BOOL "" FORCE) + set(BUILD_TRANSPORT_WINHTTP OFF CACHE BOOL "" FORCE) +endif() + FetchContent_Declare( azure-sdk-for-cpp GIT_REPOSITORY https://github.com/Azure/azure-sdk-for-cpp.git From c4d281817b65458556db0c855584fe37ec900a4d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Komor?= Date: Wed, 16 Sep 2026 11:06:33 +0200 Subject: [PATCH 08/10] fix --- servicebus/ServiceBusPlugin/CMakeLists.txt | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/servicebus/ServiceBusPlugin/CMakeLists.txt b/servicebus/ServiceBusPlugin/CMakeLists.txt index b1d87a1..f6bdc64 100644 --- a/servicebus/ServiceBusPlugin/CMakeLists.txt +++ b/servicebus/ServiceBusPlugin/CMakeLists.txt @@ -21,10 +21,7 @@ FetchContent_Declare( ) FetchContent_MakeAvailable(json) -# Pull Azure SDK sources via CMake (no vcpkg). FETCH_SOURCE_DEPS makes the -# SDK fetch its own C deps; AZURE_SDK_DISABLE_AUTO_VCPKG blocks vcpkg bootstrap. -set(ENV{AZURE_SDK_DISABLE_AUTO_VCPKG} ON) -set(AZURE_SDK_DISABLE_AUTO_VCPKG ON CACHE BOOL "" FORCE) +# Pull Azure SDK sources via CMake FetchContent. set(FETCH_SOURCE_DEPS ON CACHE BOOL "" FORCE) set(BUILD_TESTING OFF CACHE BOOL "" FORCE) set(BUILD_SAMPLES OFF CACHE BOOL "" FORCE) From 1dd63617349214ee5643a555172600456da70f8d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Komor?= Date: Wed, 16 Sep 2026 11:12:53 +0200 Subject: [PATCH 09/10] in progress --- servicebus/CMakeLists.txt | 2 +- servicebus/ServiceBusPlugin/CMakeLists.txt | 69 +++++++++++++--------- 2 files changed, 42 insertions(+), 29 deletions(-) diff --git a/servicebus/CMakeLists.txt b/servicebus/CMakeLists.txt index b07e32e..25f1c2f 100644 --- a/servicebus/CMakeLists.txt +++ b/servicebus/CMakeLists.txt @@ -1,4 +1,4 @@ -set(CMAKE_MIN 3.22) +set(CMAKE_MIN 3.25) cmake_minimum_required (VERSION ${CMAKE_MIN}) set(CMAKE_POLICY_VERSION_MINIMUM ${CMAKE_MIN}) cmake_policy(VERSION ${CMAKE_MIN}) diff --git a/servicebus/ServiceBusPlugin/CMakeLists.txt b/servicebus/ServiceBusPlugin/CMakeLists.txt index f6bdc64..7c4d7a7 100644 --- a/servicebus/ServiceBusPlugin/CMakeLists.txt +++ b/servicebus/ServiceBusPlugin/CMakeLists.txt @@ -1,10 +1,12 @@ # Graftcode Azure Service Bus plugin (AMQP 1.0 via Azure SDK for C++). +# Standalone project — same idea as rabbitmq/: FetchContent each dependency, +# then link. azure-core and azure-core-amqp are separate FetchContent projects. set(target_name ServiceBusPlugin) add_library(${target_name} SHARED - "TransportServiceBus.cpp" - "ServiceBusClient.cpp" - "ServiceBusServer.cpp" + TransportServiceBus.cpp + ServiceBusClient.cpp + ServiceBusServer.cpp ) target_include_directories(${target_name} PUBLIC @@ -13,6 +15,9 @@ target_include_directories(${target_name} PUBLIC include(FetchContent) +# --------------------------------------------------------------------------- +# nlohmann/json +# --------------------------------------------------------------------------- FetchContent_Declare( json GIT_REPOSITORY https://github.com/nlohmann/json.git @@ -21,44 +26,52 @@ FetchContent_Declare( ) FetchContent_MakeAvailable(json) -# Pull Azure SDK sources via CMake FetchContent. +# --------------------------------------------------------------------------- +# Azure SDK build switches (apply before FetchContent of azure-* targets) +# --------------------------------------------------------------------------- set(FETCH_SOURCE_DEPS ON CACHE BOOL "" FORCE) set(BUILD_TESTING OFF CACHE BOOL "" FORCE) set(BUILD_SAMPLES OFF CACHE BOOL "" FORCE) set(WARNINGS_AS_ERRORS OFF CACHE BOOL "" FORCE) set(DISABLE_AMQP OFF CACHE BOOL "" FORCE) +set(BUILD_TRANSPORT_CURL ON CACHE BOOL "" FORCE) +set(BUILD_TRANSPORT_WINHTTP ON CACHE BOOL "" FORCE) -# Windows: WinHTTP (no libcurl). Unix: libcurl. -if(WIN32) - set(BUILD_TRANSPORT_WINHTTP ON CACHE BOOL "" FORCE) - set(BUILD_TRANSPORT_CURL OFF CACHE BOOL "" FORCE) -else() - set(BUILD_TRANSPORT_CURL ON CACHE BOOL "" FORCE) - set(BUILD_TRANSPORT_WINHTTP OFF CACHE BOOL "" FORCE) -endif() +# --------------------------------------------------------------------------- +# azure-core (package name consumed by azure-core-amqp: azure-core-cpp) +# Requires CMake >= 3.25 for SOURCE_SUBDIR. +# --------------------------------------------------------------------------- +FetchContent_Declare( + azure-core + GIT_REPOSITORY https://github.com/Azure/azure-sdk-for-cpp.git + GIT_TAG azure-core-amqp_1.0.0-beta.11 + GIT_SHALLOW TRUE + SOURCE_SUBDIR sdk/core/azure-core +) +FetchContent_MakeAvailable(azure-core) + +set(azure-core-cpp_DIR "${CMAKE_CURRENT_BINARY_DIR}/azure-core-cpp-config") +file(MAKE_DIRECTORY "${azure-core-cpp_DIR}") +file(WRITE "${azure-core-cpp_DIR}/azure-core-cppConfig.cmake" + "set(azure-core-cpp_FOUND TRUE)\n" +) +set(azure-core-cpp_DIR "${azure-core-cpp_DIR}" CACHE PATH "" FORCE) +# --------------------------------------------------------------------------- +# azure-core-amqp (find_package(azure-core-cpp) → package above) +# --------------------------------------------------------------------------- FetchContent_Declare( - azure-sdk-for-cpp + azure-core-amqp GIT_REPOSITORY https://github.com/Azure/azure-sdk-for-cpp.git GIT_TAG azure-core-amqp_1.0.0-beta.11 GIT_SHALLOW TRUE + SOURCE_SUBDIR sdk/core/azure-core-amqp ) -FetchContent_GetProperties(azure-sdk-for-cpp) -if(NOT azure-sdk-for-cpp_POPULATED) - FetchContent_Populate(azure-sdk-for-cpp) - add_subdirectory( - ${azure-sdk-for-cpp_SOURCE_DIR}/sdk/core/azure-core - ${azure-sdk-for-cpp_BINARY_DIR}/azure-core - EXCLUDE_FROM_ALL - ) - add_subdirectory( - ${azure-sdk-for-cpp_SOURCE_DIR}/sdk/core/azure-core-amqp - ${azure-sdk-for-cpp_BINARY_DIR}/azure-core-amqp - EXCLUDE_FROM_ALL - ) -endif() +FetchContent_MakeAvailable(azure-core-amqp) -# uAMQP backend APIs used by this plugin (CreateMessageSender, MessageSender::Send, ...) +# --------------------------------------------------------------------------- +# Plugin +# --------------------------------------------------------------------------- target_compile_definitions(${target_name} PRIVATE ENABLE_UAMQP=1) target_link_libraries(${target_name} PUBLIC From 89f3624c1026dcb89f92b412331c1d02e68f3cb9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Komor?= Date: Wed, 16 Sep 2026 11:28:16 +0200 Subject: [PATCH 10/10] Restore servicebus plugin to main (drop local CMake experiments). Keep kafka in release.yml; put servicebus back on vcpkg toolchain. --- .github/workflows/release.yml | 4 +- servicebus/CMakeLists.txt | 2 +- servicebus/Readme.md | 40 +++++++----- servicebus/ServiceBusPlugin/CMakeLists.txt | 76 ++++------------------ servicebus/vcpkg.json | 9 +++ 5 files changed, 48 insertions(+), 83 deletions(-) create mode 100644 servicebus/vcpkg.json diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 4bac5a8..0024b7a 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -82,9 +82,9 @@ jobs: servicebus) echo "DIRECTORY=servicebus" >> "$GITHUB_ENV" echo "ARTIFACT_PATH=servicebus/build/ServiceBusPlugin" >> "$GITHUB_ENV" - echo "CONFIGURE_ARGS=" >> "$GITHUB_ENV" + echo "CONFIGURE_ARGS=-DCMAKE_TOOLCHAIN_FILE=${VCPKG_INSTALLATION_ROOT}/scripts/buildsystems/vcpkg.cmake" >> "$GITHUB_ENV" if [ "${{ runner.os }}" = "Linux" ]; then - echo "INSTALL_PACKAGES=build-essential cmake git ninja-build pkg-config libcurl4-openssl-dev libssl-dev" >> "$GITHUB_ENV" + echo "INSTALL_PACKAGES=build-essential cmake git ninja-build pkg-config" >> "$GITHUB_ENV" elif [ "${{ runner.os }}" = "macOS" ]; then echo "INSTALL_PACKAGES=cmake ninja pkg-config" >> "$GITHUB_ENV" else diff --git a/servicebus/CMakeLists.txt b/servicebus/CMakeLists.txt index 25f1c2f..b07e32e 100644 --- a/servicebus/CMakeLists.txt +++ b/servicebus/CMakeLists.txt @@ -1,4 +1,4 @@ -set(CMAKE_MIN 3.25) +set(CMAKE_MIN 3.22) cmake_minimum_required (VERSION ${CMAKE_MIN}) set(CMAKE_POLICY_VERSION_MINIMUM ${CMAKE_MIN}) cmake_policy(VERSION ${CMAKE_MIN}) diff --git a/servicebus/Readme.md b/servicebus/Readme.md index f6bc823..2b2311d 100644 --- a/servicebus/Readme.md +++ b/servicebus/Readme.md @@ -7,27 +7,35 @@ factory symbols (`CreateTransportChannel` / `DestroyTransportChannel` and `CreateServer` / `DestroyServer`). It is written in C++ and talks to Azure Service Bus over its native AMQP 1.0 protocol using -the Azure SDK for C++ AMQP library (`azure-core-amqp`), acquired through CMake `FetchContent` -(same pattern as the RabbitMQ plugin). +the Azure SDK for C++ AMQP library (`azure-core-amqp`), acquired through vcpkg. ## 1) Clone repository ```bash -git clone https://github.com/grft-dev/graftcode-extensions.git -cd graftcode-extensions/servicebus +git clone https://github.com/grft-dev/graftcode-plugins.git +cd graftcode-plugins/servicebus ``` -## 2) Configure with CMake +## 2) Get vcpkg + +The plugin depends on `azure-core-amqp-cpp` and `nlohmann-json`, declared in `vcpkg.json`. ```bash -cmake -S . -B build -DCMAKE_BUILD_TYPE=Release +git clone https://github.com/microsoft/vcpkg.git +./vcpkg/bootstrap-vcpkg.sh # on Windows: .\vcpkg\bootstrap-vcpkg.bat ``` -CMake downloads **nlohmann/json** and **Azure SDK for C++** (`azure-core` + -`azure-core-amqp`) via `FetchContent`. No vcpkg toolchain is required. First -configure needs network access for those downloads. On Windows the HTTP transport is WinHTTP (no libcurl); on Linux/macOS it uses libcurl. +## 3) Configure with CMake -## 3) Build +Point CMake at the vcpkg toolchain so the dependencies are installed and discovered +automatically (manifest mode): + +```bash +cmake -S . -B build -DCMAKE_BUILD_TYPE=Release \ + -DCMAKE_TOOLCHAIN_FILE=./vcpkg/scripts/buildsystems/vcpkg.cmake +``` + +## 4) Build ```bash cmake --build build --config Release @@ -39,11 +47,12 @@ As a result, you will receive: If the generated library is `libServiceBusPlugin.*`, use plugin name: `libServiceBusPlugin`. -## 4) Download GG +## 5) Download GG Download `gg` from: - https://github.com/grft-dev/graftcode-gateway/releases/ -## 5) Create the queues + +## 6) Create the queues In the Azure portal (or via Azure CLI) create two queues in your Service Bus namespace, for example `myqueue` (requests) and `myqueue.reply` (responses). The **reply queue must be @@ -75,7 +84,7 @@ connection string (or set `"useDevelopmentEmulator": true` in config) and point emulator endpoint. See the official emulator: https://learn.microsoft.com/azure/service-bus-messaging/test-locally-with-service-bus-emulator -## 6) Run GG with sample library +## 7) Run GG with sample library In your sample folder, create `pluginConfig.json`: @@ -101,7 +110,7 @@ Then run: ./gg .\PhysicsCalculator.dll --config .\pluginConfig.json ``` -## 7) Get installation command +## 8) Get installation command Visit `http://localhost:81/GV`, select your package manager, and copy the generated installation command. @@ -113,7 +122,7 @@ dotnet new console dotnet add package -s https://grft.dev/019cf6aa-e2e0-74e7-a2b0-be30db97ccb5__graftcode graft.nuget.physicscalculator --version 1.0.0 ``` -## 8) Configure Graft after installation +## 9) Configure Graft after installation Use this configuration: @@ -207,4 +216,3 @@ Create the topic and subscription (the subscription does **not** need sessions): az servicebus topic create --resource-group --namespace-name --name mytopic az servicebus topic subscription create --resource-group --namespace-name --topic-name mytopic --name mysubscription ``` - diff --git a/servicebus/ServiceBusPlugin/CMakeLists.txt b/servicebus/ServiceBusPlugin/CMakeLists.txt index 7c4d7a7..b6be69d 100644 --- a/servicebus/ServiceBusPlugin/CMakeLists.txt +++ b/servicebus/ServiceBusPlugin/CMakeLists.txt @@ -1,77 +1,25 @@ -# Graftcode Azure Service Bus plugin (AMQP 1.0 via Azure SDK for C++). -# Standalone project — same idea as rabbitmq/: FetchContent each dependency, -# then link. azure-core and azure-core-amqp are separate FetchContent projects. +# Graftcode Azure Service Bus plugin (AMQP 1.0 via the Azure SDK for C++). set(target_name ServiceBusPlugin) add_library(${target_name} SHARED - TransportServiceBus.cpp - ServiceBusClient.cpp - ServiceBusServer.cpp + "TransportServiceBus.cpp" + "ServiceBusClient.cpp" + "ServiceBusServer.cpp" ) target_include_directories(${target_name} PUBLIC "${CMAKE_SOURCE_DIR}/GraftcodePluginsInterfaces" ) -include(FetchContent) +# Both dependencies are provided by vcpkg (see vcpkg.json). Configure with the vcpkg +# toolchain file so find_package can locate them, e.g.: +# cmake -S . -B build -DCMAKE_TOOLCHAIN_FILE=/scripts/buildsystems/vcpkg.cmake +find_package(azure-core-amqp-cpp CONFIG REQUIRED) +find_package(nlohmann_json CONFIG REQUIRED) -# --------------------------------------------------------------------------- -# nlohmann/json -# --------------------------------------------------------------------------- -FetchContent_Declare( - json - GIT_REPOSITORY https://github.com/nlohmann/json.git - GIT_TAG v3.12.0 - GIT_SHALLOW TRUE -) -FetchContent_MakeAvailable(json) - -# --------------------------------------------------------------------------- -# Azure SDK build switches (apply before FetchContent of azure-* targets) -# --------------------------------------------------------------------------- -set(FETCH_SOURCE_DEPS ON CACHE BOOL "" FORCE) -set(BUILD_TESTING OFF CACHE BOOL "" FORCE) -set(BUILD_SAMPLES OFF CACHE BOOL "" FORCE) -set(WARNINGS_AS_ERRORS OFF CACHE BOOL "" FORCE) -set(DISABLE_AMQP OFF CACHE BOOL "" FORCE) -set(BUILD_TRANSPORT_CURL ON CACHE BOOL "" FORCE) -set(BUILD_TRANSPORT_WINHTTP ON CACHE BOOL "" FORCE) - -# --------------------------------------------------------------------------- -# azure-core (package name consumed by azure-core-amqp: azure-core-cpp) -# Requires CMake >= 3.25 for SOURCE_SUBDIR. -# --------------------------------------------------------------------------- -FetchContent_Declare( - azure-core - GIT_REPOSITORY https://github.com/Azure/azure-sdk-for-cpp.git - GIT_TAG azure-core-amqp_1.0.0-beta.11 - GIT_SHALLOW TRUE - SOURCE_SUBDIR sdk/core/azure-core -) -FetchContent_MakeAvailable(azure-core) - -set(azure-core-cpp_DIR "${CMAKE_CURRENT_BINARY_DIR}/azure-core-cpp-config") -file(MAKE_DIRECTORY "${azure-core-cpp_DIR}") -file(WRITE "${azure-core-cpp_DIR}/azure-core-cppConfig.cmake" - "set(azure-core-cpp_FOUND TRUE)\n" -) -set(azure-core-cpp_DIR "${azure-core-cpp_DIR}" CACHE PATH "" FORCE) - -# --------------------------------------------------------------------------- -# azure-core-amqp (find_package(azure-core-cpp) → package above) -# --------------------------------------------------------------------------- -FetchContent_Declare( - azure-core-amqp - GIT_REPOSITORY https://github.com/Azure/azure-sdk-for-cpp.git - GIT_TAG azure-core-amqp_1.0.0-beta.11 - GIT_SHALLOW TRUE - SOURCE_SUBDIR sdk/core/azure-core-amqp -) -FetchContent_MakeAvailable(azure-core-amqp) - -# --------------------------------------------------------------------------- -# Plugin -# --------------------------------------------------------------------------- +# The vcpkg azure-core-amqp-cpp port is built with the uAMQP backend but does not propagate +# the backend selection macro to consumers, so the public uAMQP API (CreateMessageSender, +# MessageSender::Send, etc.) is compiled out unless we define it here. target_compile_definitions(${target_name} PRIVATE ENABLE_UAMQP=1) target_link_libraries(${target_name} PUBLIC diff --git a/servicebus/vcpkg.json b/servicebus/vcpkg.json new file mode 100644 index 0000000..7e8cbe2 --- /dev/null +++ b/servicebus/vcpkg.json @@ -0,0 +1,9 @@ +{ + "name": "graftcode-servicebus-plugin", + "version": "1.0.0", + "description": "Graftcode Azure Service Bus transport/server plugin built on the Azure SDK for C++ AMQP library.", + "dependencies": [ + "azure-core-amqp-cpp", + "nlohmann-json" + ] +}