I just published a blog post on the codecentric blog site:
How to write a Kotlin DSL – e.g. for Apache Kafka and in German: Wie schreibt man eine Kotlin DSL – z.B. für Apache Kafka?
In this post I show how to implement an annotation-based Logger injection in Spring when writing the application with Kotlin.
The used technique is quite far from new, basically what I do is implement a BeanPostProcessor which scans the properties of the beans for fields annotated with a custom annotation and sets these fields to a Logger instance. Examples how to implement this in Java can be found on the web, here I show the Kotlin version.
In this example I have an innovative REST controller wich has the ability to say hello. This controller has an injected service to do the processing of the incoming name. The reason for implementing this in two classes is that I want to show 2 different ways to inject a Logger.
The whole project for this example is available on GitHub.
I use Spring-Boot to setup the project. There is nothing special in relation to boot here, but it gets me fast on the track. I created my project from within IntelliJ IDEA as a Spring-Boot project, but you can use https://start.spring.io/ as well. Language is Kotlin, and the only dependency needed is web.
I added a simple logback.xml to my project:
<?xml version="1.0" encoding="UTF-8"?>
<configuration>
<include resource="org/springframework/boot/logging/logback/base.xml"/>
<logger name="com.sothawo" level="DEBUG"/>
<root level="WARN"/>
</configuration>
this sets the global level to WARN, and everything from my code to DEBUG.
The first class I show is HelloService, this is a Spring component which has one method that takes a string, and returns it concatenated to “hello “, and of course this is logged. In this class the Logger is injected in a normal property of the class by using the @Slf4jLogger annotation which I will show later:
package com.sothawo
import org.slf4j.Logger
import org.springframework.stereotype.Service
/**
* @author P.J. Meisch (pj.meisch@sothawo.com)
*/
@Service
class HelloService {
@Slf4jLogger
lateinit var log: Logger
fun sayHello(name: String): String {
log.info("saying hello to: $name")
return "hello $name"
}
}
Using a normal property would create multiple instances of the same Logger-name if multiple HelloService instances were created (which they aren’t, but if you would change the scope this might happen). Therefore a better solution is to put the Logger in the companion object . This can be seen in the HelloController class:
package com.sothawo
import org.slf4j.Logger
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.http.ResponseEntity
import org.springframework.web.bind.annotation.GetMapping
import org.springframework.web.bind.annotation.PathVariable
import org.springframework.web.bind.annotation.RequestMapping
import org.springframework.web.bind.annotation.RestController
/**
* @author P.J. Meisch (pj.meisch@sothawo.com)
*/
@RestController
@RequestMapping("/hello")
class HelloController(val helloService: HelloService) {
@GetMapping("/{name}")
fun sayHello(@PathVariable name: String): ResponseEntity {
log.info("sayHello called with arg $name")
return ResponseEntity.ok(helloService.sayHello(name))
}
companion object {
@Slf4jLogger
lateinit var log: Logger
}
}
package com.sothawo
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.config.BeanPostProcessor
import org.springframework.stereotype.Component
import kotlin.reflect.KMutableProperty
import kotlin.reflect.full.companionObjectInstance
import kotlin.reflect.full.declaredMemberProperties
/**
* @author P.J. Meisch (pj.meisch@sothawo.com)
*/
@Target(AnnotationTarget.PROPERTY)
annotation class Slf4jLogger
@Component
class LoggingInjector : BeanPostProcessor {
override fun postProcessBeforeInitialization(bean: Any, beanName: String) =
bean.also {
try {
val loggerName = it::class.java.canonicalName!!
processObject(it, loggerName)
it::class.companionObjectInstance?.let { companion ->
processObject(companion, loggerName)
}
} catch (ignored: Throwable) {
// ignore exceptions, keep the object as it is. not every required class may be found on the classpath as
// SpringBoot tries to load notexisting stuff as well
}
}
private fun processObject(target: Any, loggerName: String) {
target::class.declaredMemberProperties.forEach { property ->
if (property is KMutableProperty<*>) {
property.annotations
.filterIsInstance(<Slf4jLogger>)
.forEach {
property.setter.call(target, LoggerFactory.getLogger(loggerName))
}
}
}
}
}
First I create the annotation with the AnnotationTarget.PROPERTY (lines 14, 15). Then I define a Spring component that implements the BeanPostProcessor interface. The postProcessAfterInitialization method does nothing, it just returns the bean that is passed in, the magic happens in postProcessBeforeInitialization. Here, if the bean that is passed in is not null, it is passed into my processObject method, and then I check if there is a companion object and analyze this in the same way. The Logger name is built from the bean’s class as I do not want to have the companion object’s name in there.
In the processObject method I iterate over the properties of the target object that was passed in. For each property I check if it has the @SLF4JLogger annotation and if so I set it’s value to a newly created Logger instance with the requested name.
When running the program and issuing a call like
curl localhost:8080/hello/world
the log output shows:
. ____ _ __ _ _ /\\ / ___'_ __ _ _(_)_ __ __ _ \ \ \ \ ( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \ \\/ ___)| |_)| | | | | || (_| | ) ) ) ) ' |____| .__|_| |_|_| |_\__, | / / / / =========|_|==============|___/=/_/_/_/ :: Spring Boot :: (v1.5.4.RELEASE) 2017-07-12 15:58:32.567 INFO 2797 --- [ main] KotlinSpringLoggerInjectionApplicationKt : Starting KotlinSpringLoggerInjectionApplicationKt on pjm.local with PID 2797 (/Users/peter/Entwicklung/sothawo/kotlin-spring-logger-injection/target/classes started by peter in /Users/peter/Entwicklung/sothawo/kotlin-spring-logger-injection) 2017-07-12 15:58:32.571 DEBUG 2797 --- [ main] KotlinSpringLoggerInjectionApplicationKt : Running with Spring Boot v1.5.4.RELEASE, Spring v4.3.9.RELEASE 2017-07-12 15:58:32.571 INFO 2797 --- [ main] KotlinSpringLoggerInjectionApplicationKt : No active profile set, falling back to default profiles: default 2017-07-12 15:58:35.423 INFO 2797 --- [ main] KotlinSpringLoggerInjectionApplicationKt : Started KotlinSpringLoggerInjectionApplicationKt in 3.355 seconds (JVM running for 3.776) 2017-07-12 15:58:51.745 INFO 2797 --- [nio-8080-exec-1] com.sothawo.HelloController : sayHello called with arg world 2017-07-12 15:58:51.745 INFO 2797 --- [nio-8080-exec-1] com.sothawo.HelloService : saying hello to: world
As shown here it is pretty simple to process annotation in Kotlin and use this for custom injection in Spring. Especially statments like annotations.filter { it is Slf4jLogger } are a pretty cool feature.
In this post I show how to combine some language / frameworks and libraries / tools to build a web-based scalable chat application. I chose the following combination of tools:
As I am bad in creating cool names for projects I just put together the first letters of the used tools and named this whole thing kovasbak. The complete source code and project is available on GitHub.
The following screenshot shows four browser windows with four users chatting:
The first thing that I have to do is to get Apache Kafka running. I downloaded the actual version (0.11.0.0) from the Apache Kafka website and unpacked the download in a local directory. According to the Kafka documentation I started first zookeeper and then one Kafka broker:
./bin/zookeeper-server-start.sh config/zookeeper.properties & ./bin/kafka-server-start.sh config/server.properties &
I am just using the default values, that gets Kafka runnning on port 9092.
I am using Java 1.8.0_131 and IntelliJ IDEA, but the project is totally maven based, so you can use the IDE / editor of your choice. To create the project, I used the Spring Intializr integration in IntelliJ, but of course you can create the project by using the Spring Initializr website.
I just selected Kotlin as language, Java version 1.8, Spring Boot 1.5.4 and additionally selected web/vaadin and io/kafka.
After creating the project you end up with the following pom.xml, I only added the highlighted lines to be able to have server-push (more on that later):
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.sothawo</groupId>
<artifactId>kovasbak</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging>
<name>kovasbak</name>
<description>a simple chat system built with Kotlin, Vaadin, spring Boot and Apache Kafka</description>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>1.5.4.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<properties>
<kotlin.compiler.incremental>true</kotlin.compiler.incremental>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<java.version>1.8</java.version>
<kotlin.version>1.1.3</kotlin.version>
<vaadin.version>8.0.6</vaadin.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>com.vaadin</groupId>
<artifactId>vaadin-spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>com.vaadin</groupId>
<artifactId>vaadin-push</artifactId>
</dependency>
<dependency>
<groupId>org.jetbrains.kotlin</groupId>
<artifactId>kotlin-stdlib-jre8</artifactId>
<version>${kotlin.version}</version>
</dependency>
<dependency>
<groupId>org.jetbrains.kotlin</groupId>
<artifactId>kotlin-reflect</artifactId>
<version>${kotlin.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>com.vaadin</groupId>
<artifactId>vaadin-bom</artifactId>
<version>${vaadin.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<build>
<sourceDirectory>${project.basedir}/src/main/kotlin</sourceDirectory>
<testSourceDirectory>${project.basedir}/src/test/kotlin</testSourceDirectory>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
<plugin>
<artifactId>kotlin-maven-plugin</artifactId>
<groupId>org.jetbrains.kotlin</groupId>
<version>${kotlin.version}</version>
<configuration>
<compilerPlugins>
<plugin>spring</plugin>
</compilerPlugins>
<jvmTarget>1.8</jvmTarget>
</configuration>
<executions>
<execution>
<id>compile</id>
<phase>compile</phase>
<goals>
<goal>compile</goal>
</goals>
</execution>
<execution>
<id>test-compile</id>
<phase>test-compile</phase>
<goals>
<goal>test-compile</goal>
</goals>
</execution>
</executions>
<dependencies>
<dependency>
<groupId>org.jetbrains.kotlin</groupId>
<artifactId>kotlin-maven-allopen</artifactId>
<version>${kotlin.version}</version>
</dependency>
</dependencies>
</plugin>
</plugins>
</build>
</project>
In this post I will only show the relevant lines from the code, I will skip package and import statements, the full code is available at GitHub.
The application class created by the initializr just gets one additional line:
@SpringBootApplication
@EnableKafka
class KovasbakApplication
fun main(args: Array<String>) {
SpringApplication.run(KovasbakApplication::class.java, *args)
}
The @EnableKafka annotation is used to tell Spring Boot to pull in the kafka related classes and libs.
The ChatDisplay is the Panel displaying the chat messages. I first used a TextArea, but had problems with programmatically scrolling to the bottom. So I created this small class that uses a Label to display the data:
class ChatDisplay : Panel() {
val text: Label
init {
setSizeFull()
text = Label().apply { contentMode = ContentMode.HTML }
content = VerticalLayout().apply { addComponent(text) }
}
fun addMessage(user: String, message: String) {
text.value = when {
text.value.isNullOrEmpty() -> "<em>$user:</em> $message"
else -> text.value + "<br/><em>$user:</em> $message"
}
scrollTop = Int.MAX_VALUE
}
}
This is the main UI class:
@SpringUI
@PreserveOnRefresh
@Push
class ChatUI : UI(), KafkaConnectorListener {
lateinit var user: String
val chatDisplay = ChatDisplay()
val userLabel = Label()
@Autowired
lateinit var kafkaConnector: KafkaConnector
// skipping content here....
companion object {
val log: Logger = LoggerFactory.getLogger(ChatUI::class.java)
}
}
It is marked as a Vaadin UI with @SpringUI, @PreserveOnRefresh keeps the session when the browser is reloaded, and @Push marks this for server-push when new messages arrive from Kafka. The class implements an interface KafkaConnectorListener which is described together with the KafkaConnector class.
The ChatUI has the following fields:
It further has a companion object containing the Logger. I now show the methods of the class:
override fun init(vaadinRequest: VaadinRequest?) {
kafkaConnector.addListener(this)
content = VerticalLayout().apply {
setSizeFull()
addComponents(chatDisplay, createInputs())
setExpandRatio(chatDisplay, 1F)
}
askForUserName()
}
private fun createInputs(): Component {
return HorizontalLayout().apply {
setWidth(100F, Sizeable.Unit.PERCENTAGE)
val messageField = TextField().apply { setWidth(100F, Sizeable.Unit.PERCENTAGE) }
val button = Button("Send").apply {
setClickShortcut(ShortcutAction.KeyCode.ENTER)
addClickListener {
kafkaConnector.send(user, messageField.value)
messageField.apply { clear(); focus() }
}
}
addComponents(userLabel, messageField, button)
setComponentAlignment(userLabel, Alignment.MIDDLE_LEFT)
setExpandRatio(messageField, 1F)
}
}
This sets up the basic layout with the ChatDisplay and the other UI elements, registers the ChatUI with the KafkaConnector. The click handler for the send button is set up so that the user name and the content of the message TextField are sent to the KafkaConnector (see marked line).
After setting up the layout, the user is asked for her name with the following method:
private fun askForUserName() {
addWindow(Window("your user:").apply {
isModal = true
isClosable = false
isResizable = false
content = VerticalLayout().apply {
val nameField = TextField().apply { focus() }
addComponent(nameField)
addComponent(Button("OK").apply {
setClickShortcut(ShortcutAction.KeyCode.ENTER)
addClickListener {
user = nameField.value
if (!user.isNullOrEmpty()) {
close()
userLabel.value = user
log.info("user entered: $user")
}
}
})
}
center()
})
}
This shows a modal window where the user’s name must be entered.
There is a method that is called when the UI is disposed:
override fun detach() {
kafkaConnector.removeListener(this)
super.detach()
log.info("session ended for user $user")
}
The code used to send the actual message to the kafka connector was already shown, the last thing in this class is the code that is called from the KafkaConnector when new messages arrive:
override fun chatMessage(user: String, message: String) {
access { chatDisplay.addMessage(user, message) }
}
The received data is added to the chatDisplay, but this is wrapped as a Runnable in the UI.access() method for two reasons:
@Push annotation on the class results in a server push to the client which is necessary so that the new messages are immediately shown.All communication with Kafka is wrapped in a Spring Component (thus being a singleton) which just has the following code:
interface KafkaConnectorListener {
fun chatMessage(user: String, message: String)
}
@Component
class KafkaConnector {
val listeners = mutableListOf<KafkaConnectorListener>()
fun addListener(listener: KafkaConnectorListener) {
listeners += listener
}
fun removeListener(listener: KafkaConnectorListener) {
listeners -= listener
}
@Autowired
lateinit var kafka: KafkaTemplate<String, String>
fun send(user: String, message: String) {
log.info("$user sending message \"$message\"")
kafka.send("kovasbak-chat", user, message)
}
@KafkaListener(topics = arrayOf("kovasbak-chat"))
fun receive(consumerRecord: ConsumerRecord<String?, String?>) {
val key: String = consumerRecord.key() ?: "???"
val value: String = consumerRecord.value() ?: "???"
log.info("got kafka record with key \"$key\" and value \"$value\"")
listeners.forEach { listener -> listener.chatMessage(key, value) }
}
companion object {
val log: Logger = LoggerFactory.getLogger(KafkaConnector::class.java)
}
}
First I defined the KafkaConnectorListener interface which the ChatUI class implements so they can be registered for new messages.
The KafkaConnector has a list of listeners and the methods to add and remove listeners. Nothing special here.
For sending a new message to kafka, the send method uses the injected KafkaTemplate (which comes from the spring-kafka library) to send the data to kafka by using the username as key and the message text as payload. The topic name that is used is kovasbak-chat.
By marking the receive method with @KafkaListener the method is called every time when a message in kafka arrives from any client. The data is parsed for the username and message body and the it is sent to all the registered clients. And finally there is a companion object with a Logger.
spring.kafka.consumer.group-id=${random.uuid}
spring.kafka.consumer.auto-offset-reset=latest
spring.kafka.bootstrap-servers=localhost:9092
I use a random kafka consumer-group id so that each instance of my webapp gets all messages, I am not interested in old messages and define the host and port of the kafka broker.
You can either run the program from within the IDE or go to the command line and:
mvn package java -jar target/kovasbak-0.0.1-SNAPSHOT.jar
you can then as well start a second instance on a different port like and access the servers on both localhost:8080 and localhost:8081
java -jar target/kovasbak-0.0.1-SNAPSHOT.jar --server.port=8081
To sum it up: with just a handful of code lines we have a scalable web-based chat-service which uses a scalable backend for message processing.