treshenkov/clearing
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

http://jira.mfd.msk:8088/browse/CLS-277

Browse source
This commit is contained in:
ialbert 2023-05-03 17:31:27 +03:00
parent ed47b5291b
commit 246dc403af
1 changed files with 0 additions and 78 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

78
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/sender/KafkaSyncRequestReplySender.java
View file

@ -1,78 +0,0 @@
package ru.spcex.clearing.platform.messaging.service.sender;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.requestreply.ReplyingKafkaTemplate;
import ru.spcex.clearing.platform.messaging.domain.ActionType;
import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
import ru.spcex.clearing.platform.messaging.service.RequestInfo;
import ru.spcex.platform.classes.base.SpcexObjectBase;
import ru.spcex.platform.utils.log.ExceptionUtils;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.function.Function;
import java.util.function.Supplier;
public class KafkaSyncRequestReplySender {
private final Logger log = LoggerFactory.getLogger(getClass());
private final Map<String, KafkaImdgInsert> allImdgMaps;
private Producer<String, Object> kafka;
private Supplier<Long> idGenerator;
private Function<String, KafkaImdgInsert> imdgGenerator;
private ReplyingKafkaTemplate<String, Object, Object> kafkaTemplate;
KafkaSyncRequestReplySender() {
this.allImdgMaps = new ConcurrentHashMap<>();
}
public static KafkaSenderBuilderImpl setup() {
return new KafkaSenderBuilderImpl();
}
public Long sendRequestToQueue(String destination, Object requestPayload) {
BaseRequest<Object> request = new BaseRequest<>();
request.setId(idGenerator.get());
request.setActionType(ActionType.SYSTEM);
request.setRequestPayload(requestPayload);
//сохраняет данные о запросе в хранилище
saveRequestToStorage(destination, request);
Future<RecordMetadata> send = kafka.send(new ProducerRecord<>(destination, request));
try {
send.get();
} catch (InterruptedException | ExecutionException e) {
log.error(ExceptionUtils.getStackTrace(e));
return null;
}
return request.getId();
}
private void saveRequestToStorage(String destination, BaseRequest<Object> request) {
KafkaImdgInsert imdgInsert = getImdg(destination);
RequestInfo requestInfo = RequestInfo.create(request.getId());
imdgInsert.insert(requestInfo);
}
@SuppressWarnings("unchecked")
private <T extends SpcexObjectBase> KafkaImdgInsert getImdg(String mapName) {
return allImdgMaps.computeIfAbsent(mapName, (mapName1) -> imdgGenerator.apply(mapName));
}
void setProducer(Producer<String, Object> kafka) {
this.kafka = kafka;
}
void setIdGenerator(Supplier<Long> idGenerator) {
this.idGenerator = idGenerator;
}
void setImdgProvider(Function<String, KafkaImdgInsert> imdgGenerator) {
this.imdgGenerator = imdgGenerator;
}
}