-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Use Kafka Streams and Infinispan to build a local materialized view o…
…f Responders.
- Loading branch information
Showing
17 changed files
with
450 additions
and
79 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
55 changes: 0 additions & 55 deletions
55
src/main/java/com/redhat/emergency/response/responder/simulator/ResponderService.java
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
2 changes: 1 addition & 1 deletion
2
...r/simulator/repository/Configuration.java → ...r/simulator/infinispan/Configuration.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
4 changes: 4 additions & 0 deletions
4
...com/redhat/emergency/response/responder/simulator/streams/ResponderNotFoundException.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,4 @@ | ||
| package com.redhat.emergency.response.responder.simulator.streams; | ||
|
|
||
| public class ResponderNotFoundException extends RuntimeException { | ||
| } |
55 changes: 55 additions & 0 deletions
55
...main/java/com/redhat/emergency/response/responder/simulator/streams/ResponderService.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,55 @@ | ||
| package com.redhat.emergency.response.responder.simulator.streams; | ||
|
|
||
| import java.time.Duration; | ||
| import java.util.function.Supplier; | ||
| import javax.enterprise.context.ApplicationScoped; | ||
| import javax.inject.Inject; | ||
|
|
||
| import io.smallrye.mutiny.Uni; | ||
| import io.vertx.core.json.JsonObject; | ||
| import org.apache.kafka.streams.KafkaStreams; | ||
| import org.apache.kafka.streams.StoreQueryParameters; | ||
| import org.apache.kafka.streams.errors.InvalidStateStoreException; | ||
| import org.apache.kafka.streams.state.QueryableStoreTypes; | ||
| import org.apache.kafka.streams.state.ReadOnlyKeyValueStore; | ||
| import org.eclipse.microprofile.config.inject.ConfigProperty; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| @ApplicationScoped | ||
| public class ResponderService { | ||
|
|
||
| private static final Logger log = LoggerFactory.getLogger(ResponderService.class); | ||
|
|
||
| @Inject | ||
| KafkaStreams streams; | ||
|
|
||
| @ConfigProperty(name = "infinispan.streams.store", defaultValue = "responder-store") | ||
| String storeName; | ||
|
|
||
| public Uni<JsonObject> responder(String id) { | ||
| return Uni.createFrom().item(() -> doGetResponder(id)) | ||
| .onFailure(ResponderNotFoundException.class).retry().withBackOff(Duration.ofMillis(500), Duration.ofMillis(1000)).atMost(10) | ||
| .onFailure().recoverWithItem((Supplier<JsonObject>) JsonObject::new); | ||
| } | ||
|
|
||
| private JsonObject doGetResponder(String id) { | ||
| String responderStr = responderStore().get(id); | ||
| if (responderStr == null) { | ||
| log.warn("Responder with id " + id + " not found in the responder store"); | ||
| throw new ResponderNotFoundException(); | ||
| } | ||
| return new JsonObject(responderStr); | ||
| } | ||
|
|
||
| private ReadOnlyKeyValueStore<String, String> responderStore() { | ||
| while (true) { | ||
| try { | ||
| return streams.store(StoreQueryParameters.fromNameAndType(storeName, QueryableStoreTypes.keyValueStore())); | ||
| } catch (InvalidStateStoreException e) { | ||
| // ignore, store not ready yet | ||
| } | ||
| } | ||
| } | ||
|
|
||
| } |
104 changes: 104 additions & 0 deletions
104
...main/java/com/redhat/emergency/response/responder/simulator/streams/TopologyProducer.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,104 @@ | ||
| package com.redhat.emergency.response.responder.simulator.streams; | ||
|
|
||
| import java.util.Arrays; | ||
| import java.util.concurrent.ThreadLocalRandom; | ||
| import java.util.stream.Collectors; | ||
| import javax.enterprise.context.ApplicationScoped; | ||
| import javax.enterprise.inject.Produces; | ||
| import javax.inject.Inject; | ||
|
|
||
| import com.redhat.emergency.response.responder.simulator.streams.infinispan.InfinispanKeyValueStoreSupplier; | ||
| import io.vertx.core.json.JsonArray; | ||
| import io.vertx.core.json.JsonObject; | ||
| import org.apache.kafka.common.serialization.Serdes; | ||
| import org.apache.kafka.streams.KeyValue; | ||
| import org.apache.kafka.streams.StreamsBuilder; | ||
| import org.apache.kafka.streams.Topology; | ||
| import org.apache.kafka.streams.kstream.Consumed; | ||
| import org.apache.kafka.streams.kstream.KeyValueMapper; | ||
| import org.apache.kafka.streams.kstream.Materialized; | ||
| import org.apache.kafka.streams.kstream.ValueMapper; | ||
| import org.eclipse.microprofile.config.inject.ConfigProperty; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| @ApplicationScoped | ||
| public class TopologyProducer { | ||
|
|
||
| private static final Logger log = LoggerFactory.getLogger(TopologyProducer.class); | ||
|
|
||
| private static final String RESPONDERS_CREATED_EVENT = "RespondersCreatedEvent"; | ||
| private static final String RESPONDERS_DELETED_EVENT = "RespondersDeletedEvent"; | ||
| private static final String[] ACCEPTED_MESSAGE_TYPES = {RESPONDERS_CREATED_EVENT, RESPONDERS_DELETED_EVENT}; | ||
|
|
||
| private static final String ACTION_CREATE = "create"; | ||
| private static final String ACTION_DELETE = "delete"; | ||
|
|
||
|
|
||
| @ConfigProperty(name = "kafka.topic.responder-event") | ||
| String responderEventTopic; | ||
|
|
||
| @ConfigProperty(name = "simulator.distance.base") | ||
| double baseDistance; | ||
|
|
||
| @ConfigProperty(name = "simulator.distance.variation") | ||
| double distanceVariation; | ||
|
|
||
| @Inject | ||
| InfinispanKeyValueStoreSupplier keyValueStoreSupplier; | ||
|
|
||
| @Produces | ||
| public Topology buildTopology() { | ||
|
|
||
| StreamsBuilder builder = new StreamsBuilder(); | ||
|
|
||
| builder.stream(responderEventTopic, Consumed.with(Serdes.String(), Serdes.String())) | ||
| .mapValues(value -> { | ||
| try { | ||
| JsonObject json = new JsonObject(value); | ||
| log.debug("Processing message: " + value); | ||
| return json; | ||
| } catch (Exception e) { | ||
| log.warn("Unexpected message which is not a valid JSON object"); | ||
| return new JsonObject(); | ||
| } | ||
| }) | ||
| .filter((key, value) -> { | ||
| String messageType = value.getString("messageType"); | ||
| if (Arrays.asList(ACCEPTED_MESSAGE_TYPES).contains(messageType)) { | ||
| return true; | ||
| } | ||
| log.debug("Message with type '" + messageType + "' is ignored"); | ||
| return false; | ||
| }) | ||
| .flatMapValues((ValueMapper<JsonObject, Iterable<JsonObject>>) value -> { | ||
| if (RESPONDERS_CREATED_EVENT.equals(value.getString("messageType"))) { | ||
| JsonArray responders = value.getJsonObject("body").getJsonArray("responders"); | ||
| return responders.stream().map(o -> (JsonObject)o).map(resp -> new JsonObject().put("action", ACTION_CREATE) | ||
| .put("id", resp.getString("id")) | ||
| .put("responder", resp.put("distanceUnit", distanceUnit()))) | ||
| .collect(Collectors.toList()); | ||
| } else { | ||
| JsonArray responderIds = value.getJsonObject("body").getJsonArray("responders"); | ||
| return responderIds.stream().map(id -> new JsonObject().put("action", ACTION_DELETE) | ||
| .put("id", id)).collect(Collectors.toList()); | ||
| } | ||
| }) | ||
| .map((KeyValueMapper<String, JsonObject, KeyValue<String, String>>) (key, value) -> { | ||
| if (ACTION_CREATE.equalsIgnoreCase(value.getString("action"))) { | ||
| return new KeyValue<>(value.getString("id"), value.getJsonObject("responder").encode()); | ||
| } else { | ||
| return new KeyValue<>(value.getString("id"), null); | ||
| } | ||
| }) | ||
| .toTable(Materialized.<String, String>as(keyValueStoreSupplier).withKeySerde(Serdes.String()).withValueSerde(Serdes.String())); | ||
|
|
||
| return builder.build(); | ||
| } | ||
|
|
||
| private double distanceUnit() { | ||
| double max = baseDistance * (1 + distanceVariation); | ||
| double min = baseDistance * (1 - distanceVariation); | ||
| return ThreadLocalRandom.current().nextDouble(max - min) + min; | ||
| } | ||
| } |
Oops, something went wrong.