You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by sn...@apache.org on 2014/03/25 16:21:04 UTC
[05/43] git commit: Refactor to use the observables. Evaluating
hazelcast
Refactor to use the observables. Evaluating hazelcast
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/da4630f5
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/da4630f5
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/da4630f5
Branch: refs/heads/two-dot-o
Commit: da4630f543fab7eb73fdbfab9535c82c04568179
Parents: 4012eb7
Author: Todd Nine <tn...@apigee.com>
Authored: Mon Mar 3 11:16:30 2014 -0700
Committer: Todd Nine <tn...@apigee.com>
Committed: Mon Mar 3 11:16:30 2014 -0700
----------------------------------------------------------------------
.../graph/consistency/AsyncProcessor.java | 8 +-
.../graph/consistency/AsyncProcessorImpl.java | 64 ++++----
.../graph/consistency/AsynchronousEvent.java | 18 ---
.../consistency/AsynchronousEventListener.java | 22 ---
.../graph/consistency/AsynchronousMessage.java | 18 +++
.../graph/consistency/ErrorListener.java | 2 +-
.../graph/consistency/LocalTimeoutQueue.java | 21 +--
.../graph/consistency/MessageListener.java | 22 +++
.../consistency/SimpleAsynchronousEvent.java | 30 ----
.../consistency/SimpleAsynchronousMessage.java | 30 ++++
.../graph/consistency/TimeoutQueue.java | 8 +-
.../persistence/graph/impl/EdgeManagerImpl.java | 153 +++++++++----------
.../graph/consistency/AsyncProcessorTest.java | 56 ++++---
.../consistency/LocalTimeoutQueueTest.java | 28 ++--
14 files changed, 245 insertions(+), 235 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/da4630f5/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/AsyncProcessor.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/AsyncProcessor.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/AsyncProcessor.java
index 3b3e0b9..f6013d9 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/AsyncProcessor.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/AsyncProcessor.java
@@ -19,16 +19,16 @@ public interface AsyncProcessor<T> {
* @param event The event to be scheduled for verification
* @param timeout The time in milliseconds we should wait before the event should fire
*/
- public AsynchronousEvent<T> setVerification( T event, long timeout );
+ public AsynchronousMessage<T> setVerification( T event, long timeout );
/**
- * Start processing the event immediately asynchronously. In the event an exception is thrown, the AsynchronousEvent should be re-tried.
+ * Start processing the event immediately asynchronously. In the event an exception is thrown, the AsynchronousMessage should be re-tried.
* It is up to the implementer to commit the event so that it does not fire again. This should never throw exceptions.
*
* @param event The event to start
*/
- public void start(AsynchronousEvent<T> event);
+ public void start(AsynchronousMessage<T> event);
/**
* Add the error listener to the list of listeners
@@ -40,7 +40,7 @@ public interface AsyncProcessor<T> {
* Add the listener to this instance
* @param listener
*/
- public void addListener(AsynchronousEventListener<T> listener);
+ public void addListener(MessageListener<T, T> listener);
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/da4630f5/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/AsyncProcessorImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/AsyncProcessorImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/AsyncProcessorImpl.java
index 5acf728..fbac644 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/AsyncProcessorImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/AsyncProcessorImpl.java
@@ -9,11 +9,13 @@ import org.slf4j.LoggerFactory;
import com.google.inject.Inject;
import com.google.inject.Singleton;
-import com.netflix.hystrix.HystrixCommand;
import com.netflix.hystrix.HystrixCommandGroupKey;
-import rx.Observer;
+import rx.Observable;
import rx.Scheduler;
+import rx.util.functions.Action0;
+import rx.util.functions.Action1;
+import rx.util.functions.FuncN;
/**
@@ -26,7 +28,7 @@ public class AsyncProcessorImpl<T> implements AsyncProcessor<T> {
private final TimeoutQueue<T> queue;
private final Scheduler scheduler;
- private final List<AsynchronousEventListener<T>> listeners = new ArrayList<AsynchronousEventListener<T>>( );
+ private final List<MessageListener<T, T>> listeners = new ArrayList<MessageListener<T, T>>();
private static final Logger LOG = LoggerFactory.getLogger( AsyncProcessor.class );
@@ -40,60 +42,57 @@ public class AsyncProcessorImpl<T> implements AsyncProcessor<T> {
}
-
@Override
- public AsynchronousEvent<T> setVerification( final T event, final long timeout ) {
+ public AsynchronousMessage<T> setVerification( final T event, final long timeout ) {
return queue.queue( event, timeout );
}
-
-
@Override
- public void start( final AsynchronousEvent<T> event ) {
+ public void start( final AsynchronousMessage<T> event ) {
- //run this in a timeout command so it doesn't run forever. If it times out, it will simply resume later
- new HystrixCommand<Void>( GRAPH_REPAIR ) {
-
- @Override
- protected Void run() throws Exception {
- final T rootEvent = event.getEvent();
+ final T data = event.getEvent();
+ /**
+ * Execute all listeners in parallel
+ */
+ List<Observable<?>> observables = new ArrayList<Observable<?>>( listeners.size() );
- for(AsynchronousEventListener<T> listener: listeners){
- listener.receive( rootEvent );
- }
+ for ( MessageListener<T, T> listener : listeners ) {
+ observables.add( listener.receive( data ).subscribeOn( scheduler ) );
+ }
- return null;
- }
- }.toObservable( scheduler ).subscribe( new Observer<Void>() {
+ //run everything in parallel and zip it up
+ Observable.zip( observables, new FuncN<AsynchronousMessage<T>>() {
@Override
- public void onCompleted() {
- queue.remove( event );
+ public AsynchronousMessage<T> call( final Object... args ) {
+ return event;
}
-
-
+ } ).doOnError( new Action1<Throwable>() {
@Override
- public void onError( final Throwable throwable ) {
+ public void call( final Throwable throwable ) {
LOG.error( "Unable to process async event", throwable );
for ( ErrorListener listener : errorListeners ) {
listener.onError( event, throwable );
}
}
-
-
+ } ).doOnCompleted( new Action0() {
+ @Override
+ public void call() {
+ queue.remove( event );
+ }
+ } ).subscribe( new Action1<AsynchronousMessage<T>>() {
@Override
- public void onNext( final Void args ) {
- //nothing to do here
+ public void call( final AsynchronousMessage<T> tAsynchronousMessage ) {
+ //To change body of implemented methods use File | Settings | File Templates.
}
} );
-
}
@Override
- public void addListener( final AsynchronousEventListener<T> listener ) {
+ public void addListener( final MessageListener<T, T> listener ) {
this.listeners.add( listener );
}
@@ -104,7 +103,4 @@ public class AsyncProcessorImpl<T> implements AsyncProcessor<T> {
public void addErrorListener( ErrorListener<T> listener ) {
this.errorListeners.add( listener );
}
-
-
-
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/da4630f5/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/AsynchronousEvent.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/AsynchronousEvent.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/AsynchronousEvent.java
deleted file mode 100644
index fe38a93..0000000
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/AsynchronousEvent.java
+++ /dev/null
@@ -1,18 +0,0 @@
-package org.apache.usergrid.persistence.graph.consistency;
-
-
-/**
- * An interface for a timeout event
- */
-public interface AsynchronousEvent<T> {
-
- /**
- * @return The event to fire when our timeout is reached
- */
- T getEvent();
-
- /**
- * @return The time in epoch millis the event will time out
- */
- long getTimeout();
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/da4630f5/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/AsynchronousEventListener.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/AsynchronousEventListener.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/AsynchronousEventListener.java
deleted file mode 100644
index 6cb93cb..0000000
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/AsynchronousEventListener.java
+++ /dev/null
@@ -1,22 +0,0 @@
-package org.apache.usergrid.persistence.graph.consistency;
-
-
-import rx.Observable;
-
-
-/**
- *
- *
- */
-public interface AsynchronousEventListener<T, R> {
-
-
- /**
- * The handler to receive the event. Any exception that is thrown is considered
- * a failure, and the event will be re-fired.
- * @param event The input event
- * @return The observable that performs the operations
- */
- Observable<R> receive(T event);
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/da4630f5/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/AsynchronousMessage.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/AsynchronousMessage.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/AsynchronousMessage.java
new file mode 100644
index 0000000..c9bd00a
--- /dev/null
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/AsynchronousMessage.java
@@ -0,0 +1,18 @@
+package org.apache.usergrid.persistence.graph.consistency;
+
+
+/**
+ * An interface for a timeout event
+ */
+public interface AsynchronousMessage<T> {
+
+ /**
+ * @return The event to fire when our timeout is reached
+ */
+ T getEvent();
+
+ /**
+ * @return The time in epoch millis the event will time out
+ */
+ long getTimeout();
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/da4630f5/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/ErrorListener.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/ErrorListener.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/ErrorListener.java
index f6cef41..f736cdb 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/ErrorListener.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/ErrorListener.java
@@ -6,5 +6,5 @@ package org.apache.usergrid.persistence.graph.consistency;
*/
public interface ErrorListener <T> {
- void onError( AsynchronousEvent<T> event, Throwable t );
+ void onError( AsynchronousMessage<T> event, Throwable t );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/da4630f5/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/LocalTimeoutQueue.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/LocalTimeoutQueue.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/LocalTimeoutQueue.java
index 3a3b2ce..d2cc17f 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/LocalTimeoutQueue.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/LocalTimeoutQueue.java
@@ -23,7 +23,7 @@ public class LocalTimeoutQueue<T> implements TimeoutQueue<T> {
/**
* For in memory queueing
*/
- private final PriorityBlockingQueue<AsynchronousEvent<T>> queue = new PriorityBlockingQueue<AsynchronousEvent<T>>( 1000, new TimeoutEventCompatator<T>() );
+ private final PriorityBlockingQueue<AsynchronousMessage<T>> queue = new PriorityBlockingQueue<AsynchronousMessage<T>>( 1000, new TimeoutEventCompatator<T>() );
private final TimeService timeService;
@@ -35,9 +35,9 @@ public class LocalTimeoutQueue<T> implements TimeoutQueue<T> {
@Override
- public AsynchronousEvent<T> queue( final T event, final long timeout ) {
+ public AsynchronousMessage<T> queue( final T event, final long timeout ) {
final long scheduledTimeout = timeService.getCurrentTime() + timeout;
- final AsynchronousEvent<T> queuedEvent = new SimpleAsynchronousEvent<T>( event, scheduledTimeout );
+ final AsynchronousMessage<T> queuedEvent = new SimpleAsynchronousMessage<T>( event, scheduledTimeout );
queue.add( queuedEvent );
@@ -46,16 +46,16 @@ public class LocalTimeoutQueue<T> implements TimeoutQueue<T> {
@Override
- public Collection<AsynchronousEvent<T>> take( final int maxSize, final long timeout ) {
+ public Collection<AsynchronousMessage<T>> take( final int maxSize, final long timeout ) {
final long now = timeService.getCurrentTime();
final long newTimeout = now+timeout;
- List<AsynchronousEvent<T>> results = new ArrayList<AsynchronousEvent<T>>(maxSize);
+ List<AsynchronousMessage<T>> results = new ArrayList<AsynchronousMessage<T>>(maxSize);
for(int i = 0; i < maxSize; i ++){
- AsynchronousEvent<T> queuedEvent = queue.peek();
+ AsynchronousMessage<T> queuedEvent = queue.peek();
//nothing to do
if(queuedEvent == null){
@@ -68,7 +68,8 @@ public class LocalTimeoutQueue<T> implements TimeoutQueue<T> {
break;
}
- final AsynchronousEvent<T> newEvent = new SimpleAsynchronousEvent<T>( queuedEvent.getEvent(), newTimeout );
+ final AsynchronousMessage<T>
+ newEvent = new SimpleAsynchronousMessage<T>( queuedEvent.getEvent(), newTimeout );
//re schedule a new event to replace this one
queue.add(newEvent);
@@ -85,16 +86,16 @@ public class LocalTimeoutQueue<T> implements TimeoutQueue<T> {
@Override
- public boolean remove( final AsynchronousEvent<T> event ) {
+ public boolean remove( final AsynchronousMessage<T> event ) {
return queue.remove( event );
}
- private static class TimeoutEventCompatator<T> implements Comparator<AsynchronousEvent<T>> {
+ private static class TimeoutEventCompatator<T> implements Comparator<AsynchronousMessage<T>> {
@Override
- public int compare( final AsynchronousEvent<T> o1, final AsynchronousEvent<T> o2 ) {
+ public int compare( final AsynchronousMessage<T> o1, final AsynchronousMessage<T> o2 ) {
return Long.compare( o1.getTimeout(), o2.getTimeout() );
}
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/da4630f5/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/MessageListener.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/MessageListener.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/MessageListener.java
new file mode 100644
index 0000000..8a21dbb
--- /dev/null
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/MessageListener.java
@@ -0,0 +1,22 @@
+package org.apache.usergrid.persistence.graph.consistency;
+
+
+import rx.Observable;
+
+
+/**
+ *
+ *
+ */
+public interface MessageListener<T, R> {
+
+
+ /**
+ * The handler to receive the event. Any exception that is thrown is considered
+ * a failure, and the event will be re-fired.
+ * @param event The input event
+ * @return The observable that performs the operations
+ */
+ Observable<T> receive(T event);
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/da4630f5/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/SimpleAsynchronousEvent.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/SimpleAsynchronousEvent.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/SimpleAsynchronousEvent.java
deleted file mode 100644
index fdfcd60..0000000
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/SimpleAsynchronousEvent.java
+++ /dev/null
@@ -1,30 +0,0 @@
-package org.apache.usergrid.persistence.graph.consistency;
-
-
-/**
- *
- *
- */
-public class SimpleAsynchronousEvent<T> implements AsynchronousEvent<T> {
-
- private final T event;
- private final long timeout;
-
-
- public SimpleAsynchronousEvent( final T event, final long timeout ) {
- this.event = event;
- this.timeout = timeout;
- }
-
-
- @Override
- public T getEvent() {
- return event;
- }
-
-
- @Override
- public long getTimeout() {
- return timeout;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/da4630f5/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/SimpleAsynchronousMessage.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/SimpleAsynchronousMessage.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/SimpleAsynchronousMessage.java
new file mode 100644
index 0000000..1e7a04b
--- /dev/null
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/SimpleAsynchronousMessage.java
@@ -0,0 +1,30 @@
+package org.apache.usergrid.persistence.graph.consistency;
+
+
+/**
+ *
+ *
+ */
+public class SimpleAsynchronousMessage<T> implements AsynchronousMessage<T> {
+
+ private final T event;
+ private final long timeout;
+
+
+ public SimpleAsynchronousMessage( final T event, final long timeout ) {
+ this.event = event;
+ this.timeout = timeout;
+ }
+
+
+ @Override
+ public T getEvent() {
+ return event;
+ }
+
+
+ @Override
+ public long getTimeout() {
+ return timeout;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/da4630f5/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/TimeoutQueue.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/TimeoutQueue.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/TimeoutQueue.java
index 556aa88..0caed9c 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/TimeoutQueue.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/consistency/TimeoutQueue.java
@@ -14,9 +14,9 @@ public interface TimeoutQueue<T> {
*
* @param event The event to queue
* @param timeout The timeout to set on the queue element
- * @return The AsynchronousEvent that has been queued
+ * @return The AsynchronousMessage that has been queued
*/
- public AsynchronousEvent<T> queue( T event, long timeout );
+ public AsynchronousMessage<T> queue( T event, long timeout );
/**
@@ -29,7 +29,7 @@ public interface TimeoutQueue<T> {
*
* @return A collection of events.
*/
- public Collection<AsynchronousEvent<T>> take( int maxSize, long timeout );
+ public Collection<AsynchronousMessage<T>> take( int maxSize, long timeout );
/**
@@ -39,5 +39,5 @@ public interface TimeoutQueue<T> {
*
* @return True if the element was removed. False otherwise
*/
- public boolean remove( AsynchronousEvent<T> event );
+ public boolean remove( AsynchronousMessage<T> event );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/da4630f5/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/EdgeManagerImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/EdgeManagerImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/EdgeManagerImpl.java
index 7fd9246..1cc2fce 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/EdgeManagerImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/EdgeManagerImpl.java
@@ -36,8 +36,8 @@ import org.apache.usergrid.persistence.graph.SearchByIdType;
import org.apache.usergrid.persistence.graph.SearchEdgeType;
import org.apache.usergrid.persistence.graph.SearchIdType;
import org.apache.usergrid.persistence.graph.consistency.AsyncProcessor;
-import org.apache.usergrid.persistence.graph.consistency.AsynchronousEvent;
-import org.apache.usergrid.persistence.graph.consistency.AsynchronousEventListener;
+import org.apache.usergrid.persistence.graph.consistency.AsynchronousMessage;
+import org.apache.usergrid.persistence.graph.consistency.MessageListener;
import org.apache.usergrid.persistence.graph.guice.EdgeDelete;
import org.apache.usergrid.persistence.graph.guice.EdgeWrite;
import org.apache.usergrid.persistence.graph.guice.NodeDelete;
@@ -135,7 +135,7 @@ public class EdgeManagerImpl implements EdgeManager {
mutation.mergeShallow( edgeMutation );
- final AsynchronousEvent<Edge> event = edgeWriteAsyncProcessor.setVerification( edge, getTimeout() );
+ final AsynchronousMessage<Edge> event = edgeWriteAsyncProcessor.setVerification( edge, getTimeout() );
try {
mutation.execute();
@@ -161,7 +161,7 @@ public class EdgeManagerImpl implements EdgeManager {
public Edge call( final Edge edge ) {
final MutationBatch edgeMutation = edgeSerialization.markEdge( scope, edge );
- final AsynchronousEvent<Edge> event = edgeDeleteAsyncProcessor.setVerification( edge, getTimeout() );
+ final AsynchronousMessage<Edge> event = edgeDeleteAsyncProcessor.setVerification( edge, getTimeout() );
try {
@@ -191,7 +191,7 @@ public class EdgeManagerImpl implements EdgeManager {
final MutationBatch nodeMutation = nodeSerialization.mark( scope, id, deleteTime );
- final AsynchronousEvent<Id> event = nodeDeleteAsyncProcessor.setVerification( node, getTimeout() );
+ final AsynchronousMessage<Id> event = nodeDeleteAsyncProcessor.setVerification( node, getTimeout() );
try {
@@ -420,10 +420,10 @@ public class EdgeManagerImpl implements EdgeManager {
/**
* Construct the asynchronous edge lister for the repair operation.
*/
- public class EdgeWriteListener implements AsynchronousEventListener<Edge, Integer> {
+ public class EdgeWriteListener implements MessageListener<Edge, Edge> {
@Override
- public Observable<Integer> receive( final Edge write ) {
+ public Observable<Edge> receive( final Edge write ) {
final UUID maxVersion = write.getVersion();
@@ -452,9 +452,9 @@ public class EdgeManagerImpl implements EdgeManager {
return UUIDComparator.staticCompare( markedEdge.getVersion(), maxVersion ) < 0;
}
//buffer the deletes and issue them in a single mutation
- } ).buffer( graphFig.getScanPageSize() ).map( new Func1<List<MarkedEdge>, Integer>() {
+ } ).buffer( graphFig.getScanPageSize() ).map( new Func1<List<MarkedEdge>, Edge>() {
@Override
- public Integer call( final List<MarkedEdge> markedEdges ) {
+ public Edge call( final List<MarkedEdge> markedEdges ) {
final int size = markedEdges.size();
@@ -473,7 +473,7 @@ public class EdgeManagerImpl implements EdgeManager {
throw new RuntimeException( "Unable to issue write to cassandra", e );
}
- return size;
+ return write;
}
} );
}
@@ -483,10 +483,10 @@ public class EdgeManagerImpl implements EdgeManager {
/**
* Construct the asynchronous delete operation from the listener
*/
- public class EdgeDeleteListener implements AsynchronousEventListener<Edge, Integer> {
+ public class EdgeDeleteListener implements MessageListener<Edge, Edge> {
@Override
- public Observable<Integer> receive( final Edge delete ) {
+ public Observable<Edge> receive( final Edge delete ) {
final UUID maxVersion = delete.getVersion();
@@ -496,76 +496,57 @@ public class EdgeManagerImpl implements EdgeManager {
//search by edge type and target type. If any other edges with this target type exist,
// we can't delete it
- Observable<Integer> sourceIdType= loadEdgesFromSourceByType(
+ Observable<MutationBatch> sourceIdType = loadEdgesFromSourceByType(
new SimpleSearchByIdType( edge.getSourceNode(), edge.getType(), maxVersion,
- edge.getTargetNode().getType(), null ) ).take( 2 ).count().doOnEach(
- new Action1<Integer>() {
+ edge.getTargetNode().getType(), null ) ).take( 2 ).count()
+ .map( new Func1<Integer, MutationBatch>() {
@Override
- public void call( final Integer count ) {
+ public MutationBatch call( final Integer count ) {
//There's nothing to do, we have 2 different edges with the same edge type and
// target type. Don't delete meta data
if ( count == 2 ) {
- return;
+ return null;
}
- try {
- edgeMetadataSerialization.removeEdgeTypeFromSource( scope, delete ).execute();
- }
- catch ( ConnectionException e ) {
- throw new RuntimeException( "Unable to execute mutation", e );
- }
+ return edgeMetadataSerialization.removeEdgeTypeFromSource( scope, delete );
}
} );
- Observable<Integer> targetIdType = loadEdgesToTargetByType(
+ Observable<MutationBatch> targetIdType = loadEdgesToTargetByType(
new SimpleSearchByIdType( edge.getTargetNode(), edge.getType(), maxVersion,
edge.getSourceNode().getType(), null ) ).take( 2 ).count()
- .doOnEach( new Action1<Integer>() {
-
-
+ .map( new Func1<Integer, MutationBatch>() {
@Override
- public void call( final Integer count ) {
-
-
+ public MutationBatch call( final Integer count ) {
//There's nothing to do, we have 2 different edges with the same edge type and
// target type. Don't delete meta data
if ( count == 2 ) {
- return;
+ return null;
}
- try {
- edgeMetadataSerialization.removeEdgeTypeToTarget( scope, delete ).execute();
- }
- catch ( ConnectionException e ) {
- throw new RuntimeException( "Unable to execute mutation", e );
- }
+
+ return edgeMetadataSerialization.removeEdgeTypeToTarget( scope, delete );
}
} );
//search by edge type and target type. If any other edges with this target type exist,
// we can't delete it
- Observable<Integer> sourceType = loadEdgesFromSource(
+ Observable<MutationBatch> sourceType = loadEdgesFromSource(
new SimpleSearchByEdgeType( edge.getSourceNode(), edge.getType(), maxVersion, null ) )
- .take( 2 ).count().doOnEach( new Action1<Integer>() {
-
-
+ .take( 2 ).count().map( new Func1<Integer, MutationBatch>() {
@Override
- public void call( final Integer count ) {
+ public MutationBatch call( final Integer count ) {
//There's nothing to do, we have 2 different edges with the same edge type and
// target type. Don't delete meta data
if ( count == 2 ) {
- return;
+ return null;
}
- try {
- edgeMetadataSerialization.removeEdgeTypeFromSource( scope, delete ).execute();
- }
- catch ( ConnectionException e ) {
- throw new RuntimeException( "Unable to execute mutation", e );
- }
+
+ return edgeMetadataSerialization.removeEdgeTypeFromSource( scope, delete );
}
} );
@@ -625,12 +606,19 @@ public class EdgeManagerImpl implements EdgeManager {
}
} );
}
- } ).map( new Func1<MutationBatch, MutationBatch>() {
+ } ).map( new Func1<MutationBatch, Edge>() {
@Override
- public MutationBatch call( final MutationBatch mutationBatch ) {
- return mutationBatch;
+ public Edge call( final MutationBatch mutationBatch ) {
+ try {
+ mutationBatch.execute();
+ }
+ catch ( ConnectionException e ) {
+ throw new RuntimeException( "Unable to execute mutation", e );
+ }
+
+ return delete;
}
- } ).count();
+ } );
}
}
@@ -638,35 +626,44 @@ public class EdgeManagerImpl implements EdgeManager {
/**
* Construct the asynchronous node delete from the q
*/
- public class NodeDeleteListener implements AsynchronousEventListener<Id, Integer> {
+ public class NodeDeleteListener implements MessageListener<Id, Id> {
@Override
- public Observable<Integer> receive( final Id node ) {
+ public Observable<Id> receive( final Id node ) {
- final Optional<UUID> mark = nodeSerialization.getMaxVersion( scope, node );
- //Nothing to do, just exist with 0 count
- if ( !mark.isPresent() ) {
- return Observable.just( 0 );
- }
-
- final UUID maxVersion = mark.get();
-
- return getEdgeTypesToTarget( new SimpleSearchEdgeType( node, null ) )
- .flatMap( new Func1<String, Observable<Edge>>() {
- @Override
- public Observable<Edge> call( final String edgeType ) {
-
- //for each edge type, we want to search all edges < this version to the node and delete them. We might want to batch this up for efficiency
- return loadEdgesToTarget( new SimpleSearchByEdgeType( node, edgeType, maxVersion, null ) )
- .doOnEach( new Action1<Edge>() {
- @Override
- public void call( final Edge edge ) {
- deleteEdge( edge );
- }
- } );
- }
- } ).count();
+ return Observable.from( node ).map( new Func1<Id, Optional<UUID>>() {
+ @Override
+ public Optional<UUID> call( final Id id ) {
+ return nodeSerialization.getMaxVersion( scope, node );
+ }
+ } ).flatMap( new Func1<Optional<UUID>, Observable<Edge>>() {
+ @Override
+ public Observable<Edge> call( final Optional<UUID> uuidOptional ) {
+ return getEdgeTypesToTarget( new SimpleSearchEdgeType( node, null ) )
+ .flatMap( new Func1<String, Observable<Edge>>() {
+ @Override
+ public Observable<Edge> call( final String edgeType ) {
+
+ //for each edge type, we want to search all edges < this version to the node and
+ // delete them. We might want to batch this up for efficiency
+ return loadEdgesToTarget(
+ new SimpleSearchByEdgeType( node, edgeType, uuidOptional.get(), null ) )
+ .doOnEach( new Action1<Edge>() {
+ @Override
+ public void call( final Edge edge ) {
+ deleteEdge( edge );
+ }
+ } );
+ }
+ } );
+ }
+ } ).map( new Func1<Edge, Id>() {
+ @Override
+ public Id call( final Edge edge ) {
+ return node;
+ }
+ } );
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/da4630f5/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/consistency/AsyncProcessorTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/consistency/AsyncProcessorTest.java b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/consistency/AsyncProcessorTest.java
index 6224b0a..fc53705 100644
--- a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/consistency/AsyncProcessorTest.java
+++ b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/consistency/AsyncProcessorTest.java
@@ -27,7 +27,9 @@ import org.junit.Test;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
+import rx.Observable;
import rx.concurrency.Schedulers;
+import rx.util.functions.Action1;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -52,7 +54,7 @@ public class AsyncProcessorTest {
final TestEvent event = new TestEvent();
- final AsynchronousEvent<TestEvent> asynchronousEvent = new AsynchronousEvent<TestEvent>() {
+ final AsynchronousMessage<TestEvent> asynchronousMessage = new AsynchronousMessage<TestEvent>() {
@Override
public TestEvent getEvent() {
return event;
@@ -72,13 +74,13 @@ public class AsyncProcessorTest {
//mock up the queue
- when( queue.queue( event, timeout ) ).thenReturn( asynchronousEvent );
+ when( queue.queue( event, timeout ) ).thenReturn( asynchronousMessage );
- AsynchronousEvent<TestEvent> returned = asyncProcessor.setVerification( event, timeout );
+ AsynchronousMessage<TestEvent> returned = asyncProcessor.setVerification( event, timeout );
//ensure the timeouts are returned from the Queue subsystem
- assertSame( asynchronousEvent, returned );
+ assertSame( asynchronousMessage, returned );
}
@@ -90,7 +92,7 @@ public class AsyncProcessorTest {
final TestEvent event = new TestEvent();
- final AsynchronousEvent<TestEvent> asynchronousEvent = new AsynchronousEvent<TestEvent>() {
+ final AsynchronousMessage<TestEvent> asynchronousMessage = new AsynchronousMessage<TestEvent>() {
@Override
public TestEvent getEvent() {
return event;
@@ -111,7 +113,7 @@ public class AsyncProcessorTest {
final CountDownLatch latch = new CountDownLatch( 1 );
//mock up the ack to allow us to block the test until the async confirm fires
- when( queue.remove( asynchronousEvent ) ).thenAnswer( new Answer<Boolean>() {
+ when( queue.remove( asynchronousMessage ) ).thenAnswer( new Answer<Boolean>() {
@Override
public Boolean answer( final InvocationOnMock invocation ) throws Throwable {
latch.countDown();
@@ -120,7 +122,7 @@ public class AsyncProcessorTest {
} );
- asyncProcessor.start( asynchronousEvent );
+ asyncProcessor.start( asynchronousMessage );
//block until the event is fired. The correct invocation is implicitly verified by the remove mock
@@ -145,7 +147,7 @@ public class AsyncProcessorTest {
final boolean[] invoked = new boolean[] { false, false };
- final AsynchronousEvent<TestEvent> asynchronousEvent = new AsynchronousEvent<TestEvent>() {
+ final AsynchronousMessage<TestEvent> asynchronousMessage = new AsynchronousMessage<TestEvent>() {
@Override
public TestEvent getEvent() {
return event;
@@ -165,12 +167,12 @@ public class AsyncProcessorTest {
final CountDownLatch latch = new CountDownLatch( 1 );
- final AsynchronousEvent<?>[] errorEvents = { null };
+ final AsynchronousMessage<?>[] errorEvents = { null };
//countdown the latch so the test can proceed
asyncProcessor.addErrorListener( new ErrorListener<TestEvent>() {
@Override
- public void onError( final AsynchronousEvent<TestEvent> event, final Throwable t ) {
+ public void onError( final AsynchronousMessage<TestEvent> event, final Throwable t ) {
errorEvents[0] = event;
invoked[1] = true;
latch.countDown();
@@ -179,7 +181,7 @@ public class AsyncProcessorTest {
} );
//throw an error if remove is called. This shouldn't happen
- when( queue.remove( asynchronousEvent ) ).then( new Answer<Boolean>() {
+ when( queue.remove( asynchronousMessage ) ).then( new Answer<Boolean>() {
@Override
public Boolean answer( final InvocationOnMock invocation ) throws Throwable {
invoked[0] = true;
@@ -189,7 +191,7 @@ public class AsyncProcessorTest {
//fire the event
- asyncProcessor.start( asynchronousEvent );
+ asyncProcessor.start( asynchronousMessage );
//block until the event is fired. The invocation verification is part of the error listener unlocking
@@ -210,7 +212,7 @@ public class AsyncProcessorTest {
/**
* Construct the async processor
*/
- public <T> AsyncProcessorImpl<T> constructProcessor( TimeoutQueue<T> queue , AsynchronousEventListener<T> listener) {
+ public <T> AsyncProcessorImpl<T> constructProcessor( TimeoutQueue<T> queue , MessageListener<T, T> listener) {
AsyncProcessorImpl<T> processor = new AsyncProcessorImpl( queue,Schedulers.threadPoolForIO() );
processor.addListener( listener );
@@ -230,7 +232,7 @@ public class AsyncProcessorTest {
}
- public static class TestListener implements AsynchronousEventListener<TestEvent> {
+ public static class TestListener implements MessageListener<TestEvent, TestEvent> {
public final Stack<TestEvent> events = new Stack<TestEvent>();
@@ -239,9 +241,16 @@ public class AsyncProcessorTest {
}
+
@Override
- public void receive( final TestEvent event ) {
- events.push( event );
+ public Observable<TestEvent> receive( final TestEvent event ) {
+
+ return Observable.from( event ).doOnEach(new Action1<TestEvent>() {
+ @Override
+ public void call( final TestEvent testEvent ) {
+ events.push( testEvent );
+ }
+ });
}
}
@@ -249,15 +258,22 @@ public class AsyncProcessorTest {
/**
* Throw error after the event is fired
*/
- public static class AsynchronousErrorListener implements AsynchronousEventListener<TestEvent> {
+ public static class AsynchronousErrorListener implements MessageListener<TestEvent, TestEvent> {
public final Stack<TestEvent> events = new Stack<TestEvent>();
@Override
- public void receive( final TestEvent event ) {
- events.push( event );
- throw new RuntimeException( "Test Exception thrown. Failed to process event" );
+ public Observable<TestEvent> receive( final TestEvent event ) {
+ return Observable.from( event ).doOnEach(new Action1<TestEvent>() {
+ @Override
+ public void call( final TestEvent testEvent ) {
+ events.push( testEvent );
+ throw new RuntimeException( "Test Exception thrown. Failed to process event" );
+ }
+ });
}
+
+
}
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/da4630f5/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/consistency/LocalTimeoutQueueTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/consistency/LocalTimeoutQueueTest.java b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/consistency/LocalTimeoutQueueTest.java
index 76e09f6..49ae789 100644
--- a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/consistency/LocalTimeoutQueueTest.java
+++ b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/consistency/LocalTimeoutQueueTest.java
@@ -34,14 +34,14 @@ public class LocalTimeoutQueueTest {
final TestEvent event = new TestEvent();
- AsynchronousEvent<TestEvent> asynchronousEvent = queue.queue( event, timeout );
+ AsynchronousMessage<TestEvent> asynchronousMessage = queue.queue( event, timeout );
- assertNotNull( asynchronousEvent );
+ assertNotNull( asynchronousMessage );
- assertEquals( event, asynchronousEvent.getEvent() );
- assertEquals( time + timeout, asynchronousEvent.getTimeout() );
+ assertEquals( event, asynchronousMessage.getEvent() );
+ assertEquals( time + timeout, asynchronousMessage.getTimeout() );
- Collection<AsynchronousEvent<TestEvent>> results = queue.take( 100, timeout );
+ Collection<AsynchronousMessage<TestEvent>> results = queue.take( 100, timeout );
assertEquals( "Time not yet elapsed", 0, results.size() );
@@ -55,9 +55,9 @@ public class LocalTimeoutQueueTest {
assertEquals( "Time elapsed", 1, results.size() );
//validate we get a new timeout event since the old one was re-scheduled
- Iterator<AsynchronousEvent<TestEvent>> events = results.iterator();
+ Iterator<AsynchronousMessage<TestEvent>> events = results.iterator();
- AsynchronousEvent<TestEvent> message = events.next();
+ AsynchronousMessage<TestEvent> message = events.next();
assertEquals( event, message.getEvent() );
@@ -97,18 +97,18 @@ public class LocalTimeoutQueueTest {
final TestEvent event = new TestEvent();
- AsynchronousEvent<TestEvent> asynchronousEvent = queue.queue( event, timeout );
+ AsynchronousMessage<TestEvent> asynchronousMessage = queue.queue( event, timeout );
events.add( event );
- assertNotNull( asynchronousEvent );
+ assertNotNull( asynchronousMessage );
- assertEquals( event, asynchronousEvent.getEvent() );
- assertEquals( time + timeout, asynchronousEvent.getTimeout() );
+ assertEquals( event, asynchronousMessage.getEvent() );
+ assertEquals( time + timeout, asynchronousMessage.getTimeout() );
}
- Collection<AsynchronousEvent<TestEvent>> results = queue.take( 100, timeout );
+ Collection<AsynchronousMessage<TestEvent>> results = queue.take( 100, timeout );
assertEquals( "Time not yet elapsed", 0, results.size() );
@@ -133,11 +133,11 @@ public class LocalTimeoutQueueTest {
assertEquals( "Time elapsed", 100, results.size() );
//validate we get a new timeout event since the old one was re-scheduled
- Iterator<AsynchronousEvent<TestEvent>> eventIterator = results.iterator();
+ Iterator<AsynchronousMessage<TestEvent>> eventIterator = results.iterator();
while(eventIterator.hasNext()){
- AsynchronousEvent<TestEvent> message = eventIterator.next();
+ AsynchronousMessage<TestEvent> message = eventIterator.next();
assertTrue( events.remove( message.getEvent() ) );