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

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

Browse source
This commit is contained in:
ialbert 2022-08-17 19:20:55 +03:00
parent 52b928cca6
commit a127b85e12
18 changed files with 380 additions and 44 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/cud/CudController.java
View file

@ -31,7 +31,7 @@ public class CudController {
this.operator = operator;
}
@ApiOperation(value = "create/update/delete business objects. See meta.xml for field descriptions.")
@ApiOperation(value = "create business objects. See meta.xml for field descriptions.")
@ApiResponses(value = {@ApiResponse(code = 200, message = "OK", response = CudResponse.class), @ApiResponse(code = 400, message = "Ошибка валидации", response = BasicSpcexResponse.class)})
@RequestMapping(value = "/{destination}", method = RequestMethod.POST)
@ResponseBody
@ -45,13 +45,13 @@ public class CudController {
return processRequest(destination, body);
}
@ApiOperation(value = "create/update/delete business objects. See meta.xml for field descriptions.")
@ApiOperation(value = "update business objects. See meta.xml for field descriptions.")
@ApiResponses(value = {@ApiResponse(code = 200, message = "OK", response = CudResponse.class), @ApiResponse(code = 400, message = "Ошибка валидации", response = BasicSpcexResponse.class)})
@RequestMapping(value = "/{destination}", method = RequestMethod.PUT)
@ResponseBody
public CudResponse update(
@ApiParam(value = "Последняя часть URL определяет 'направление', по которому пойдет запрос. " +
"Должно биться с форматом запроса.", required = true, example = "money-market-security-new")
"Должно биться с форматом запроса.", required = true, example = "money-market-security-update")
@PathVariable("destination")
String destination,
@ApiParam(value = "Параметры команды в JSON формате, поля см. в meta.xml.", required = true)
@ -65,7 +65,7 @@ public class CudController {
@ResponseBody
public CudResponse delete(
@ApiParam(value = "Последняя часть URL определяет 'направление', по которому пойдет запрос. " +
"Должно биться с форматом запроса.", required = true, example = "money-market-security-new")
"Должно биться с форматом запроса.", required = true, example = "money-market-security-delete")
@PathVariable("destination")
String destination,
@ApiParam(value = "Параметры команды в JSON формате, поля см. в meta.xml.", required = true)

44
clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/request/cud/common/CommonDeleteAction.java Normal file
View file

@ -0,0 +1,44 @@
package ru.spcex.clearing.backendapi.controller.request.cud.common;
import com.fasterxml.jackson.annotation.JsonProperty;
import ru.spcex.clearing.backendapi.domain.actions.IAction;
import ru.spcex.clearing.backendapi.errors.BackEndError;
import ru.spcex.clearing.platform.messaging.domain.ActionType;
import ru.spcex.clearing.platform.messaging.domain.cud.common.CommonDeleteRequest;
import ru.spcex.platform.utils.enumeration.EnumMessage;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
public class CommonDeleteAction implements IAction<CommonDeleteRequest> {
@JsonProperty
public Long id;
@Override
public Collection<EnumMessage> validate() {
if (this.id == null)
return List.of(new EnumMessage(BackEndError.ValidationError, "id"));
else return Collections.emptyList();
}
@Override
public CommonDeleteRequest toRequest() {
var req = new CommonDeleteRequest();
req.setId(this.id);
return req;
}
@Override
public ActionType getActionType() {
return ActionType.DELETE;
}
public Long getId() {
return id;
}
public void setId(Long id) {
this.id = id;
}
}

8
clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/request/cud/MoneyMarketSecurityNewAction.java → clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/request/cud/securities/MoneyMarketSecurityNewAction.java
View file

@ -1,10 +1,11 @@
package ru.spcex.clearing.backendapi.controller.request.cud;
package ru.spcex.clearing.backendapi.controller.request.cud.securities;
import com.fasterxml.jackson.annotation.JsonFormat;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
import ru.spcex.clearing.backendapi.domain.actions.IAction;
import ru.spcex.clearing.backendapi.errors.BackEndError;
import ru.spcex.clearing.platform.messaging.domain.ActionType;
import ru.spcex.clearing.platform.messaging.domain.cud.securitites.MoneyMarketSecurityNewRequest;
import ru.spcex.clearing.platform.messaging.domain.json.deserialize.InstantDeserializer;
import ru.spcex.platform.utils.enumeration.EnumMessage;
@ -57,6 +58,11 @@ public class MoneyMarketSecurityNewAction implements IAction<MoneyMarketSecurity
return req;
}
@Override
public ActionType getActionType() {
return ActionType.NEW;
}
public Instant getStartDate() {
return startDate;
}

105
clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/request/cud/securities/MoneyMarketSecurityUpdateAction.java Normal file
View file

@ -0,0 +1,105 @@
package ru.spcex.clearing.backendapi.controller.request.cud.securities;
import com.fasterxml.jackson.annotation.JsonFormat;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
import ru.spcex.clearing.backendapi.domain.actions.IAction;
import ru.spcex.clearing.platform.messaging.domain.ActionType;
import ru.spcex.clearing.platform.messaging.domain.cud.securitites.MoneyMarketSecurityUpdateRequest;
import ru.spcex.clearing.platform.messaging.domain.json.deserialize.InstantDeserializer;
import java.time.Instant;
public class MoneyMarketSecurityUpdateAction implements IAction<MoneyMarketSecurityUpdateRequest> {
@JsonProperty
public Long id;
@JsonFormat(pattern = "yyyy-MM-dd", timezone = "Europe/Moscow")
@JsonDeserialize(using = InstantDeserializer.class)
@JsonProperty
public Instant endDate;
@JsonProperty
public Double nominalValue;
@JsonProperty
public Long nominalCurrencyId;
@JsonProperty
public String instrumentType;
@JsonProperty
public String fullName;
@JsonProperty
public Double lotSize;
@Override
public MoneyMarketSecurityUpdateRequest toRequest() {
var req = new MoneyMarketSecurityUpdateRequest();
req.setId(this.getId());
req.setEndDate(this.getEndDate());
req.setNominalValue(this.getNominalValue());
req.setNominalCurrencyId(this.getNominalCurrencyId());
req.setInstrumentType(this.getInstrumentType());
req.setFullName(this.getFullName());
req.setLotSize(this.getLotSize());
return req;
}
@Override
public ActionType getActionType() {
return ActionType.UPDATE;
}
public Instant getEndDate() {
return endDate;
}
public void setEndDate(Instant endDate) {
this.endDate = endDate;
}
public Double getNominalValue() {
return nominalValue;
}
public void setNominalValue(Double nominalValue) {
this.nominalValue = nominalValue;
}
public Long getNominalCurrencyId() {
return nominalCurrencyId;
}
public void setNominalCurrencyId(Long nominalCurrencyId) {
this.nominalCurrencyId = nominalCurrencyId;
}
public String getInstrumentType() {
return instrumentType;
}
public void setInstrumentType(String instrumentType) {
this.instrumentType = instrumentType;
}
public String getFullName() {
return fullName;
}
public void setFullName(String fullName) {
this.fullName = fullName;
}
public Double getLotSize() {
return lotSize;
}
public void setLotSize(Double lotSize) {
this.lotSize = lotSize;
}
public Long getId() {
return id;
}
public void setId(Long id) {
this.id = id;
}
}

5
clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/domain/actions/IAction.java
View file

@ -1,12 +1,13 @@
package ru.spcex.clearing.backendapi.domain.actions;
import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
import ru.spcex.clearing.platform.messaging.domain.ActionType;
import ru.spcex.platform.utils.enumeration.EnumMessage;
import java.util.Collection;
import java.util.Collections;
public interface IAction<T extends BaseRequest> {
public interface IAction<T> {
T toRequest();
ActionType getActionType();
default Collection<EnumMessage> validate() {return Collections.emptyList();}
}

9
clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/meta/CudMetaService.java
View file

@ -1,13 +1,18 @@
package ru.spcex.clearing.backendapi.meta;
import org.springframework.stereotype.Service;
import ru.spcex.clearing.backendapi.controller.request.cud.MoneyMarketSecurityNewAction;
import ru.spcex.clearing.backendapi.controller.request.cud.common.CommonDeleteAction;
import ru.spcex.clearing.backendapi.controller.request.cud.securities.MoneyMarketSecurityNewAction;
import ru.spcex.clearing.backendapi.controller.request.cud.securities.MoneyMarketSecurityUpdateAction;
import ru.spcex.clearing.backendapi.domain.actions.IAction;
import ru.spcex.clearing.platform.messaging.domain.Consts;
import java.util.HashMap;
import java.util.Map;
/**
* todo временная заглушка: нужно добавить парсинг meta.xml и получать инфу оттуда
*/
@Service
public class CudMetaService {
private final Map<String, Class<? extends IAction<?>>> mapping;
@ -15,6 +20,8 @@ public class CudMetaService {
public CudMetaService() {
this.mapping = new HashMap<>();
this.mapping.put(Consts.DESTINATION_MONEY_MARKET_SECURITY_NEW, MoneyMarketSecurityNewAction.class);
this.mapping.put(Consts.DESTINATION_MONEY_MARKET_SECURITY_UPDATE, MoneyMarketSecurityUpdateAction.class);
this.mapping.put(Consts.DESTINATION_MONEY_MARKET_SECURITY_DELETE, CommonDeleteAction.class);
}
public <T extends IAction<?>> Class<T> byDestination(String destination) {

4
clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/service/impl/OperatorImpl.java
View file

@ -30,8 +30,10 @@ public class OperatorImpl implements IOperator {
@Override
public QueueSuccessResponse sendRequestToQueue(String destination, IAction<?> iAction) throws ExecutionException, InterruptedException {
throwValidate(iAction);
BaseRequest request = iAction.toRequest();
BaseRequest<Object> request = new BaseRequest<>();
request.setId(idGenerator.nextId());
request.setActionType(iAction.getActionType());
request.setRequestPayload(iAction.toRequest());
Future<RecordMetadata> send = kafka.send(new ProducerRecord<>(destination, request));
send.get();
return new QueueSuccessResponse(request.getActionType(), request.getId());

31
clearing-parent/securities-service/src/main/java/ru/spcex/clearing/securities/service/cud/MoneyMarketSecurityService.java
View file

@ -8,8 +8,11 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import ru.clearing.classes.StaticData.Misc.MoneyMarketSecurity;
import ru.spcex.clearing.imdg.IMDGDistributedNames;
import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
import ru.spcex.clearing.platform.messaging.domain.Consts;
import ru.spcex.clearing.platform.messaging.domain.cud.common.CommonDeleteRequest;
import ru.spcex.clearing.platform.messaging.domain.cud.securitites.MoneyMarketSecurityNewRequest;
import ru.spcex.clearing.platform.messaging.domain.cud.securitites.MoneyMarketSecurityUpdateRequest;
import ru.spcex.clearing.platform.messaging.service.QueueConsumer;
import ru.spcex.platform.imdg.api.Imdg;
import ru.spcex.platform.imdg.api.ImdgProvider;
@ -31,10 +34,17 @@ public class MoneyMarketSecurityService extends QueueConsumer implements Initial
callback(MoneyMarketSecurityNewRequest.class)
.setConsumer(this::newMoneyMarket)
.forDestination(Consts.DESTINATION_MONEY_MARKET_SECURITY_NEW, callbacks::put);
callback(MoneyMarketSecurityUpdateRequest.class)
.setConsumer(this::updateMoneyMarket)
.forDestination(Consts.DESTINATION_MONEY_MARKET_SECURITY_UPDATE, callbacks::put);
callback(CommonDeleteRequest.class)
.setConsumer(this::deleteMoneyMarket)
.forDestination(Consts.DESTINATION_MONEY_MARKET_SECURITY_DELETE, callbacks::put);
init();
}
private void newMoneyMarket(MoneyMarketSecurityNewRequest req) {
private void newMoneyMarket(BaseRequest<MoneyMarketSecurityNewRequest> userRequest) {
MoneyMarketSecurityNewRequest req = userRequest.getRequestPayload();
log.debug("MoneyMarketSecurityNewRequest received");
MoneyMarketSecurity mms = new MoneyMarketSecurity();
mms.setStartDate(req.getStartDate());
@ -47,4 +57,23 @@ public class MoneyMarketSecurityService extends QueueConsumer implements Initial
moneyMarketSecurityMap.insert(mms);
log.debug("successfully processed, new id {}", mms.getId());
}
private void updateMoneyMarket(BaseRequest<MoneyMarketSecurityUpdateRequest> userRequest) {
MoneyMarketSecurityUpdateRequest req = userRequest.getRequestPayload();
log.debug("MoneyMarketSecurityUpdateRequest received id = {}", req.getId());
MoneyMarketSecurity mms = moneyMarketSecurityMap.getSingleObjectByID(req.getId());
mms.setEndDate(req.getEndDate());
mms.setNominalValue(req.getNominalValue() != null ? BigDecimal.valueOf(req.getNominalValue()) : null);
mms.setNominalCurrency(req.getNominalCurrencyId());
mms.setInstrumentType(req.getInstrumentType());
mms.setFullName(req.getFullName());
moneyMarketSecurityMap.update(mms);
}
private void deleteMoneyMarket(BaseRequest<CommonDeleteRequest> userRequest) {
CommonDeleteRequest req = userRequest.getRequestPayload();
log.debug("CommonDeleteRequest received id = {}", req.getId());
MoneyMarketSecurity mms = moneyMarketSecurityMap.getSingleObjectByID(req.getId());
moneyMarketSecurityMap.delete(mms);
}
}

15
platform-parent/platform-imdg-api-hazelcast-impl/src/main/java/ru/spcex/platform/imdg/iml/hazelcast/adapter/ImdgHazelcast.java
View file

@ -31,4 +31,19 @@ public class ImdgHazelcast<T extends SpcexObjectBase> implements Imdg<T> {
paramT.setId(idGenerator.newId());
map.put(paramT.getId(), paramT);
}
@Override
public void update(T paramT) {
map.put(paramT.getId(), paramT);
}
@Override
public void delete(T paramT) {
map.remove(paramT.getId());
}
@Override
public T getSingleObjectByID(Long paramLong) {
return map.get(paramLong);
}
}

2
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/ActionType.java
View file

@ -1,5 +1,5 @@
package ru.spcex.clearing.platform.messaging.domain;
public enum ActionType {
NEW
NEW, UPDATE, DELETE;
}

45
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/BaseRequest.java
View file

@ -1,8 +1,47 @@
package ru.spcex.clearing.platform.messaging.domain;
import com.fasterxml.jackson.annotation.JsonProperty;
import ru.spcex.platform.classes.base.interfaces.WithId;
public interface BaseRequest extends WithId {
void setId(Long id);
ActionType getActionType();
public class BaseRequest<T> implements WithId {
/**
* Идентификатор запроса
*/
@JsonProperty
private Long id;
/**
* Тип запроса
*/
@JsonProperty
private ActionType actionType;
/**
* Параметры запроса
*/
@JsonProperty
private T requestPayload;
@Override
public Long getId() {
return id;
}
public void setId(Long id) {
this.id = id;
}
public ActionType getActionType() {
return actionType;
}
public void setActionType(ActionType actionType) {
this.actionType = actionType;
}
public T getRequestPayload() {
return requestPayload;
}
public void setRequestPayload(T requestPayload) {
this.requestPayload = (T) requestPayload;
}
}

3
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java
View file

@ -2,5 +2,6 @@ package ru.spcex.clearing.platform.messaging.domain;
public interface Consts {
String DESTINATION_MONEY_MARKET_SECURITY_NEW = "money-market-security-new";
String DESTINATION_MONEY_MARKET_SECURITY_UPDATE = "money-market-security-update";
String DESTINATION_MONEY_MARKET_SECURITY_DELETE = "money-market-security-delete";
}

16
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/common/CommonDeleteRequest.java Normal file
View file

@ -0,0 +1,16 @@
package ru.spcex.clearing.platform.messaging.domain.cud.common;
import com.fasterxml.jackson.annotation.JsonProperty;
public class CommonDeleteRequest {
@JsonProperty
public Long id;
public Long getId() {
return id;
}
public void setId(Long id) {
this.id = id;
}
}

21
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/securitites/MoneyMarketSecurityNewRequest.java
View file

@ -3,16 +3,12 @@ package ru.spcex.clearing.platform.messaging.domain.cud.securitites;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
import com.fasterxml.jackson.databind.annotation.JsonSerialize;
import ru.spcex.clearing.platform.messaging.domain.ActionType;
import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
import ru.spcex.clearing.platform.messaging.domain.json.deserialize.InstantDeserializer;
import ru.spcex.clearing.platform.messaging.domain.json.serialize.InstantSerializer;
import java.time.Instant;
public class MoneyMarketSecurityNewRequest implements BaseRequest {
@JsonProperty
public Long id;
public class MoneyMarketSecurityNewRequest {
@JsonProperty
@JsonSerialize(using = InstantSerializer.class)
@JsonDeserialize(using = InstantDeserializer.class)
@ -34,21 +30,6 @@ public class MoneyMarketSecurityNewRequest implements BaseRequest {
@JsonProperty
public Double lotSize;
@Override
public Long getId() {
return id;
}
@Override
public ActionType getActionType() {
return ActionType.NEW;
}
@Override
public void setId(Long id) {
this.id = id;
}
public Instant getStartDate() {
return startDate;
}

86
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/securitites/MoneyMarketSecurityUpdateRequest.java Normal file
View file

@ -0,0 +1,86 @@
package ru.spcex.clearing.platform.messaging.domain.cud.securitites;
import com.fasterxml.jackson.annotation.JsonFormat;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
import com.fasterxml.jackson.databind.annotation.JsonSerialize;
import ru.spcex.clearing.platform.messaging.domain.json.deserialize.InstantDeserializer;
import ru.spcex.clearing.platform.messaging.domain.json.serialize.InstantSerializer;
import java.time.Instant;
public class MoneyMarketSecurityUpdateRequest {
@JsonProperty
public Long id;
@JsonFormat(pattern = "yyyy-MM-dd", timezone = "Europe/Moscow")
@JsonSerialize(using = InstantSerializer.class)
@JsonDeserialize(using = InstantDeserializer.class)
@JsonProperty
public Instant endDate;
@JsonProperty
public Double nominalValue;
@JsonProperty
public Long nominalCurrencyId;
@JsonProperty
public String instrumentType;
@JsonProperty
public String fullName;
@JsonProperty
public Double lotSize;
public Long getId() {
return id;
}
public void setId(Long id) {
this.id = id;
}
public Instant getEndDate() {
return endDate;
}
public void setEndDate(Instant endDate) {
this.endDate = endDate;
}
public Double getNominalValue() {
return nominalValue;
}
public void setNominalValue(Double nominalValue) {
this.nominalValue = nominalValue;
}
public Long getNominalCurrencyId() {
return nominalCurrencyId;
}
public void setNominalCurrencyId(Long nominalCurrencyId) {
this.nominalCurrencyId = nominalCurrencyId;
}
public String getInstrumentType() {
return instrumentType;
}
public void setInstrumentType(String instrumentType) {
this.instrumentType = instrumentType;
}
public String getFullName() {
return fullName;
}
public void setFullName(String fullName) {
this.fullName = fullName;
}
public Double getLotSize() {
return lotSize;
}
public void setLotSize(Double lotSize) {
this.lotSize = lotSize;
}
}

4
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/logic/functional/BuilderConsumerStep.java
View file

@ -1,7 +1,9 @@
package ru.spcex.clearing.platform.messaging.logic.functional;
import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
import java.util.function.Consumer;
public interface BuilderConsumerStep<T1> {
BuilderDestinationStep setConsumer(Consumer<T1> consumer);
BuilderDestinationStep setConsumer(Consumer<BaseRequest<T1>> consumer);
}

12
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/logic/functional/ConsumerSpecificClass.java
View file

@ -1,11 +1,13 @@
package ru.spcex.clearing.platform.messaging.logic.functional;
import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
public class ConsumerSpecificClass<T> implements BuilderConsumerStep<T>, BuilderDestinationStep {
private Class<T> clazz;
private java.util.function.Consumer<T> consumer;
private java.util.function.Consumer<BaseRequest<T>> consumer;
public ConsumerSpecificClass(Class<T> clazz) {
this.clazz = clazz;
@ -15,12 +17,8 @@ public class ConsumerSpecificClass<T> implements BuilderConsumerStep<T>, Builder
fun.accept(destination, this);
}
public void accept(T obj) {
this.consumer.accept(obj);
}
public void acceptRaw(Object obj) {
this.consumer.accept((T) obj);
this.consumer.accept((BaseRequest<T>) obj);
}
public Class<T> getClazz() {
@ -32,7 +30,7 @@ public class ConsumerSpecificClass<T> implements BuilderConsumerStep<T>, Builder
}
@Override
public BuilderDestinationStep setConsumer(Consumer<T> consumer) {
public BuilderDestinationStep setConsumer(Consumer<BaseRequest<T>> consumer) {
this.consumer = consumer;
return this;
}

6
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/QueueConsumer.java
View file

@ -1,5 +1,6 @@
package ru.spcex.clearing.platform.messaging.service;
import com.fasterxml.jackson.databind.JavaType;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
@ -7,6 +8,7 @@ import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.errors.WakeupException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
import ru.spcex.clearing.platform.messaging.logic.functional.BuilderConsumerStep;
import ru.spcex.clearing.platform.messaging.logic.functional.ConsumerSpecificClass;
import ru.spcex.platform.utils.log.ExceptionUtils;
@ -43,7 +45,9 @@ public class QueueConsumer implements AutoCloseable {
ConsumerRecords<String, Object> records = consumer.poll(Duration.of(10, ChronoUnit.SECONDS));
for (ConsumerRecord<String, Object> next : records) {
ConsumerSpecificClass<?> callback = callbacks.get(next.topic());
Object o = json.readValue((String) next.value(), callback.getClazz());
Class<?> clazz = callback.getClazz();
JavaType payloadType = json.getTypeFactory().constructParametricType(BaseRequest.class, clazz);
Object o = json.readValue((String) next.value(), payloadType);
callback.acceptRaw(o);
}
} catch (Throwable e) {