InfinispanStorage.java

import org.infinispan.Cache;
import org.infinispan.configuration.cache.CacheMode;
import org.infinispan.configuration.cache.Configuration;
import org.infinispan.configuration.cache.ConfigurationBuilder;
import org.infinispan.configuration.cache.FileCacheStoreConfigurationBuilder;
import org.infinispan.configuration.global.GlobalConfigurationBuilder;
import org.infinispan.eviction.EvictionStrategy;
import org.infinispan.io.GridFile;
import org.infinispan.io.GridFilesystem;
import org.infinispan.manager.DefaultCacheManager;
import org.infinispan.manager.EmbeddedCacheManager;
import org.infinispan.transaction.TransactionMode;

import java.io.*;
import java.util.List;
import java.util.Map;

public class InfinispanStorage implements INimbusStorage {
    private GridFilesystem fs;

    private static Configuration getCacheConfiguration(String location) {
        return new ConfigurationBuilder()
                .transaction().transactionMode(TransactionMode.TRANSACTIONAL).autoCommit(true)
                .eviction().strategy(EvictionStrategy.LRU).maxEntries(1000)
                .clustering().cacheMode(CacheMode.DIST_SYNC)
                .loaders().addFileCacheStore().fetchPersistentState(true).
                ignoreModifications(false).purgeOnStartup(false).location(location).fsyncMode(FileCacheStoreConfigurationBuilder.FsyncMode.PER_WRITE)
                .build();
    }

    public static void main(String... args) throws IOException {

        EmbeddedCacheManager manager = new DefaultCacheManager(
                GlobalConfigurationBuilder.defaultClusteredBuilder()
                        .transport().defaultTransport()
                        .build());

        manager.defineConfiguration("nimbus-data", getCacheConfiguration("./storm-ha-data"));
        manager.defineConfiguration("nimbus-meta", getCacheConfiguration("./storm-ha-meta"));

        Cache<String, byte[]> data = manager.getCache("nimbus-data");
        Cache<String, GridFile.Metadata> metadata = manager.getCache("nimbus-meta");

        InfinispanStorage storage = new InfinispanStorage();
        storage.fs = new GridFilesystem(data, metadata);

        String filePath = "/my/file.txt";

        storage.mkdirs("/my");

        DataOutputStream os = new DataOutputStream(storage.create(filePath));
        os.writeUTF("Hello");
        os.close();

        DataInputStream is = new DataInputStream(storage.open(filePath));
        System.out.println("READ : " + is.readUTF());
        is.close();
    }

    @Override
    public void init(Map conf) {

    }

    @Override
    public InputStream open(String path) {
        try {
            return fs.getInput(path);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public OutputStream create(String path) {
        try {
            return fs.getOutput(path);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public List<String> list(String path) {
        return null;  //To change body of implemented methods use File | Settings | File Templates.
    }

    @Override
    public void delete(String path) {
        //To change body of implemented methods use File | Settings | File Templates.
    }

    @Override
    public void mkdirs(String path) {
        fs.getFile(path).mkdirs();
    }

    @Override
    public boolean isSupportDistributed() {
        return false;  //To change body of implemented methods use File | Settings | File Templates.
    }
}