Highly optimize topic name

This commit is contained in:
Andrea Cavalli 2022-01-23 21:57:43 +01:00
parent 344be2b320
commit aa887ba954
3 changed files with 19 additions and 3 deletions

View File

@ -34,10 +34,10 @@ public class KafkaProducer {
sender = KafkaSender.create(senderOptions.maxInFlight(1024)); sender = KafkaSender.create(senderOptions.maxInFlight(1024));
} }
public Mono<Void> sendMessages(long userId, Flux<ClientBoundEvent> eventsFlux) { public Mono<Void> sendMessages(UserTopic userId, Flux<ClientBoundEvent> eventsFlux) {
return eventsFlux return eventsFlux
.<SenderRecord<Integer, ClientBoundEvent, Integer>>map(event -> SenderRecord.create(new ProducerRecord<>( .<SenderRecord<Integer, ClientBoundEvent, Integer>>map(event -> SenderRecord.create(new ProducerRecord<>(
"tdlib.event.%d".formatted(userId), userId.getTopic(),
event event
), null)) ), null))
.transform(sender::send) .transform(sender::send)

View File

@ -75,6 +75,7 @@ public abstract class ReactiveApiPublisher {
private final AtomicReference<State> state = new AtomicReference<>(new State(LOGGED_OUT)); private final AtomicReference<State> state = new AtomicReference<>(new State(LOGGED_OUT));
protected final long userId; protected final long userId;
protected final UserTopic userTopic;
protected final long liveId; protected final long liveId;
private final String dynamicIdResolveSubject; private final String dynamicIdResolveSubject;
@ -90,6 +91,7 @@ public abstract class ReactiveApiPublisher {
this.eventService = atomix.getEventService(); this.eventService = atomix.getEventService();
this.resultingEventTransformerSet = resultingEventTransformerSet; this.resultingEventTransformerSet = resultingEventTransformerSet;
this.userId = userId; this.userId = userId;
this.userTopic = new UserTopic(userId);
this.liveId = liveId; this.liveId = liveId;
this.dynamicIdResolveSubject = SubjectNaming.getDynamicIdResolveSubject(userId); this.dynamicIdResolveSubject = SubjectNaming.getDynamicIdResolveSubject(userId);
this.rawTelegramClient = ClientManager.createReactive(); this.rawTelegramClient = ClientManager.createReactive();
@ -197,7 +199,7 @@ public abstract class ReactiveApiPublisher {
// Buffer requests to avoid halting the event loop // Buffer requests to avoid halting the event loop
.onBackpressureBuffer(); .onBackpressureBuffer();
kafkaProducer.sendMessages(userId, messagesToSend).subscribeOn(Schedulers.parallel()).subscribe(); kafkaProducer.sendMessages(userTopic, messagesToSend).subscribeOn(Schedulers.parallel()).subscribe();
publishedResultingEvents publishedResultingEvents
// Obtain only cluster-bound events // Obtain only cluster-bound events

View File

@ -0,0 +1,14 @@
package it.tdlight.reactiveapi;
public class UserTopic {
private final String value;
public UserTopic(long userId) {
value = "tdlib.event.%d".formatted(userId);
}
public String getTopic() {
return value;
}
}