Tests loading from cache store using functional API

FunctionalCachestoreTest.java

package org.infinispan.functional;

import org.infinispan.Cache;
import org.infinispan.commons.api.functional.EntryView.ReadWriteEntryView;
import org.infinispan.commons.api.functional.FunctionalMap.ReadWriteMap;
import org.infinispan.configuration.cache.CacheMode;
import org.infinispan.configuration.cache.ConfigurationBuilder;
import org.infinispan.functional.impl.ReadWriteMapImpl;
import org.infinispan.persistence.dummy.DummyInMemoryStore;
import org.infinispan.persistence.dummy.DummyInMemoryStoreConfigurationBuilder;
import org.infinispan.persistence.manager.PersistenceManager;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;

import java.io.Serializable;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.function.Function;

import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertTrue;

/**
 * @author Radim Vansa <rvansa@redhat.com>
 */
public class FunctionalCachestoreTest extends AbstractFunctionalTest {
   private ReadWriteMap<Object, String> rw;

   @Override
   protected void createCacheManagers() throws Throwable {
      // Create local caches as default in a cluster of 2
      createClusteredCaches(2, new ConfigurationBuilder());
      // Create distributed caches
      ConfigurationBuilder distBuilder = new ConfigurationBuilder();
      distBuilder.clustering().cacheMode(CacheMode.DIST_SYNC).hash().numOwners(1)
                 .persistence().addStore(DummyInMemoryStoreConfigurationBuilder.class);
      cacheManagers.stream().forEach(cm -> cm.defineConfiguration(DIST, distBuilder.build()));
      // Create replicated caches
      ConfigurationBuilder replBuilder = new ConfigurationBuilder();
      replBuilder.clustering().cacheMode(CacheMode.REPL_SYNC)
                 .persistence().addStore(DummyInMemoryStoreConfigurationBuilder.class);
      cacheManagers.stream().forEach(cm -> cm.defineConfiguration(REPL, replBuilder.build()));
      // Wait for cluster to form
      waitForClusterToForm(DIST, REPL);
   }

   @Override
   @BeforeClass
   public void createBeforeClass() throws Throwable {
      super.createBeforeClass();
      rw = ReadWriteMapImpl.create(fmapD1);
   }

   @Test
   public void testLoadSame() throws ExecutionException, InterruptedException {
      doTest(0);
   }

   @Test
   public void testLoadRemote() throws ExecutionException, InterruptedException {
      doTest(1);
   }

   private void doTest(int targetNode) throws InterruptedException, ExecutionException {
      Cache<Object, String> cache = cache(targetNode, DIST);
      Object key = getKeyForCache(targetNode, DIST);
      rw.eval(key, (Function<ReadWriteEntryView<Object, String>, Object> & Serializable) view -> {
         assertFalse(view.find().isPresent());
         view.set("value");
         return null;
      }).get();
      assertEquals(cache.get(key), "value");
      cache.evict(key);
      assertFalse(cache.getAdvancedCache().getDataContainer().containsKey(key));
      Set<DummyInMemoryStore> stores = cache.getAdvancedCache().getComponentRegistry().getComponent(PersistenceManager.class).getStores(DummyInMemoryStore.class);
      DummyInMemoryStore store = stores.iterator().next();
      assertTrue(store.contains(key));
      rw.eval(key, (Function<ReadWriteEntryView<Object, String>, Object> & Serializable) view -> {
         assertEquals(view.get(), "value");
         assertTrue(view.find().isPresent());
         return null;
      }).get();
   }
}