kafka.test.js.snap

// Jest Snapshot v1, https://goo.gl/fbAQLP

exports[`template integration tests for generated files using the generator and mqtt example should generate proper config, services and DTOs files for provided kafka 1`] = `
"package com.asyncapi.infrastructure;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.IntegerDeserializer;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.config.SaslConfigs;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.*;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import org.springframework.kafka.support.serializer.JsonSerializer;

import javax.annotation.processing.Generated;
import java.util.HashMap;
import java.util.Map;

@Generated(value="com.asyncapi.generator.template.spring", date="AnyDate")
@Configuration
@EnableKafka
public class Config {

    @Value("\${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("\${spring.kafka.listener.poll-timeout}")
    private long pollTimeout;

    @Value("\${spring.kafka.listener.concurrency}")
    private int concurrency;
    @Bean
    public KafkaTemplate<Integer, Object> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

    @Bean
    public ProducerFactory<Integer, Object> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT");
        props.put(JsonSerializer.TYPE_MAPPINGS,
        "lightMeasuredPayload:com.asyncapi.model.LightMeasuredPayload"
        );
        // See https://kafka.apache.org/documentation/#producerconfigs for more properties
        return props;
    }

    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, Object>>
    kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<Integer, Object> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(concurrency);
        factory.getContainerProperties().setPollTimeout(pollTimeout);
        return factory;
    }

    @Bean
    public ConsumerFactory<Integer, Object> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT");
        props.put(JsonDeserializer.TYPE_MAPPINGS,
        "lightMeasuredPayload:com.asyncapi.model.LightMeasuredPayload"
        );
        props.put(JsonDeserializer.TRUSTED_PACKAGES, "com.asyncapi.model");
        return props;
    }

}
"
`;

exports[`template integration tests for generated files using the generator and mqtt example should generate proper config, services and DTOs files for provided kafka 2`] = `
"package com.asyncapi.service;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import com.asyncapi.model.LightMeasuredPayload;
import javax.annotation.processing.Generated;

@Generated(value="com.asyncapi.generator.template.spring", date="AnyDate")
@Service
public class PublisherService {

    @Autowired
    private KafkaTemplate<Integer, Object> kafkaTemplate;

    
    public void updateLightMeasurement(Integer key, LightMeasuredPayload lightMeasuredPayload) {
        Message<LightMeasuredPayload> message = MessageBuilder.withPayload(lightMeasuredPayload)
                .setHeader(KafkaHeaders.TOPIC, "event.lighting.measured")
                .setHeader(KafkaHeaders.KEY, key)
                .build();
        kafkaTemplate.send(message);
    }
}
"
`;

exports[`template integration tests for generated files using the generator and mqtt example should generate proper config, services and DTOs files for provided kafka 3`] = `
"package com.asyncapi.service;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Service;

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
    
import com.asyncapi.model.LightMeasuredPayload;

import javax.annotation.processing.Generated;

@Generated(value="com.asyncapi.generator.template.spring", date="AnyDate")
@Service
public class MessageHandlerService {

    private static final Logger LOGGER = LoggerFactory.getLogger(MessageHandlerService.class);

    
    
    @KafkaListener(topics = "event.lighting.measured", groupId = "my-group")
    public void readLightMeasurement(@Payload LightMeasuredPayload payload,
                       @Header(KafkaHeaders.RECEIVED_KEY) Integer key,
                       @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
                       @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long timestamp) {
        LOGGER.info("Key: " + key + ", Payload: " + payload.toString() + ", Timestamp: " + timestamp + ", Partition: " + partition);
    }
    

}
"
`;

exports[`template integration tests for generated files using the generator and mqtt example should generate proper config, services and DTOs files for provided kafka 4`] = `
"package com.asyncapi.model;


import jakarta.validation.constraints.*;
import jakarta.validation.Valid;

import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonValue;

import javax.annotation.processing.Generated;
import java.util.List;
import java.util.Objects;


@Generated(value="com.asyncapi.generator.template.spring", date="AnyDate")
public class LightMeasuredPayload {
    
    private @Valid Integer lumens;
    
    private @Valid java.time.OffsetDateTime sentAt;
    

    

    /**
     * Light intensity measured in lumens.
     */
    @JsonProperty("lumens")
    public Integer getLumens() {
        return lumens;
    }

    public void setLumens(Integer lumens) {
        this.lumens = lumens;
    }
    

    /**
     * Date and time when the message was sent.
     */
    @JsonProperty("sentAt")
    public java.time.OffsetDateTime getSentAt() {
        return sentAt;
    }

    public void setSentAt(java.time.OffsetDateTime sentAt) {
        this.sentAt = sentAt;
    }
    
    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        LightMeasuredPayload lightMeasuredPayload = (LightMeasuredPayload) o;
        return 
            Objects.equals(this.lumens, lightMeasuredPayload.lumens) &&
            Objects.equals(this.sentAt, lightMeasuredPayload.sentAt);
    }

    @Override
    public int hashCode() {
        return Objects.hash(lumens, sentAt);
    }

    @Override
    public String toString() {
        return "class LightMeasuredPayload {\\n" +
        
                "    lumens: " + toIndentedString(lumens) + "\\n" +
                "    sentAt: " + toIndentedString(sentAt) + "\\n" +
                "}";
    }

    /**
     * Convert the given object to string with each line indented by 4 spaces (except the first line).
     */
    private String toIndentedString(Object o) {
        if (o == null) {
           return "null";
        }
        return o.toString().replace("\\n", "\\n    ");
    }
}"
`;

exports[`template integration tests for generated files using the generator and mqtt example should generate proper config, services and DTOs files for provided kafka 5`] = `
"package com.asyncapi.model;

import javax.annotation.processing.Generated;

import jakarta.validation.Valid;
import java.util.Objects;
import java.util.List;


@Generated(value="com.asyncapi.generator.template.spring", date="AnyDate")
public class LightMeasured {
    private @Valid LightMeasuredPayload payload;

    public LightMeasuredPayload getPayload() {
        return payload;
    }

    public void setPayload(LightMeasuredPayload payload) {
        this.payload = payload;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        LightMeasured event = (LightMeasured) o;
        return Objects.equals(this.payload, event.payload);
    }

    @Override
    public int hashCode() {
        return Objects.hash(payload);
    }

    @Override
    public String toString() {
        return "class LightMeasured {\\n" +
                "    payload: " + toIndentedString(payload) + "\\n" +
                "}";
    }

    /**
     * Convert the given object to string with each line indented by 4 spaces (except the first line).
     */
    private String toIndentedString(Object o) {
        if (o == null) {
            return "null";
        }
        return o.toString().replace("\\n", "\\n    ");
    }
}"
`;

exports[`template integration tests for generated files using the generator and mqtt example should generate proper config, services and DTOs files for provided kafka 6`] = `
"package com.asyncapi;

 
import com.asyncapi.model.LightMeasuredPayload;
 
 
import com.asyncapi.model.LightMeasuredPayload;
 
import com.asyncapi.service.PublisherService;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.IntegerDeserializer;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.support.serializer.JsonSerializer;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import org.springframework.test.context.junit4.SpringRunner;
import org.testcontainers.containers.KafkaContainer;
import org.testcontainers.shaded.com.google.common.collect.Lists;

import javax.annotation.processing.Generated;
import java.time.Duration;
import java.util.*;

import static java.util.Collections.emptyList;
import static org.apache.kafka.clients.consumer.ConsumerConfig.*;
import static org.apache.kafka.clients.producer.ProducerConfig.*;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotEquals;

/**
 * Example of tests for kafka based on testcontainers library
 */
@Generated(value="com.asyncapi.generator.template.spring", date="AnyDate")
@RunWith(SpringRunner.class)
@SpringBootTest
public class TestcontainerKafkaTest {

     
    private static final String UPDATELIGHTMEASUREMENT_SUBSCRIBE_TOPIC = "event.lighting.measured";
     
    private static final String READLIGHTMEASUREMENT_PUBLISH_TOPIC = "event.lighting.measured";
     
    @ClassRule
    public static KafkaContainer kafka = new KafkaContainer();
    
    @Autowired
    private PublisherService publisherService;
    
    @DynamicPropertySource
    public static void kafkaProperties(DynamicPropertyRegistry registry) {
        registry.add("spring.kafka.bootstrap-servers", kafka::getBootstrapServers);
    }
     
    @Test
    public void updateLightMeasurementProducerTestcontainers() {
        LightMeasuredPayload payload = new LightMeasuredPayload();
        Integer key = 1;
        Integer wrongKey = key + 1;

        consumeMessages(UPDATELIGHTMEASUREMENT_SUBSCRIBE_TOPIC);

        publisherService.updateLightMeasurement(key, payload);

        ConsumerRecord<Integer, Object> consumedMessage = consumeMessage(UPDATELIGHTMEASUREMENT_SUBSCRIBE_TOPIC);

        assertEquals("Key is wrong", key, consumedMessage.key());
        assertNotEquals("Key is wrong", wrongKey, consumedMessage.key());
    }
     
    @Test
    public void readLightMeasurementConsumerTestcontainers() throws Exception {
        Integer key = 1;
        LightMeasuredPayload payload = new LightMeasuredPayload();

        ProducerRecord<Integer, Object> producerRecord = new ProducerRecord<>(READLIGHTMEASUREMENT_PUBLISH_TOPIC, key, payload);

        sendMessage(producerRecord);

        Thread.sleep(1_000);
    }
    
    
    
    protected void sendMessage(ProducerRecord message) throws Exception {
        try (KafkaProducer<Integer, Object> kafkaProducer = createProducer()) {
            kafkaProducer.send(message).get();
        }
    }

    protected void sendMessage(String topic, Object message) throws Exception {
        try (KafkaProducer<Integer, Object> kafkaProducer = createProducer()) {
            kafkaProducer.send(new ProducerRecord<>(topic, message)).get();
        }
    }

    protected KafkaProducer<Integer, Object> createProducer() {
        return new KafkaProducer<>(getKafkaProducerConfiguration());
    }

    protected Map<String, Object> getKafkaProducerConfiguration() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
        configs.put(KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName());
        configs.put(VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class.getName());
        return configs;
    }
    
    protected ConsumerRecord<Integer, Object> consumeMessage(String topic) {
        return consumeMessages(topic)
                .stream()
                .findFirst()
                .orElseThrow(() -> new IllegalStateException("no message received"));
    }

    protected List<ConsumerRecord<Integer, Object>> consumeMessages(String topic) {
        try (KafkaConsumer<Integer, Object> consumer = createConsumer(topic)) {
            return pollForRecords(consumer);
        }
    }

    protected KafkaConsumer<Integer, Object> createConsumer(String topic) {
        Properties properties = new Properties();
        properties.putAll(getKafkaConsumerConfiguration());
        KafkaConsumer<Integer, Object> consumer = new KafkaConsumer<>(properties);
        consumer.subscribe(Collections.singleton(topic));
        return consumer;
    }

    protected static <K, V> List<ConsumerRecord<K, V>> pollForRecords(KafkaConsumer<K, V> consumer) {
        ConsumerRecords<K, V> received = consumer.poll(Duration.ofSeconds(10L));
        return received == null ? emptyList() : Lists.newArrayList(received);
    }

    protected Map<String, Object> getKafkaConsumerConfiguration() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
        configs.put(GROUP_ID_CONFIG, "testGroup");
        configs.put(AUTO_OFFSET_RESET_CONFIG, "earliest");
        configs.put(KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class.getName());
        configs.put(VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class.getName());
        configs.put(JsonDeserializer.TYPE_MAPPINGS,
        "lightMeasuredPayload:com.asyncapi.model.LightMeasuredPayload"
        );
        configs.put(JsonDeserializer.TRUSTED_PACKAGES, "com.asyncapi.model");
        return configs;
    }
    
}
"
`;

exports[`template integration tests for generated files using the generator and mqtt example should generate proper config, services and DTOs files for provided kafka 7`] = `
"plugins {
	id 'org.springframework.boot' version "$springBootVersion"
	id 'io.spring.dependency-management' version "$springDependencyManager"
	id 'java'
}

group = "com.asyncapi"
version = "0.0.1-SNAPSHOT"
sourceCompatibility = JavaVersion.VERSION_17

repositories {
	mavenCentral()
}

dependencies {
	implementation('org.springframework.kafka:spring-kafka')
	testImplementation('org.springframework.kafka:spring-kafka-test')
	testImplementation('junit:junit:4.13.1')
	testImplementation('org.testcontainers:kafka:1.16.3')
	implementation('com.fasterxml.jackson.jaxrs:jackson-jaxrs-json-provider')
	implementation('com.fasterxml.jackson.datatype:jackson-datatype-jsr310')
	implementation('jakarta.validation:jakarta.validation-api')
	implementation('org.springframework.boot:spring-boot-starter-integration')
	testImplementation('org.springframework.boot:spring-boot-starter-test')
	testImplementation('org.testcontainers:testcontainers:1.16.3')
}
"
`;

exports[`template integration tests for generated files using the generator and mqtt example should generate proper config, services and DTOs files for provided kafka 8`] = `
"springBootVersion=3.1.3
springDependencyManager=1.1.3"
`;