Skip to content
Permalink
Browse files

Manually porting many of @Cabeer's modeshape40 b/c kernel/kernel-impl…

… reorg made rebasing modeshape40 branch very messy
  • Loading branch information...
escowles committed Aug 6, 2014
1 parent ae12d03 commit 9915c5027e48aa87feddcdc87a084674e894dd7b
Showing with 397 additions and 802 deletions.
  1. +3 −3 fcrepo-connector-file/src/main/java/org/fcrepo/connector/file/FedoraFileSystemConnector.java
  2. +2 −2 fcrepo-connector-file/src/test/java/org/fcrepo/connector/file/FedoraFileSystemConnectorTest.java
  3. +5 −0 fcrepo-http-commons/pom.xml
  4. +8 −0 fcrepo-kernel-impl/pom.xml
  5. +0 −46 fcrepo-kernel-impl/src/main/java/org/fcrepo/kernel/impl/services/functions/GetCacheStore.java
  6. +21 −20 fcrepo-kernel-impl/src/main/java/org/fcrepo/kernel/impl/utils/impl/DistributedFixityCheck.java
  7. +7 −2 fcrepo-kernel-impl/src/main/java/org/fcrepo/kernel/impl/utils/impl/InfinispanCacheStoreEntry.java
  8. +200 −0 ...ernel-impl/src/main/java/org/fcrepo/kernel/impl/utils/infinispan/CacheLoaderChunkInputStream.java
  9. +0 −183 fcrepo-kernel-impl/src/main/java/org/fcrepo/kernel/impl/utils/infinispan/StoreChunkInputStream.java
  10. +0 −143 fcrepo-kernel-impl/src/main/java/org/fcrepo/kernel/impl/utils/infinispan/StoreChunkOutputStream.java
  11. +50 −0 fcrepo-kernel-impl/src/main/java/org/modeshape/jcr/value/binary/infinispan/ChunkBinaryMetadata.java
  12. +60 −0 fcrepo-kernel-impl/src/main/java/org/modeshape/jcr/value/binary/infinispan/InfinispanUtils.java
  13. +0 −64 ...pl/src/test/java/org/fcrepo/integration/kernel/impl/utils/infinispan/StoreChunkInputStreamIT.java
  14. +0 −77 fcrepo-kernel-impl/src/test/java/org/fcrepo/kernel/impl/services/functions/GetCacheStoreTest.java
  15. +23 −18 ...nel-impl/src/test/java/org/fcrepo/kernel/impl/services/functions/GetClusterConfigurationTest.java
  16. +0 −157 ...-kernel-impl/src/test/java/org/fcrepo/kernel/impl/utils/infinispan/StoreChunkInputStreamTest.java
  17. +0 −85 ...kernel-impl/src/test/java/org/fcrepo/kernel/impl/utils/infinispan/StoreChunkOutputStreamTest.java
  18. +18 −2 pom.xml
@@ -41,9 +41,9 @@
import org.modeshape.connector.filesystem.FileSystemConnector;
import org.modeshape.jcr.api.value.DateTime;
import org.modeshape.jcr.api.nodetype.NodeTypeManager;
import org.modeshape.jcr.federation.spi.DocumentChanges;
import org.modeshape.jcr.federation.spi.DocumentReader;
import org.modeshape.jcr.federation.spi.DocumentWriter;
import org.modeshape.jcr.spi.federation.DocumentChanges;
import org.modeshape.jcr.spi.federation.DocumentReader;
import org.modeshape.jcr.spi.federation.DocumentWriter;
import org.modeshape.jcr.value.BinaryValue;
import org.modeshape.jcr.value.Name;
import org.modeshape.jcr.value.Property;
@@ -24,8 +24,8 @@
import org.modeshape.jcr.ExecutionContext;
import org.modeshape.jcr.api.nodetype.NodeTypeManager;
import org.modeshape.jcr.cache.document.DocumentTranslator;
import org.modeshape.jcr.federation.spi.DocumentReader;
import org.modeshape.jcr.federation.spi.ExtraPropertiesStore;
import org.modeshape.jcr.spi.federation.DocumentReader;
import org.modeshape.jcr.spi.federation.ExtraPropertiesStore;
import org.modeshape.jcr.value.BinaryValue;
import org.modeshape.jcr.value.NameFactory;
import org.modeshape.jcr.value.Property;
@@ -76,6 +76,11 @@
<groupId>org.modeshape</groupId>
<artifactId>modeshape-jcr-api</artifactId>
</dependency>
<dependency>
<groupId>org.codehaus.jackson</groupId>
<artifactId>jackson-core-asl</artifactId>
<version>1.9.2</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
@@ -88,6 +88,14 @@
<artifactId>commons-io</artifactId>
<version>2.4</version>
</dependency>
<dependency>
<groupId>org.infinispan</groupId>
<artifactId>infinispan-core</artifactId>
</dependency>
<dependency>
<groupId>org.infinispan</groupId>
<artifactId>infinispan-commons</artifactId>
</dependency>
<dependency>
<groupId>org.infinispan</groupId>
<artifactId>infinispan-cachestore-leveldb</artifactId>

This file was deleted.

Oops, something went wrong.
@@ -17,17 +17,17 @@

import com.google.common.collect.ImmutableSet;
import org.apache.commons.io.IOUtils;
import org.fcrepo.kernel.utils.ContentDigest;
import org.fcrepo.kernel.impl.utils.FixityInputStream;
import org.fcrepo.kernel.utils.FixityResult;
import org.fcrepo.kernel.impl.utils.FixityResultImpl;
import org.fcrepo.kernel.impl.utils.infinispan.StoreChunkInputStream;
import org.fcrepo.kernel.impl.utils.infinispan.CacheLoaderChunkInputStream;
import org.fcrepo.kernel.utils.ContentDigest;
import org.fcrepo.kernel.utils.FixityResult;
import org.infinispan.Cache;
import org.infinispan.CacheImpl;
import org.infinispan.distexec.DistributedCallable;
import org.infinispan.loaders.CacheLoaderManager;
import org.infinispan.loaders.CacheStore;
import org.infinispan.loaders.decorators.ChainingCacheStore;

import org.infinispan.persistence.manager.PersistenceManager;
import org.infinispan.persistence.spi.CacheLoader;

import java.io.InputStream;
import java.io.Serializable;
@@ -46,28 +46,34 @@
public class DistributedFixityCheck implements DistributedCallable<String, byte[], Collection<FixityResult>>,
Serializable {
private final String dataKey;
private final int chunkSize;
private final long length;
private Cache<String, byte[]> cache;

/**
*
* @param dataKey
*/
public DistributedFixityCheck(final String dataKey) {
public DistributedFixityCheck(final String dataKey, final int chunkSize, final long length) {
this.dataKey = dataKey;
this.chunkSize = chunkSize;
this.length = length;
}

@Override
public Collection<FixityResult> call() throws Exception {
final ImmutableSet.Builder<FixityResult> fixityResults = new ImmutableSet.Builder<>();

for (final CacheStore store : stores()) {
for (final CacheLoader store : stores()) {

final String digest = ContentDigest.getAlgorithm(new URI("urn:sha1"));

FixityInputStream fixityInputStream;
final InputStream cacheLoaderChunkInputStream = new StoreChunkInputStream(store, dataKey);
final InputStream cacheLoaderChunkInputStream = new CacheLoaderChunkInputStream(
store, dataKey, chunkSize, length);

final FixityInputStream fixityInputStream = new FixityInputStream(
cacheLoaderChunkInputStream, MessageDigest.getInstance(digest));

fixityInputStream = new FixityInputStream(cacheLoaderChunkInputStream, MessageDigest.getInstance(digest));
IOUtils.copy(fixityInputStream, NULL_OUTPUT_STREAM);

final URI calculatedChecksum = ContentDigest.asURI(digest, fixityInputStream.getMessageDigest().digest());
@@ -79,7 +85,7 @@ public DistributedFixityCheck(final String dataKey) {
return fixityResults.build();
}

private String getExternalIdentifier(final CacheStore store) {
private String getExternalIdentifier(final CacheLoader store) {
final String address;

if (cache.getCacheManager().getAddress() != null) {
@@ -96,13 +102,8 @@ public void setEnvironment(final Cache<String, byte[]> cache, final Set<String>
this.cache = cache;
}

private Set<CacheStore> stores() {
final CacheLoaderManager cacheLoaderManager
= ((CacheImpl) cache).getComponentRegistry().getLocalComponent(CacheLoaderManager.class);
if (cacheLoaderManager.getCacheLoader() instanceof ChainingCacheStore) {
return ((ChainingCacheStore)cacheLoaderManager.getCacheLoader()).getStores().keySet();
} else {
return ImmutableSet.of(cacheLoaderManager.getCacheStore());
}
private Set<CacheLoader> stores() {
return ((CacheImpl)cache).getComponentRegistry().getLocalComponent(PersistenceManager.class)
.getStores(CacheLoader.class);
}
}
@@ -28,7 +28,9 @@
import org.fcrepo.kernel.utils.FixityResult;
import org.infinispan.distexec.DistributedExecutorService;
import org.modeshape.jcr.value.BinaryKey;
import org.modeshape.jcr.value.binary.infinispan.ChunkBinaryMetadata;
import org.modeshape.jcr.value.binary.infinispan.InfinispanBinaryStore;
import org.modeshape.jcr.value.binary.infinispan.InfinispanUtils;
import org.slf4j.Logger;

import javax.jcr.Property;
@@ -65,9 +67,12 @@ public InfinispanCacheStoreEntry(final InfinispanBinaryStore store, final Proper
final ImmutableSet.Builder<FixityResult> fixityResults = new ImmutableSet.Builder<>();

if (store().hasBinary(key)) {
final String dataKey = dataKeyFor(key);
final String dataKey = InfinispanUtils.dataKeyFrom((InfinispanBinaryStore)store(), key);
final ChunkBinaryMetadata metadata = InfinispanUtils.getMetadata((InfinispanBinaryStore)store(), key);

final DistributedFixityCheck task = new DistributedFixityCheck(dataKey, metadata.getChunkSize(),
metadata.getLength());

final DistributedFixityCheck task = new DistributedFixityCheck(dataKey);
final List<Future<Collection<FixityResult>>> futures
= clusterExecutor().submitEverywhere(task, dataKey + "-0");

Oops, something went wrong.

0 comments on commit 9915c50

Please sign in to comment.
You can’t perform that action at this time.