diff --git a/clearing-parent/balance-service/src/test/java/ru/spcex/clearing/balance/service/Sdf08ServiceTest.java b/clearing-parent/balance-service/src/test/java/ru/spcex/clearing/balance/service/Sdf08ServiceTest.java index eb34fbbd0..0da633a66 100644 --- a/clearing-parent/balance-service/src/test/java/ru/spcex/clearing/balance/service/Sdf08ServiceTest.java +++ b/clearing-parent/balance-service/src/test/java/ru/spcex/clearing/balance/service/Sdf08ServiceTest.java @@ -1,22 +1,33 @@ package ru.spcex.clearing.balance.service; import com.hazelcast.core.IMap; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.TopicPartition; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; +import ru.clearing.classes.statics.data.account.BankAccount; import ru.clearing.classes.statics.data.sdf.SDf08; import ru.clearing.classes.statics.data.statement.Statement; +import ru.spcex.clearing.balance.utils.ImapEvent; import ru.spcex.clearing.balance.utils.MatcherFactory; +import ru.spcex.clearing.imdg.IMDGDistributedNames; import ru.spcex.clearing.platform.messaging.domain.cud.balance.StatementRequest; import javax.annotation.PostConstruct; import java.time.Instant; import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; import static ru.spcex.clearing.balance.utils.MatcherFactory.usingIgnoringFieldsComparator; class Sdf08ServiceTest extends AbstractServiceTest { private static final MatcherFactory.Matcher SDF_08_MATCHER = usingIgnoringFieldsComparator("created", "comment", "outSDfId", "generationTime", "generationId", "id"); + @Autowired + private Consumer mockConsumer; + @Autowired Sdf08Service sdf08Service; @@ -37,7 +48,7 @@ class Sdf08ServiceTest extends AbstractServiceTest { * {@link SDf08#generationTime} - текущее время
*/ @Test - void newSDf08() { + void newSDf08() throws InterruptedException { IMap sdf08Map = sdf08Imdg.getMap(); SDf08 sDf08 = new SDf08(); sDf08.setNumber(idGenerator.nextId().toString()); @@ -46,5 +57,24 @@ class Sdf08ServiceTest extends AbstractServiceTest { sDf08.setGenerationTime(now); sDf08.setGenerationId(idGenerator.nextId()); sdf08Imdg.insert(sDf08); + + //KAFKA + mockConsumer.schedulePollTask(() -> { + mockConsumer.rebalance(Collections.singletonList(new TopicPartition(TOPIC_ACCOUNT_NEW, PARTITION))); + mockConsumer.addRecord(new ConsumerRecord<>(TOPIC_ACCOUNT_NEW, PARTITION, 0, "key", jsonBaseNewRequest)); + }); + + + HashMap startOffsets = new HashMap<>(); + TopicPartition tp = new TopicPartition(TOPIC_ACCOUNT_NEW, PARTITION); + startOffsets.put(tp, 0L); + mockConsumer.updateBeginningOffsets(startOffsets); + + IMap iMap = hazelcastServiceTest.getHazelcast().getMap(IMDGDistributedNames.Map_BankAccount); + + //waiting for hazelcast map item updates + ImapEvent imapEvent = new ImapEvent(iMap); + imapEvent.waitWhenHappened(); + } } \ No newline at end of file diff --git a/clearing-parent/balance-service/src/test/java/ru/spcex/clearing/balance/utils/ImapEvent.java b/clearing-parent/balance-service/src/test/java/ru/spcex/clearing/balance/utils/ImapEvent.java new file mode 100644 index 000000000..a9e80da03 --- /dev/null +++ b/clearing-parent/balance-service/src/test/java/ru/spcex/clearing/balance/utils/ImapEvent.java @@ -0,0 +1,63 @@ +package ru.spcex.clearing.balance.utils; + +import com.hazelcast.core.IMap; +import com.hazelcast.map.listener.EntryAddedListener; +import com.hazelcast.map.listener.EntryRemovedListener; +import com.hazelcast.map.listener.EntryUpdatedListener; + +import java.util.Timer; +import java.util.TimerTask; +import java.util.concurrent.atomic.AtomicBoolean; + +public class ImapEvent { + private final IMap iMap; + private final String listenerAdding; + private final String listenerUpdating; + private final String listenerRemoving; + private final AtomicBoolean checkEventHappened = new AtomicBoolean(false); + + public ImapEvent(IMap iMap) { + this.iMap = iMap; + listenerAdding = iMap.addEntryListener((EntryAddedListener) entryEvent -> { + synchronized (checkEventHappened) { + checkEventHappened.set(true); + checkEventHappened.notify(); + } + }, false); + listenerUpdating = iMap.addEntryListener((EntryUpdatedListener) entryEvent -> { + synchronized (checkEventHappened) { + checkEventHappened.set(true); + checkEventHappened.notify(); + } + }, false); + listenerRemoving = iMap.addEntryListener((EntryRemovedListener) entryEvent -> { + synchronized (checkEventHappened) { + checkEventHappened.set(true); + checkEventHappened.notify(); + } + }, false); + } + + public void waitWhenHappened() throws InterruptedException { + //running timer task as daemon thread + Timer timer = new Timer(true); + timer.scheduleAtFixedRate(new TimerTask() { + boolean secondRan; + + @Override + public void run() { + checkEventHappened.set(secondRan);//если что-то пойдет не так не тормозить основной поток + secondRan = true; + } + }, 0, 60 * 1000); + synchronized (checkEventHappened) { + while (!checkEventHappened.get()) { + checkEventHappened.wait(100); + } + } + //preparing hazelcastImdgProvider for next test + iMap.removeEntryListener(listenerAdding); + iMap.removeEntryListener(listenerUpdating); + iMap.removeEntryListener(listenerRemoving); + } +}