simple cheng.

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<SDf08> SDF_08_MATCHER = usingIgnoringFieldsComparator("created", "comment", "outSDfId", "generationTime", "generationId", "id");
 
+    @Autowired
+    private Consumer<String, Object> mockConsumer;
+
     @Autowired
     Sdf08Service sdf08Service;
 
@@ -37,7 +48,7 @@ class Sdf08ServiceTest extends AbstractServiceTest {
      * {@link SDf08#generationTime} - текущее время<br>
      */
     @Test
-    void newSDf08() {
+    void newSDf08() throws InterruptedException {
         IMap<Long, SDf08> 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<TopicPartition, Long> startOffsets = new HashMap<>();
+        TopicPartition tp = new TopicPartition(TOPIC_ACCOUNT_NEW, PARTITION);
+        startOffsets.put(tp, 0L);
+        mockConsumer.updateBeginningOffsets(startOffsets);
+
+        IMap<Long, BankAccount> iMap = hazelcastServiceTest.getHazelcast().getMap(IMDGDistributedNames.Map_BankAccount);
+
+        //waiting for hazelcast map item updates
+        ImapEvent imapEvent = new ImapEvent(iMap);
+        imapEvent.waitWhenHappened();
+
     }
 }

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<T> {
+    private final IMap<Long, T> iMap;
+    private final String listenerAdding;
+    private final String listenerUpdating;
+    private final String listenerRemoving;
+    private final AtomicBoolean checkEventHappened = new AtomicBoolean(false);
+
+    public ImapEvent(IMap<Long, T> iMap) {
+        this.iMap = iMap;
+        listenerAdding = iMap.addEntryListener((EntryAddedListener<Long, T>) entryEvent -> {
+            synchronized (checkEventHappened) {
+                checkEventHappened.set(true);
+                checkEventHappened.notify();
+            }
+        }, false);
+        listenerUpdating = iMap.addEntryListener((EntryUpdatedListener<Long, T>) entryEvent -> {
+            synchronized (checkEventHappened) {
+                checkEventHappened.set(true);
+                checkEventHappened.notify();
+            }
+        }, false);
+        listenerRemoving = iMap.addEntryListener((EntryRemovedListener<Long, T>) 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);
+    }
+}