Skip to content

Instantly share code, notes, and snippets.

Avatar

Diogo Soares Ferreira diogodanielsoaresferreira

View GitHub Profile
View CamelRoutes.java
//...
rest("/api").get("/hello").to("direct:hello");
from("direct:hello").transform().constant("Hello world!");
//...
View CamelRoutes.java
//...
errorHandler(deadLetterChannel("kafka:dead_letter?brokers=localhost:9092")
.useOriginalMessage());
//...
View CamelRoutes.java
//...
from("timer://foo?period=5000").log("timer");
//...
View CamelRoutes.java
public class CamelRoutes extends EndpointRouteBuilder {
public void configure() {
from("kafka:input_topic?brokers=localhost:9092&groupId=group1")
.unmarshal().json(UserMessage.class)
.filter(simple("${body.type} == 'chat'"))
.to("seda:incoming_event");
from("seda:incoming_event?multipleConsumers=true")
View CamelRoutes.java
//...
from("kafka:input_topic?brokers=localhost:9092&groupId=group1")
.unmarshal().json(UserMessage.class)
.filter(simple("${body.type} == 'chat'"))
.log("${body}")
.split(simple("${body.devices}"))
.log("${body}")
.end()
.marshal().json()
.to("kafka:output_topic?brokers=localhost:9092")
View CamelRoutes.java
from("kafka:input_topic?brokers=localhost:9092&groupId=group2")
.unmarshal().json(UserMessage.class)
.aggregate(simple("${body.emitter}"), new CombinedUserMessagesAggregationStrategy())
.completionInterval(5000)
.bean(NLPUtils.class, "createUserMessages")
.log("${body}");
View NLPUtils.java
public class NLPUtils {
public static UserMessages createUserMessages(final CombinedUserMessage event) {
return UserMessages.builder()
.emitter(event.getEmitter())
.text(String.join(". ", event.getText()))
.build();
}
}
View CamelRoutes.java
from("kafka:input_topic?brokers=localhost:9092&groupId=group2")
.unmarshal().json(UserMessage.class)
.aggregate(simple("${body.emitter}"), new CombinedUserMessagesAggregationStrategy())
.completionInterval(5000)
.bean(NLPUtils.class, "createTranscription")
.log("${body}");
View CombinedUserMessagesAggregationStrategy.java
public class CombinedUserMessagesAggregationStrategy implements AggregationStrategy {
@Override
public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
if (oldExchange == null) {
final UserMessage body = newExchange.getIn().getBody(UserMessage.class);
final CombinedUserMessage newEventBody = CombinedUserMessage.builder()
.emitter(body.getEmitter())
.text(List.of(body.getText()))
.build();
newExchange.getIn().setBody(newEventBody);
View CamelRoutes.java
//...
from("kafka:input_topic?brokers=localhost:9092&groupId=group2")
.unmarshal().json(UserMessage.class)
.aggregate(simple("${body.emitter}"), new CombinedUserMessagesAggregationStrategy())
.completionInterval(5000)
.log("${body}");
//...