App.java

package com.github.eutkin;

import org.apache.ignite.Ignite;
import org.apache.ignite.Ignition;
import org.apache.ignite.cache.CacheEntryProcessor;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.lang.IgniteFuture;
import org.apache.ignite.lang.IgniteInClosure;

import javax.cache.processor.EntryProcessorException;
import javax.cache.processor.MutableEntry;
import java.math.BigDecimal;
import java.time.OffsetDateTime;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

public class App {

    public static void main(String[] args) throws InterruptedException {
        Ignite ign = Ignition.start(new IgniteConfiguration().setPeerClassLoadingEnabled(true).setClientMode(true));

        final var processor = new CacheEntryProcessor<String, List<Map<String, Object>>, Long>() {

            @Override
            public Long process(MutableEntry<String, List<Map<String, Object>>> entry, Object... arg) throws EntryProcessorException {
                final var count = (int) arg[0];
                entry.setValue(IntStream.range(0, count).mapToObj(v -> Map.<String, Object>of(
                        "id", UUID.randomUUID().toString(),
                        "timestamp", OffsetDateTime.now(),
                        "amount", BigDecimal.valueOf(10000)
                )).collect(Collectors.toList()));
                return (Long) arg[1];
            }
        };

        try (ign) {
            final var counter = new AtomicLong();
            final var cache = ign
                    .<String, List<Map<String, Object>>>getOrCreateCache("biplanCacheTestCache");
            final var latch = new CountDownLatch(80);
            IntStream.range(0, 80)
                    .forEach(v -> {
                        cache
                                .invokeAsync("consumer-" + counter.incrementAndGet(), processor, 5000,
                                        System.currentTimeMillis())
                                .listen(f -> {
                                    final var start = f.get();
                                    System.out.println((System.currentTimeMillis() - start) + " ms");
                                    latch.countDown();
                                });
                    });
            latch.await();
            ign.destroyCache("biplanCacheTestCache");
        }

    }
}