Skip to content

Commit

Permalink
reformatting to remove tab characters
Browse files Browse the repository at this point in the history
  • Loading branch information
@cbeer
cbeer committed Jun 24, 2013
1 parent 0f06b15 commit 9beffae
Show file tree
Hide file tree
Showing 13 changed files with 232 additions and 235 deletions.
191 changes: 95 additions & 96 deletions fcrepo-kernel/src/main/java/org/fcrepo/services/LowLevelStorageService.java
View file
Expand Up @@ -85,36 +85,36 @@ public <T> Collection<T> transformLowLevelCacheEntries(final Property jcrBinaryP
final Function <LowLevelCacheEntry, T>
transform)
throws RepositoryException {

return transformLowLevelCacheEntries(getBinaryKey.apply(jcrBinaryProperty), transform);
}

public <T> Collection<T> transformLowLevelCacheEntries(final BinaryKey key,
final Function <LowLevelCacheEntry, T>
transform)
throws RepositoryException {

final BinaryStore store = getBinaryStore.apply(repo);

if (store == null) {
return ImmutableSet.of();
return ImmutableSet.of();
}
if (store instanceof CompositeBinaryStore) {
return
transformLowLevelCacheEntries((CompositeBinaryStore) store,
transformLowLevelCacheEntries((CompositeBinaryStore) store,
key, transform);

} else if (store instanceof InfinispanBinaryStore) {
try {
return getClusterResults(
(InfinispanBinaryStore) store, key, transform);
} catch (InterruptedException e) {
logger.error(e.getMessage(), e);
return ImmutableSet.of();
} catch (ExecutionException e) {
logger.error(e.getMessage(), e);
return ImmutableSet.of();
}
return getClusterResults(
(InfinispanBinaryStore) store, key, transform);
} catch (InterruptedException e) {
logger.error(e.getMessage(), e);
return ImmutableSet.of();
} catch (ExecutionException e) {
logger.error(e.getMessage(), e);
return ImmutableSet.of();
}
} else {
final ImmutableSet.Builder<T> blobs = builder();
blobs.add(transform.apply(new LocalBinaryStoreEntry(store, key)));
Expand All @@ -128,7 +128,7 @@ public <T> Collection<T> transformLowLevelCacheEntries(final BinaryKey key,
* @return a set of transformed objects
*/
protected <T> Set<T> transformLowLevelCacheEntries(
final CompositeBinaryStore compositeStore,
final CompositeBinaryStore compositeStore,
final BinaryKey key,
final Function <LowLevelCacheEntry, T> transform) {

Expand All @@ -145,9 +145,9 @@ protected <T> Set<T> transformLowLevelCacheEntries(
if (bs.hasBinary(key)) {

final Function <LowLevelCacheEntry, T> decorator =
new ExternalIdDecorator<>(entry.getKey(), transform);
new ExternalIdDecorator<>(entry.getKey(), transform);
final Set<T> transformeds =
transformLowLevelCacheEntries(bs, key, decorator);
transformLowLevelCacheEntries(bs, key, decorator);
results.addAll(transformeds);

}
Expand All @@ -163,21 +163,21 @@ protected <T> Set<T> transformLowLevelCacheEntries(
* @return a set of transformed objects
*/
protected <T> Set<T> transformLowLevelCacheEntries(
final BinaryStore store,
final BinaryStore store,
final BinaryKey key,
final Function <LowLevelCacheEntry, T> transform) {

if (store == null) {
return ImmutableSet.of();
return ImmutableSet.of();
}
if (store instanceof CompositeBinaryStore) {
return
transformLowLevelCacheEntries((CompositeBinaryStore) store,
transformLowLevelCacheEntries((CompositeBinaryStore) store,
key, transform);

} else if (store instanceof InfinispanBinaryStore) {
try {
return getClusterResults(
return getClusterResults(
(InfinispanBinaryStore) store, key, transform);
} catch (InterruptedException e) {
logger.error(e.getMessage(), e);
Expand Down Expand Up @@ -234,7 +234,7 @@ public Set<LowLevelCacheEntry> getLowLevelCacheEntries(final BinaryKey key) {
final BinaryStore store = getBinaryStore.apply(repo);

if (store == null) {
return ImmutableSet.of();
return ImmutableSet.of();
}

return getLowLevelCacheEntriesFromStore(store, key);
Expand Down Expand Up @@ -298,40 +298,40 @@ protected Set<LowLevelCacheEntry> getLowLevelCacheEntriesFromStore(final Infinis
"InfinispanBinaryStore {}", key, ispnStore);

try {
return getClusterResults(ispnStore, key, new Echo());
} catch (InterruptedException e) {
logger.error(e.getMessage(), e);
return ImmutableSet.of();
} catch (ExecutionException e) {
logger.error(e.getMessage(), e);
return ImmutableSet.of();
}
return getClusterResults(ispnStore, key, new Echo());
} catch (InterruptedException e) {
logger.error(e.getMessage(), e);
return ImmutableSet.of();
} catch (ExecutionException e) {
logger.error(e.getMessage(), e);
return ImmutableSet.of();
}
}

public <T> Set<T> getClusterResults(InfinispanBinaryStore cacheStore,
BinaryKey key, Function<LowLevelCacheEntry, T> transform)
throws InterruptedException, ExecutionException {
DistributedExecutorService exec =
getClusterExecutor(cacheStore);
@SuppressWarnings( {"synthetic-access", "unchecked", "rawtypes"} )
BinaryKey key, Function<LowLevelCacheEntry, T> transform)
throws InterruptedException, ExecutionException {
DistributedExecutorService exec =
getClusterExecutor(cacheStore);
@SuppressWarnings( {"synthetic-access", "unchecked", "rawtypes"} )
List<Future<Collection<T>>> futures = exec.submitEverywhere(
new CacheLocalTransform(key, new Unroll<T>(transform)));
Set<T> results = new HashSet<T>(futures.size());
while(futures.size() > 0) {
Iterator<Future<Collection<T>>> futureIter =
futures.iterator();
while(futureIter.hasNext()) {
Future<Collection<T>> future = futureIter.next();
try {
Collection<T> result = future.get(100, TimeUnit.MILLISECONDS);
futureIter.remove();
results.addAll(result);
} catch (TimeoutException e) {
// we're just going to ignore this and try again!
}
}
}
return results;
new CacheLocalTransform(key, new Unroll<T>(transform)));
Set<T> results = new HashSet<T>(futures.size());
while(futures.size() > 0) {
Iterator<Future<Collection<T>>> futureIter =
futures.iterator();
while(futureIter.hasNext()) {
Future<Collection<T>> future = futureIter.next();
try {
Collection<T> result = future.get(100, TimeUnit.MILLISECONDS);
futureIter.remove();
results.addAll(result);
} catch (TimeoutException e) {
// we're just going to ignore this and try again!
}
}
}
return results;
}

/**
Expand All @@ -358,59 +358,58 @@ public void setGetBinaryKey(final GetBinaryKey getBinaryKey) {
static class ExternalIdDecorator<T>
implements Function<LowLevelCacheEntry, T>, Serializable {

/**
* Must be serializable
*/
private static final long serialVersionUID = 7375231595038804409L;
final String externalId;
final Function<LowLevelCacheEntry, T> transform;
ExternalIdDecorator(String externalId, Function<LowLevelCacheEntry, T> transform) {
this.externalId = externalId;
this.transform = transform;
}
@Override
public T apply(LowLevelCacheEntry input) {
input.setExternalId(externalId);
return transform.apply(input);
}
/**
* Must be serializable
*/
private static final long serialVersionUID = 7375231595038804409L;
final String externalId;
final Function<LowLevelCacheEntry, T> transform;
ExternalIdDecorator(String externalId, Function<LowLevelCacheEntry, T> transform) {
this.externalId = externalId;
this.transform = transform;
}

@Override
public T apply(LowLevelCacheEntry input) {
input.setExternalId(externalId);
return transform.apply(input);
}

}

static class Echo implements Function<LowLevelCacheEntry, LowLevelCacheEntry>, Serializable {

private static final long serialVersionUID = -1L;
@Override
public LowLevelCacheEntry apply(LowLevelCacheEntry input) {
return input;
}
private static final long serialVersionUID = -1L;

@Override
public LowLevelCacheEntry apply(LowLevelCacheEntry input) {
return input;
}

}

static class Unroll<T> implements Function<LowLevelCacheEntry, Collection<T>>, Serializable {

private final Function<LowLevelCacheEntry, T> transform;

Unroll(Function<LowLevelCacheEntry, T> transform) {
this.transform = transform;
}

private static final long serialVersionUID = -1L;

@Override
public Collection<T> apply(LowLevelCacheEntry input) {
final ImmutableSet.Builder<T> entries = builder();
if (input instanceof ChainingCacheStoreEntry) {
entries.addAll(
transform(((ChainingCacheStoreEntry)input).chainedEntries(), transform)
);
} else {
entries.add(transform.apply(input)).build();
}
return entries.build();
}

private final Function<LowLevelCacheEntry, T> transform;

Unroll(Function<LowLevelCacheEntry, T> transform) {
this.transform = transform;
}

private static final long serialVersionUID = -1L;

@Override
public Collection<T> apply(LowLevelCacheEntry input) {
final ImmutableSet.Builder<T> entries = builder();
if (input instanceof ChainingCacheStoreEntry) {
entries.addAll(
transform(((ChainingCacheStoreEntry)input).chainedEntries(), transform)
);
} else {
entries.add(transform.apply(input)).build();
}
return entries.build();
}
}

}
2 changes: 1 addition & 1 deletion fcrepo-kernel/src/main/java/org/fcrepo/services/ServiceHelpers.java
View file
Expand Up @@ -119,7 +119,7 @@ public static Long getContentSize(final Node ds)
public static DistributedExecutorService getClusterExecutor(InfinispanBinaryStore cacheStore) {
// Watch out! This is trying to pluck out the blob cache store. This works as long as
// modeshape continues to be ordered..
return new DefaultExecutorService(cacheStore.getCaches().get(1));
return new DefaultExecutorService(cacheStore.getCaches().get(1));
}

/**
Expand Down
66 changes: 33 additions & 33 deletions fcrepo-kernel/src/main/java/org/fcrepo/services/functions/CacheLocalTransform.java
View file
Expand Up @@ -19,38 +19,38 @@
public class CacheLocalTransform<K, V, T>
implements DistributedCallable<K, V, T>, Serializable {

/**
* Because this class will be communicated between cache nodes,
* it must be serializable
*/
private static final long serialVersionUID = -7014104738830230123L;

private static GetCacheStore TRANSFORM = new GetCacheStore();

private BinaryKey key;
private Function<LowLevelCacheEntry, T> entryTransform;
private CacheStore store;
private String cacheName = "";

public CacheLocalTransform(final BinaryKey key,
final Function<LowLevelCacheEntry, T> entryTransform) {
this.key = key;
this.entryTransform = entryTransform;
}

@Override
public T call() throws Exception {
LowLevelCacheEntry entry =
(store instanceof ChainingCacheStore) ?
new ChainingCacheStoreEntry((ChainingCacheStore)store, cacheName, key) :
new CacheStoreEntry(store, cacheName, key);
return this.entryTransform.apply(entry);
}

@Override
public void setEnvironment(Cache<K, V> cache, Set<K> keys) {
this.store = TRANSFORM.apply(cache);
/**
* Because this class will be communicated between cache nodes,
* it must be serializable
*/
private static final long serialVersionUID = -7014104738830230123L;

private static GetCacheStore TRANSFORM = new GetCacheStore();

private BinaryKey key;
private Function<LowLevelCacheEntry, T> entryTransform;
private CacheStore store;
private String cacheName = "";

public CacheLocalTransform(final BinaryKey key,
final Function<LowLevelCacheEntry, T> entryTransform) {

this.key = key;
this.entryTransform = entryTransform;
}

@Override
public T call() throws Exception {
LowLevelCacheEntry entry =
(store instanceof ChainingCacheStore) ?
new ChainingCacheStoreEntry((ChainingCacheStore)store, cacheName, key) :
new CacheStoreEntry(store, cacheName, key);
return this.entryTransform.apply(entry);
}

@Override
public void setEnvironment(Cache<K, V> cache, Set<K> keys) {
this.store = TRANSFORM.apply(cache);
this.cacheName = ((CacheImpl<K, V>) cache).getName();
}
}
}

0 comments on commit 9beffae

Please sign in to comment.