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

GatewayRequester

Browse source
This commit is contained in:
ialbert 2023-09-11 20:01:13 +03:00
parent 71cb78b72c
commit 6aa252f801
1 changed files with 9 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/impl/GatewayRequester.java
View file

@ -44,6 +44,7 @@ public class GatewayRequester extends QueueConsumer implements InitializingBean
private final Lock gatewayLock = new ReentrantLock(); private final Lock gatewayLock = new ReentrantLock();
private final Condition gatewayCondition = gatewayLock.newCondition(); private final Condition gatewayCondition = gatewayLock.newCondition();
private Long omtId = null; private Long omtId = null;
private SingleAssetResponse assetOperationApprovalRequest = null; private SingleAssetResponse assetOperationApprovalRequest = null;
@ -66,7 +67,7 @@ public class GatewayRequester extends QueueConsumer implements InitializingBean
} }
public Optional<Boolean> gatewayRequestAndWait(Registry om_t) { public Optional<Boolean> gatewayRequestAndWait(Registry om_t) {
boolean gatewayReceived = false; SingleAssetResponse responseReceived = null;
gatewayLock.lock(); gatewayLock.lock();
try { try {
if (this.omtId != null) { if (this.omtId != null) {
@ -89,16 +90,19 @@ public class GatewayRequester extends QueueConsumer implements InitializingBean
log.debug("sending om*t.id={} to gateway", om_t.getId()); log.debug("sending om*t.id={} to gateway", om_t.getId());
kafkaSender.sendRequestToQueue(Consts.ASSET_OPERATION, assetOperationListRequest); kafkaSender.sendRequestToQueue(Consts.ASSET_OPERATION, assetOperationListRequest);
this.omtId = om_t.getId(); this.omtId = om_t.getId();
gatewayReceived = gatewayCondition.await(GATEWAY_TIMEOUT, TimeUnit.SECONDS); boolean gatewayReceived = gatewayCondition.await(GATEWAY_TIMEOUT, TimeUnit.SECONDS);
if (gatewayReceived) {
responseReceived = assetOperationApprovalRequest;
assetOperationApprovalRequest = null;
}
} catch (InterruptedException e) { } catch (InterruptedException e) {
log.error(ExceptionUtils.getStackTrace(e)); log.error(ExceptionUtils.getStackTrace(e));
} finally { } finally {
gatewayLock.unlock(); gatewayLock.unlock();
} }
if (gatewayReceived) { if (responseReceived != null) {
boolean approved = assetOperationApprovalRequest.isApproved(); boolean approved = responseReceived.isApproved();
log.debug("om*t.id={} answer from gateway received. approval: {}", om_t.getId(), approved); log.debug("om*t.id={} answer from gateway received. approval: {}", om_t.getId(), approved);
assetOperationApprovalRequest = null;
return Optional.of(approved); return Optional.of(approved);
} else { } else {
log.error("om*t.id={} answer from gateway not received.", om_t.getId()); log.error("om*t.id={} answer from gateway not received.", om_t.getId());