andreacavalli
/
tdlib-session-container
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Code cleanup

This commit is contained in:
Andrea Cavalli 2022-09-22 16:26:55 +02:00
parent b322400d87
commit 2b2e690da4
3 changed files with 10 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
src/main/java/it/tdlight/reactiveapi/KafkaClientBoundConsumer.java
View File

@ -4,6 +4,7 @@ import it.tdlight.reactiveapi.Event.ClientBoundEvent;
public class KafkaClientBoundConsumer extends KafkaConsumer<ClientBoundEvent> { public class KafkaClientBoundConsumer extends KafkaConsumer<ClientBoundEvent> {
private static final KafkaChannelCodec CODEC = KafkaChannelCodec.CLIENT_BOUND_EVENT;
private final String lane; private final String lane;
private final String name; private final String name;
@ -11,15 +12,15 @@ public class KafkaClientBoundConsumer extends KafkaConsumer<ClientBoundEvent> {
super(kafkaParameters); super(kafkaParameters);
this.lane = lane; this.lane = lane;
if (lane.isBlank()) { if (lane.isBlank()) {
this.name = KafkaChannelCodec.CLIENT_BOUND_EVENT.getKafkaName(); this.name = CODEC.getKafkaName();
} else { } else {
this.name = KafkaChannelCodec.CLIENT_BOUND_EVENT.getKafkaName() + "-" + lane; this.name = CODEC.getKafkaName() + "-" + lane;
} }
} }
@Override @Override
public KafkaChannelCodec getChannelCodec() { public KafkaChannelCodec getChannelCodec() {
return KafkaChannelCodec.CLIENT_BOUND_EVENT; return CODEC;
} }
@Override @Override

8
src/main/java/it/tdlight/reactiveapi/KafkaClientBoundProducer.java
View File

@ -4,20 +4,22 @@ import it.tdlight.reactiveapi.Event.ClientBoundEvent;
public class KafkaClientBoundProducer extends KafkaProducer<ClientBoundEvent> { public class KafkaClientBoundProducer extends KafkaProducer<ClientBoundEvent> {
private static final KafkaChannelCodec CODEC = KafkaChannelCodec.CLIENT_BOUND_EVENT;
private final String name; private final String name;
public KafkaClientBoundProducer(KafkaParameters kafkaParameters, String lane) { public KafkaClientBoundProducer(KafkaParameters kafkaParameters, String lane) {
super(kafkaParameters); super(kafkaParameters);
if (lane.isBlank()) { if (lane.isBlank()) {
this.name = KafkaChannelCodec.CLIENT_BOUND_EVENT.getKafkaName(); this.name = CODEC.getKafkaName();
} else { } else {
this.name = KafkaChannelCodec.CLIENT_BOUND_EVENT.getKafkaName() + "-" + lane; this.name = CODEC.getKafkaName() + "-" + lane;
} }
} }
@Override @Override
public KafkaChannelCodec getChannelCodec() { public KafkaChannelCodec getChannelCodec() {
return KafkaChannelCodec.CLIENT_BOUND_EVENT; return CODEC;
} }
@Override @Override

2
src/main/java/it/tdlight/reactiveapi/KafkaProducer.java
View File

@ -44,7 +44,7 @@ public abstract class KafkaProducer<K> {
var channelName = getChannelName(); var channelName = getChannelName();
return eventsFlux return eventsFlux
.<SenderRecord<Integer, K, Integer>>map(event -> .<SenderRecord<Integer, K, Integer>>map(event ->
SenderRecord.create(new ProducerRecord<>(channelName, event), null)) SenderRecord.create(new ProducerRecord<>("tdlib." + channelName, event), null))
.log("produce-messages-" + channelName, .log("produce-messages-" + channelName,
Level.FINEST, Level.FINEST,
SignalType.REQUEST, SignalType.REQUEST,