Kafka codec

source/common/common/logger.h

@@ -36,6 +36,7 @@ namespace Logger {
   FUNCTION(http)                 \
   FUNCTION(http2)                \
   FUNCTION(hystrix)              \
+  FUNCTION(kafka)                \
   FUNCTION(lua)                  \
   FUNCTION(main)                 \
   FUNCTION(misc)                 \

source/extensions/filters/network/kafka/BUILD

@@ -0,0 +1,86 @@
+licenses(["notice"])  # Apache 2
+
+# Kafka network filter.
+# Public docs: docs/root/configuration/network_filters/kafka_filter.rst
+
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_cc_library",
+    "envoy_package",
+)
+
+envoy_package()
+
+envoy_cc_library(
+    name = "kafka_request_codec_lib",
+    srcs = ["request_codec.cc"],
+    hdrs = [
+        "codec.h",
+        "request_codec.h",
+    ],
+    deps = [
+        ":kafka_request_lib",
+        "//source/common/buffer:buffer_lib",
+    ],
+)
+
+envoy_cc_library(
+    name = "kafka_request_lib",
+    srcs = [
+        "generated/kafka_request_resolver.cc",
+        "kafka_request.cc",
+    ],
+    hdrs = [
+        "generated/requests.h",
+        "kafka_request.h",
+    ],
+    deps = [
+        ":parser_lib",
+        ":serialization_lib",
+        "//source/common/common:assert_lib",
+        "//source/common/common:minimal_logger_lib",
+    ],
+)
+
+envoy_cc_library(
+    name = "parser_lib",
+    hdrs = ["parser.h"],
+    deps = [
+        ":kafka_protocol_lib",
+        ":message_lib",
+        "//source/common/common:minimal_logger_lib",
+    ],
+)
+
+envoy_cc_library(
+    name = "message_lib",
+    hdrs = [
+        "message.h",
+    ],
+    deps = [
+    ],
+)
+
+envoy_cc_library(
+    name = "serialization_lib",
+    hdrs = [
+        "generated/serialization_composite.h",
+        "serialization.h",
+    ],
+    deps = [
+        ":kafka_protocol_lib",
+        "//include/envoy/buffer:buffer_interface",
+        "//source/common/common:byte_order_lib",
+    ],
+)
+
+envoy_cc_library(
+    name = "kafka_protocol_lib",
+    hdrs = [
+        "kafka_types.h",
+    ],
+    external_deps = ["abseil_optional"],
+    deps = [
+        "//source/common/common:macros",
+    ],
+)

source/extensions/filters/network/kafka/codec.h

@@ -0,0 +1,44 @@
+#pragma once
+
+#include "envoy/buffer/buffer.h"
+#include "envoy/common/pure.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace Kafka {
+
+/**
+ * Kafka message decoder
+ * @tparam MessageType message type (Kafka request or Kafka response)
+ */
+template <typename MessageType> class MessageDecoder {
+public:
+  virtual ~MessageDecoder() = default;
+
+  /**
+   * Processes given buffer attempting to decode messages of type MessageType container within
+   * @param data buffer instance
+   */
+  virtual void onData(Buffer::Instance& data) PURE;
+};
+
+/**
+ * Kafka message decoder
+ * @tparam MessageType message type (Kafka request or Kafka response)
+ */
+template <typename MessageType> class MessageEncoder {
+public:
+  virtual ~MessageEncoder() = default;
+
+  /**
+   * Encodes given message
+   * @param message message to be encoded
+   */
+  virtual void encode(const MessageType& message) PURE;
+};
+
+} // namespace Kafka
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy

source/extensions/filters/network/kafka/generated/kafka_request_resolver.cc

@@ -0,0 +1,36 @@
+// DO NOT EDIT - THIS FILE WAS GENERATED
+// clang-format off
+#include "extensions/filters/network/kafka/generated/requests.h"
+#include "extensions/filters/network/kafka/kafka_request.h"
+#include "extensions/filters/network/kafka/parser.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace Kafka {
+
+const RequestParserResolver RequestParserResolver::INSTANCE;
+
+ParserSharedPtr RequestParserResolver::createParser(int16_t api_key, int16_t api_version,
+                                                    RequestContextSharedPtr context) const {
+
+  if (8 == api_key && 0 == api_version) {
+    return std::make_shared<OffsetCommitRequestV0Parser>(context);
+  }
+  if (8 == api_key && 1 == api_version) {
+    return std::make_shared<OffsetCommitRequestV1Parser>(context);
+  }
+  if (8 == api_key && 2 == api_version) {
+    return std::make_shared<OffsetCommitRequestV2Parser>(context);
+  }
+  if (8 == api_key && 3 == api_version) {
+    return std::make_shared<OffsetCommitRequestV3Parser>(context);
+  }
+  return std::make_shared<SentinelParser>(context);
+}
+
+} // namespace Kafka
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy
+// clang-format on