Simplify UserState (#18)

* Simplify UserState

* Improve README

Author: rogervinas

README.md

@@ -17,7 +17,7 @@ It offers an abstraction (the **binding**) that works the same whatever undernea
 
 You can also check out [Spring Cloud Stream Kafka Streams first steps](https://github.com/rogervinas/spring-cloud-stream-kafka-streams-first-steps) where I got working a simple example using **Kafka Streams binder**.
 
-In this one the goal is to use the **Kafka Streams binder** and the  [Kafka Streams Processor API](https://kafka.apache.org/documentation/streams/developer-guide/processor-api.html) to implement the following scenario:
+In this one the goal is to use the **Kafka Streams binder** and the [Kafka Streams Processor API](https://kafka.apache.org/documentation/streams/developer-guide/processor-api.html) to implement the following scenario:
 
 ![Diagram](doc/diagram.png)
 
@@ -97,7 +97,7 @@ fun `should publish expired event for one user`() {
 
 We start first with our **UserStateStream** implementation as a **Function**:
 * Which input is a **KStream<String, UserTokenEvent>**, as we want a **String** as the Kafka message's key and a **UserTokenEvent** as the Kafka message's value
-* Which output is a **KStream<String, UserStateEvent>**, same here, **String** as the key and **UserStateEvent** as the value 
+* Which output is a **KStream<String, UserStateEvent>**, same here, **String** as the key and **UserStateEvent** as the value
 
 ```kotlin
 class UserStateStream(
@@ -156,7 +156,7 @@ We can generate **completed** **UserStateEvents** straightaway once we receive t
 ```kotlin
 data class UserState(val userId: String = "", val tokens: List<Int> = emptyList()) {
   // ...
-  fun isCompleted() = tokens.containsAll(listOf(1, 2, 3, 4, 5))
+  val completed = tokens.containsAll(listOf(1, 2, 3, 4, 5))
 }
 
 class UserStateStream(
@@ -170,7 +170,7 @@ class UserStateStream(
       .mapValues { state ->
         logger.info("State $state")
         when {
-          state.isCompleted() -> UserStateEvent(state.userId, COMPLETED)
+          state.completed -> UserStateEvent(state.userId, COMPLETED)
           else -> null
         }
       }
@@ -211,9 +211,8 @@ class UserStateProcessor(
 Just apply the expiration logic this way:
 
 ```kotlin
-data class UserState(val userId: String = "", val tokens: List<Int> = emptyList(), private val expired: Boolean = false) {
+data class UserState(val userId: String = "", val tokens: List<Int> = emptyList(), val expired: Boolean = false) {
   // ...
-  fun isExpired() = expired
   fun expire() = UserState(userId, tokens, true)
 }
 
@@ -228,7 +227,7 @@ class UserStateProcessor(
       stateStore.all().forEachRemaining {
         val age = Duration.ofMillis(time - it.value.timestamp())
         if (age > expiration) {
-          if (it.value.value().isExpired()) {
+          if (it.value.value().expired) {
             // if it is already expired from a previous execution, we delete it
             logger.info("Delete ${it.key}")
             stateStore.delete(it.key)
@@ -264,9 +263,9 @@ class UserStateStream(
           // null states are sent downstream by UserStateProcessor when deleting entries from the store
           state == null -> null // "null" value generated by UserStateProcessor deleting values from the store
           // completed states are sent downstream from upstream
-          state.isCompleted() -> UserStateEvent(state.userId, COMPLETED)
+          state.completed -> UserStateEvent(state.userId, COMPLETED)
           // expired states are sent downstream by UserStateProcessor when updating entries from the store
-          state.isExpired() -> UserStateEvent(state.userId, EXPIRED)
+          state.expired -> UserStateEvent(state.userId, EXPIRED)
           else -> null
         }
       }
@@ -469,7 +468,7 @@ docker-compose up -d
 docker-compose down
 ```
 
-Then you can use [kcat](https://github.com/edenhill/kcat) (formerly know as **kafkacat**) to produce/consume to/from **Kafka**:
+Then you can use [kcat](https://github.com/edenhill/kcat) to produce/consume to/from **Kafka**:
 ```shell
 # consume
 kcat -b localhost:9094 -C -t pub.user.token -f '%k %s\n'

src/main/kotlin/com/rogervinas/kafkastreams/stream/UserState.kt

@@ -1,10 +1,8 @@
 package com.rogervinas.kafkastreams.stream
 
-data class UserState(val userId: String = "", val tokens: List<Int> = emptyList(), private val expired: Boolean = false) {
+data class UserState(val userId: String = "", val tokens: List<Int> = emptyList(), val expired: Boolean = false) {
 
-  fun isCompleted() = tokens.containsAll(listOf(1, 2, 3, 4, 5))
-
-  fun isExpired() = expired
+  val completed = tokens.containsAll(listOf(1, 2, 3, 4, 5))
 
   fun expire() = UserState(userId, tokens, true)
 

src/main/kotlin/com/rogervinas/kafkastreams/stream/UserStateStream.kt

@@ -47,8 +47,8 @@ class UserStateStream(
         logger.info("State $state")
         when {
           state == null -> null
-          state.isCompleted() -> UserStateEvent(state.userId, COMPLETED)
-          state.isExpired() -> UserStateEvent(state.userId, EXPIRED)
+          state.completed -> UserStateEvent(state.userId, COMPLETED)
+          state.expired -> UserStateEvent(state.userId, EXPIRED)
           else -> null
         }
       }
@@ -73,7 +73,7 @@ class UserStateProcessor(
       stateStore.all().forEachRemaining {
         val age = Duration.ofMillis(time - it.value.timestamp())
         if (age > expiration) {
-          if (it.value.value().isExpired()) {
+          if (it.value.value().expired) {
             logger.info("Delete ${it.key}")
             stateStore.delete(it.key)
           } else {