@Component
@ConditionalOnProperty(name = "enable.kafka", havingValue="true")
public class KafkaSenderService {
    private static Logger logger = LoggerFactory.getLogger(DatabaseInitialService.class);

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @EventListener(ApplicationStartedEvent.class)
    public void send() {
        kafkaTemplate.send("testcontainers", "kafka", "TestcontainersKafka").addCallback(result -> {
            if (result != null) {
                RecordMetadata recordMetadata = result.getRecordMetadata();
                logger.info("producer send data to {}, {}, {}", recordMetadata.topic(), recordMetadata.partition(),
                        recordMetadata.offset());
            }
        }, ex -> {
            logger.error("something wrong...", ex);
        });

        kafkaTemplate.flush();
    }
}

@SpringBootTest
class KafkaTest {
    private static Logger logger = LoggerFactory.getLogger(KafkaTest.class);

    static KafkaContainer kafkaContainer = new KafkaContainer();

    @DynamicPropertySource
    static void kafkaProperties(DynamicPropertyRegistry registry) {
        kafkaContainer.start();
        registry.add("spring.kafka.properties.bootstrap.servers", kafkaContainer::getBootstrapServers);
        registry.add("spring.kafka.consumer.group-id", () -> "testcontainersapp");
        registry.add("spring.kafka.consumer.properties.auto.offset.reset", () -> "earliest");
    }

    @Autowired
    private KafkaProperties properties;

    @Test
    public void testKafkaProducerSendDataAndConsumerReceiveData() {
        final Consumer<String, String>[] consumer = new Consumer[]{createConsumer("testcontainers")};

        String actual = "";
        while (true) {
            ConsumerRecords<String, String> records = KafkaTestUtils.getRecords(consumer[0], 10000);
            if (records.isEmpty()) {
                break;
            }
            for (ConsumerRecord<String, String> record : records) {
                actual = record.value();
            }
        }

        assertEquals("TestcontainersKafka", actual);
    }

    private Consumer<String, String> createConsumer(String topicName) {
        Consumer<String, String>
                consumer = new DefaultKafkaConsumerFactory<>(properties.buildConsumerProperties(), StringDeserializer::new,
                        StringDeserializer::new).createConsumer();

        consumer.subscribe(Collections.singletonList(topicName));
        return consumer;
    }
}

/** Model */
@Entity
@Table(name = "book_category", schema = "dbo")
public class BookCategory {
    private int bookSeq;
    private String categoryName;
    private int sort;
    private String suspend;

    @Id
    @GeneratedValue(strategy=GenerationType.IDENTITY)
    @Column(name = "book_seq")
    public int getBookSeq() {
        return bookSeq;
    }

    public void setBookSeq(int bookSeq) {
        this.bookSeq = bookSeq;
    }

    @Basic
    @Column(name = "category_name")
    public String getCategoryName() {
        return categoryName;
    }

    public void setCategoryName(String categoryName) {
        this.categoryName = categoryName;
    }

    @Basic
    @Column(name = "sort")
    public int getSort() {
        return sort;
    }

    public void setSort(int sort) {
        this.sort = sort;
    }

    @Basic
    @Column(name = "suspend")
    public String getSuspend() {
        return suspend;
    }

    public void setSuspend(String suspend) {
        this.suspend = suspend;
    }

    public static BookCategory createBookCategory(String categoryName, int sort) {
        BookCategory bookCategory = new BookCategory();
        bookCategory.setCategoryName(categoryName);
        bookCategory.setSort(sort);
        bookCategory.setSuspend("N");
        return bookCategory;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        BookCategory that = (BookCategory) o;
        return bookSeq == that.bookSeq &&
            sort == that.sort &&
            Objects.equals(categoryName, that.categoryName) &&
            Objects.equals(suspend, that.suspend);
    }

    @Override
    public int hashCode() {
        return Objects.hash(bookSeq, categoryName, sort, suspend);
    }
}

/** Repository */
public interface BookCategoryRepository extends JpaRepository<BookCategory, Integer> {
}

@SpringBootTest
public class DatabaseTest {
    private static Logger logger = LoggerFactory.getLogger(DatabaseTest.class);

    static MSSQLServerContainer mssqlserver = (MSSQLServerContainer) new MSSQLServerContainer()
        .withInitScript("doc/ddl.sql");

    @DynamicPropertySource
    static void mssqlProperties(DynamicPropertyRegistry registry) {
        mssqlserver.start();
        registry.add("spring.datasource.driver-class-name", mssqlserver::getDriverClassName);
        registry.add("spring.datasource.url", () -> mssqlserver.getJdbcUrl());
        registry.add("spring.datasource.username", mssqlserver::getUsername);
        registry.add("spring.datasource.password", mssqlserver::getPassword);
    }

    @Autowired
    private BookCategoryRepository bookCategoryRepository;

    @Test
    void testBookCategoryListSizeIs10() {
        List<BookCategory> bookCategoryList = bookCategoryRepository.findAll();
        assertEquals(10, bookCategoryList.size());
    }
}

@Configuration
public class RedisConfiguration {
    @Bean
    public RedisTemplate redisTemplate(RedisConnectionFactory connectionFactory) {
        RedisTemplate redisTemplate = new RedisTemplate();
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setConnectionFactory(connectionFactory);
        return redisTemplate;
    }
}

@SpringBootTest
public class RedisTest {
    private static Logger logger = LoggerFactory.getLogger(RedisTest.class);

    static GenericContainer redis = new GenericContainer("redis:5.0.5")
        .withExposedPorts(6379);

    @DynamicPropertySource
    static void redisProperties(DynamicPropertyRegistry registry) {
        redis.start();
        registry.add("spring.redis.host", redis::getContainerIpAddress);
        registry.add("spring.redis.port", redis::getFirstMappedPort);
    }

    @Autowired
    private RedisTemplate redisTemplate;

    @Test
    public void testSetAndGetWithString() {
        redisTemplate.opsForValue().set("k1", "v1");
        assertEquals("v1", redisTemplate.opsForValue().get("k1"));
    }

    @Test
    public void testSetAndGetWithObject() {
        BookCategory bookCategory = BookCategory.createBookCategory("生活風格", 10);
        redisTemplate.opsForValue().set("k2", bookCategory);
        String categoryName = ((BookCategory) redisTemplate.opsForValue().get("k2")).getCategoryName();
        assertEquals("生活風格", categoryName);
    }

    @After
    public void destory() {
        redis.stop();
    }
}