RiakCacheStore.java

import java.util.Set;

import com.basho.riak.client.IRiakClient;
import com.basho.riak.client.IRiakObject;
import com.basho.riak.client.RiakException;
import com.basho.riak.client.RiakFactory;
import com.basho.riak.client.bucket.Bucket;
import com.tangosol.net.cache.BinaryEntryStore;
import com.tangosol.util.Base;
import com.tangosol.util.Binary;
import com.tangosol.util.BinaryEntry;

public class RiakCacheStore extends Base implements BinaryEntryStore {
	private IRiakClient riakClient;
	private Bucket myBucket;

	public RiakCacheStore(String address, int port, String bucketName) {
		System.out.println("RiakCacheStore.RiakCacheStore() - Connecting to Riak [" + address + "," + port + "]");
		try {
			riakClient = RiakFactory.pbcClient(address, port);
			myBucket = riakClient.createBucket(bucketName).execute();
		} catch (Exception e) {
			throw ensureRuntimeException(e);
		}
	}

	@Override
	public void erase(BinaryEntry bEntry) {
	}

	@Override
	public void eraseAll(Set arg0) {
	}

	@Override
	public void load(BinaryEntry bEntry) {
		String riakKey = new String(bEntry.getBinaryKey().toByteArray());
		System.out.println("RiakCacheStore.load() - from Riak to Coherence [key=" + riakKey +"]");
		try {
			IRiakObject myData = myBucket.fetch(riakKey).execute();
			bEntry.updateBinaryValue(new Binary(myData.getValue()));
		} catch (RiakException e) {
			throw ensureRuntimeException(e);
		}
	}

	@Override
	public void loadAll(Set arg0) {
	}

	@Override
	public void store(BinaryEntry bEntry){
		String riakKey = new String(bEntry.getBinaryKey().toByteArray());
		System.out.println("RiakCacheStore.store() - from Coherence to Riak [key=" + riakKey +"]");
		try {
			myBucket.store(riakKey, bEntry.getBinaryValue().toByteArray()).execute();
		} catch (RiakException e) {
			throw ensureRuntimeException(e);
		}
	}

	@Override
	public void storeAll(Set arg0) {
	}

}