Merge pull request #3 from Amerousful/v2.0

Release 2.0

Author: Amerousful

README.md

@@ -11,7 +11,7 @@ Add to your `pom.xml`
 <dependency>
     <groupId>io.github.amerousful</groupId>
     <artifactId>gatling-kafka</artifactId>
-    <version>1.1</version>
+    <version>2.0</version>
 </dependency>
 ```
 
@@ -20,7 +20,7 @@ Add to your `pom.xml`
 Add to your `build.sbt`
 
 ```scala
-libraryDependencies += "io.github.amerousful" % "gatling-kafka" % "1.1"
+libraryDependencies += "io.github.amerousful" % "gatling-kafka" % "2.0"
 ```
 
 Import:
@@ -70,6 +70,26 @@ Request and reply:
   .key("#{id} #{key}")
   .check(jsonPath("$.m").is("#{payload}_1"))
 ```
+***
+Scenario:
+```scala
+scenario("Kafka Scenario")
+  .exec(kafkaFireAndForget)
+```
+
+Inject:
+
+```scala
+setUp(
+  scn.inject(
+    constantUsersPerSec(2) during(10 seconds)
+  ).protocols(kafkaProtocol)
+)
+```
+
+
+### [Java example](src/test/scala/KafkaJavaExample.java)
+### [Kotlin example](src/test/scala/KafkaKotlinExample.kt)
 
 ***
 
@@ -119,7 +139,7 @@ object CustomMatcher extends KafkaMatcher {
 .messageMatcher(CustomMatcher)
 
 ```
-\
+
 ### Chain for build a request:
 ```text
 send -> 

build.sbt

@@ -1,11 +1,15 @@
 name := "gatling-kafka"
 
-version := "1.1"
+version := "2.0"
 
-scalaVersion := "2.13.8"
+scalaVersion := "2.13.12"
+
+val gatlingVersion = "3.9.5"
 
 libraryDependencies ++= Seq(
-  "io.gatling" % "gatling-core" % "3.10.3" % "provided",
+  "io.gatling" % "gatling-core" % gatlingVersion % "provided",
+  "io.gatling" % "gatling-core-java" % gatlingVersion % "provided",
+  "com.github.spotbugs" % "spotbugs-annotations" % "4.7.3",
 
   "com.typesafe.akka" %% "akka-stream-kafka" % "4.0.2",
   "com.typesafe.akka" %% "akka-protobuf-v3" % "2.6.20",

changelog.txt

@@ -0,0 +1,7 @@
+Changelog 2.0 (09-02-2024)
+----------------------------
+* Implement support for both Java and Kotlin
+* Downgrade Gatling to version '3.9.5'. The backward compatibility is broken for ActionBuilder in version 3.10.x
+* Add support for regular expression (regex) check
+* Add support for header check
+* Fix initialization issue with tracker poll consumer for Send request

src/main/java/io/github/amerousful/kafka/javaapi/Kafka.java

@@ -0,0 +1,26 @@
+
+package io.github.amerousful.kafka.javaapi;
+
+import edu.umd.cs.findbugs.annotations.NonNull;
+import io.github.amerousful.kafka.request.KafkaDslBuilderBase;
+import io.gatling.commons.validation.Validation;
+import io.gatling.core.session.Session;
+import scala.Function1;
+
+public class Kafka {
+  private final KafkaDslBuilderBase wrapped;
+
+  public Kafka(Function1<Session, Validation<String>> name) {
+    wrapped = new KafkaDslBuilderBase(name);
+  }
+
+  @NonNull
+  public KafkaSendActionBuilder.Topic send() {
+    return new KafkaSendActionBuilder.Topic(wrapped.send());
+  }
+
+  @NonNull
+  public RequestReplyActionBuilder.Topic requestReply() {
+    return new RequestReplyActionBuilder.Topic(wrapped.requestReply());
+  }
+}

src/main/java/io/github/amerousful/kafka/javaapi/KafkaBroker.java

@@ -0,0 +1,14 @@
+package io.github.amerousful.kafka.javaapi;
+
+public final class KafkaBroker {
+
+    private final io.github.amerousful.kafka.protocol.KafkaBroker wrapped;
+
+    KafkaBroker(io.github.amerousful.kafka.protocol.KafkaBroker wrapped) {
+        this.wrapped = wrapped;
+    }
+
+    public io.github.amerousful.kafka.protocol.KafkaBroker asScala() {
+        return wrapped;
+    }
+}

src/main/java/io/github/amerousful/kafka/javaapi/KafkaDsl.java

@@ -0,0 +1,86 @@
+package io.github.amerousful.kafka.javaapi;
+
+import edu.umd.cs.findbugs.annotations.NonNull;
+import io.gatling.core.check.Check;
+import io.gatling.core.check.CheckMaterializer;
+import io.gatling.javaapi.core.CheckBuilder;
+import io.gatling.javaapi.core.Session;
+import io.github.amerousful.kafka.javaapi.internal.KafkaCheckType;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+
+import java.util.function.Function;
+
+import static io.gatling.javaapi.core.internal.Converters.toScalaFunction;
+import static io.gatling.javaapi.core.internal.Expressions.*;
+import static io.gatling.javaapi.core.internal.Expressions.toExpression;
+
+public final class KafkaDsl {
+    private KafkaDsl() {
+    }
+
+    public static KafkaProtocolBuilder kafka =
+            new KafkaProtocolBuilder(io.github.amerousful.kafka.Predef.kafka(io.gatling.core.Predef.configuration()));
+
+    @NonNull
+    public static KafkaBroker KafkaBroker(@NonNull String host, int port) {
+        return new KafkaBroker(new io.github.amerousful.kafka.protocol.KafkaBroker(host, port));
+    }
+
+    @NonNull
+    public static Kafka kafka(@NonNull String name) {
+        return new Kafka(toStringExpression(name));
+    }
+
+    @NonNull
+    public static Kafka kafka(@NonNull Function<Session, String> name) {
+        return new Kafka(javaFunctionToExpression(name));
+    }
+
+    @NonNull
+    public static CheckBuilder.MultipleFind<String> header(@NonNull CharSequence name) {
+        return new CheckBuilder.MultipleFind.Default<>(
+                io.github.amerousful.kafka.Predef.header(toStaticValueExpression(name)),
+                KafkaCheckType.Header,
+                String.class,
+                null);
+    }
+
+    @NonNull
+    public static CheckBuilder.MultipleFind<String> header(@NonNull String name) {
+        return new CheckBuilder.MultipleFind.Default<>(
+                io.github.amerousful.kafka.Predef.header(toExpression(name, CharSequence.class)),
+                KafkaCheckType.Header,
+                String.class,
+                null);
+    }
+
+    @NonNull
+    public static CheckBuilder.MultipleFind<String> header(
+            @NonNull Function<Session, CharSequence> name) {
+        return new CheckBuilder.MultipleFind.Default<>(
+                io.github.amerousful.kafka.Predef.header(javaFunctionToExpression(name)),
+                KafkaCheckType.Header,
+                String.class,
+                null);
+    }
+
+    @SuppressWarnings("rawtypes")
+    public static CheckBuilder simpleCheck(Function<ConsumerRecord<String, String>, Boolean> f) {
+        return new CheckBuilder() {
+            @Override
+            public io.gatling.core.check.CheckBuilder<?, ?> asScala() {
+                return new io.gatling.core.check.CheckBuilder() {
+                    @Override
+                    public Check<?> build(CheckMaterializer materializer) {
+                        return io.github.amerousful.kafka.Predef.simpleCheck(toScalaFunction(f::apply));
+                    }
+                };
+            }
+
+            @Override
+            public CheckType type() {
+                return KafkaCheckType.Simple;
+            }
+        };
+    }
+}

src/main/java/io/github/amerousful/kafka/javaapi/KafkaMessageMatcher.java

@@ -0,0 +1,16 @@
+
+package io.github.amerousful.kafka.javaapi;
+
+import edu.umd.cs.findbugs.annotations.NonNull;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.producer.ProducerRecord;
+
+
+public interface KafkaMessageMatcher {
+
+  @NonNull
+  String requestMatchId(@NonNull ProducerRecord<String, String> msg);
+
+  @NonNull
+  String responseMatchId(@NonNull ConsumerRecord<String, String> msg);
+}

src/main/java/io/github/amerousful/kafka/javaapi/KafkaProtocolBuilder.java

@@ -0,0 +1,122 @@
+package io.github.amerousful.kafka.javaapi;
+
+import edu.umd.cs.findbugs.annotations.NonNull;
+import io.gatling.core.protocol.Protocol;
+import io.gatling.javaapi.core.ProtocolBuilder;
+import io.github.amerousful.kafka.protocol.javaapi.KafkaMessageMatchers;
+import scala.Enumeration;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static io.gatling.javaapi.core.internal.Converters.toScalaDuration;
+import static io.gatling.javaapi.core.internal.Converters.toScalaSeq;
+
+public final class KafkaProtocolBuilder implements ProtocolBuilder {
+
+    private final io.github.amerousful.kafka.protocol.KafkaProtocolBuilder wrapped;
+
+    KafkaProtocolBuilder(io.github.amerousful.kafka.protocol.KafkaProtocolBuilder wrapped) {
+        this.wrapped = wrapped;
+    }
+
+    @NonNull
+    public KafkaProtocolBuilder broker(KafkaBroker broker) {
+        return new KafkaProtocolBuilder(wrapped.broker(broker.asScala()));
+    }
+
+    @NonNull
+    public KafkaProtocolBuilder brokers(KafkaBroker... brokers) {
+        List<io.github.amerousful.kafka.protocol.KafkaBroker> listOfBrokersAsScala =
+                Arrays.stream(brokers).map(KafkaBroker::asScala).collect(Collectors.toList());
+        return new KafkaProtocolBuilder(wrapped.brokers(toScalaSeq(listOfBrokersAsScala)));
+    }
+
+    @NonNull
+    public KafkaProtocolBuilder acks(@NonNull String acks) {
+        return new KafkaProtocolBuilder(wrapped.acks(acks));
+    }
+
+    @NonNull
+    public KafkaProtocolBuilder producerKeySerializer(@NonNull String serializer) {
+        return new KafkaProtocolBuilder(wrapped.producerKeySerializer(serializer));
+    }
+
+    @NonNull
+    public KafkaProtocolBuilder producerValueSerializer(@NonNull String serializer) {
+        return new KafkaProtocolBuilder(wrapped.producerValueSerializer(serializer));
+    }
+
+    @NonNull
+    public KafkaProtocolBuilder consumerKeyDeserializer(@NonNull String deserializer) {
+        return new KafkaProtocolBuilder(wrapped.consumerKeyDeserializer(deserializer));
+    }
+
+    @NonNull
+    public KafkaProtocolBuilder consumerValueDeserializer(@NonNull String deserializer) {
+        return new KafkaProtocolBuilder(wrapped.consumerValueDeserializer(deserializer));
+    }
+
+    @NonNull
+    public KafkaProtocolBuilder consumerIdenticalDeserializer(@NonNull String deserializer) {
+        return consumerKeyDeserializer(deserializer)
+                .consumerValueDeserializer(deserializer);
+    }
+
+    @NonNull
+    public KafkaProtocolBuilder producerIdenticalSerializer(@NonNull String serializer) {
+        return new KafkaProtocolBuilder(wrapped.producerIdenticalSerializer(serializer));
+    }
+
+    @NonNull
+    public KafkaProtocolBuilder addProducerProperty(@NonNull String key, @NonNull String value) {
+        return new KafkaProtocolBuilder(wrapped.addProducerProperty(key, value));
+    }
+
+    @NonNull
+    public KafkaProtocolBuilder addConsumerProperty(@NonNull String key, @NonNull String value) {
+        return new KafkaProtocolBuilder(wrapped.addConsumerProperty(key, value));
+    }
+
+    @NonNull
+    public KafkaProtocolBuilder credentials(@NonNull String username, @NonNull String password, boolean sslEnabled, @NonNull Enumeration.Value saslMechanism) {
+        return new KafkaProtocolBuilder(wrapped.credentials(username, password, sslEnabled, saslMechanism));
+    }
+
+    @NonNull
+    public KafkaProtocolBuilder replyTimeout(long timeout) {
+        return replyTimeout(Duration.ofSeconds(timeout));
+    }
+
+    @NonNull
+    public KafkaProtocolBuilder replyTimeout(@NonNull Duration timeout) {
+        return new KafkaProtocolBuilder(wrapped.replyTimeout(toScalaDuration(timeout)));
+    }
+
+    @NonNull
+    public KafkaProtocolBuilder messageMatcher(@NonNull KafkaMessageMatcher matcher) {
+        return new KafkaProtocolBuilder(wrapped.messageMatcher(KafkaMessageMatchers.toScala(matcher)));
+    }
+
+    @NonNull
+    public KafkaProtocolBuilder matchByKey() {
+        return new KafkaProtocolBuilder(wrapped.matchByKey());
+    }
+
+    @NonNull
+    public KafkaProtocolBuilder matchByValue() {
+        return new KafkaProtocolBuilder(wrapped.matchByValue());
+    }
+
+    @NonNull
+    public KafkaProtocolBuilder replyConsumerName(@NonNull String name) {
+        return new KafkaProtocolBuilder(wrapped.replyConsumerName(name));
+    }
+
+    @Override
+    public Protocol protocol() {
+        return wrapped.build();
+    }
+}