import de.juplo.kafka.wordcount.counter.TestCounter;
import de.juplo.kafka.wordcount.counter.TestWord;
-import org.apache.kafka.common.header.Header;
-import org.apache.kafka.common.header.Headers;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
-import java.util.Map;
-import java.util.Properties;
-import java.util.stream.Collectors;
import java.util.stream.Stream;
import static org.assertj.core.api.Assertions.assertThat;
assertRankingEqualsRankingFromLastMessage(KLAUS, store.get(KLAUS));
}
+ static void assertExpectedNumberOfMessagesForUsers(MultiValueMap<User, Ranking> receivedMessages)
+ {
+ assertThat(countMessagesForUser(PETER, receivedMessages));
+ assertThat(countMessagesForUser(KLAUS, receivedMessages));
+ }
+
+ static int countMessagesForUser(User user, MultiValueMap<User, Ranking> messagesForUsers)
+ {
+ return messagesForUsers.get(user).size();
+ }
+
+
static void assertExpectedLastMessagesForUsers(MultiValueMap<User, Ranking> receivedMessages)
{
assertRankingEqualsRankingFromLastMessage(PETER, getLastMessageFor(PETER, receivedMessages));
.forEach(keyValue -> expectedMessages.add(keyValue.key, keyValue.value));
return expectedMessages;
}
-
- static Map<String, Object> convertToMap(Properties properties)
- {
- return properties
- .entrySet()
- .stream()
- .collect(
- Collectors.toMap(
- entry -> (String)entry.getKey(),
- entry -> entry.getValue()
- ));
- }
-
- static String parseHeader(Headers headers, String key)
- {
- Header header = headers.lastHeader(key);
- if (header == null)
- {
- return key + "=null";
- }
- else
- {
- return key + "=" + new String(header.value());
- }
- }
}