You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by to...@apache.org on 2015/02/04 20:34:37 UTC
incubator-usergrid git commit: Separated short circuit logic for
entities and connections
Repository: incubator-usergrid
Updated Branches:
refs/heads/two-dot-o-import 4b226cbb4 -> aafa3d47d
Separated short circuit logic for entities and connections
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/aafa3d47
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/aafa3d47
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/aafa3d47
Branch: refs/heads/two-dot-o-import
Commit: aafa3d47d49ac57b385938989a64dd34c53a5975
Parents: 4b226cb
Author: Todd Nine <tn...@apigee.com>
Authored: Wed Feb 4 12:34:34 2015 -0700
Committer: Todd Nine <tn...@apigee.com>
Committed: Wed Feb 4 12:34:34 2015 -0700
----------------------------------------------------------------------
.../importer/FileImportStatistics.java | 14 ++++++-
.../management/importer/ImportServiceImpl.java | 42 +++++++++++++++-----
2 files changed, 45 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/aafa3d47/stack/services/src/main/java/org/apache/usergrid/management/importer/FileImportStatistics.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/management/importer/FileImportStatistics.java b/stack/services/src/main/java/org/apache/usergrid/management/importer/FileImportStatistics.java
index 055fae7..04ddcdf 100644
--- a/stack/services/src/main/java/org/apache/usergrid/management/importer/FileImportStatistics.java
+++ b/stack/services/src/main/java/org/apache/usergrid/management/importer/FileImportStatistics.java
@@ -192,12 +192,13 @@ public class FileImportStatistics {
return getConnectionsFailed() + getConnectionsWritten();
}
+
/**
* Returns true if we should stop processing. This will use the following logic
*
* We've attempted to import over 1k entities After 1k, we have over a 50% failure rate
*/
- public boolean shouldStopProcessing() {
+ public boolean shouldStopProcessingEntities() {
//TODO Dave, George. What algorithm should we use here?
return false;
@@ -205,6 +206,17 @@ public class FileImportStatistics {
/**
+ * Returns true if we should stop processing. This will use the following logic
+ *
+ * We've attempted to import over 1k connections After 1k, we have over a 50% failure rate
+ */
+ public boolean shouldStopProcessingConnections() {
+
+ //TODO Dave, George. What algorithm should we use here?
+ return false;
+ }
+
+ /**
* Get the number of entities written
* @return
*/
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/aafa3d47/stack/services/src/main/java/org/apache/usergrid/management/importer/ImportServiceImpl.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/management/importer/ImportServiceImpl.java b/stack/services/src/main/java/org/apache/usergrid/management/importer/ImportServiceImpl.java
index 1f8a33e..238c0ab 100644
--- a/stack/services/src/main/java/org/apache/usergrid/management/importer/ImportServiceImpl.java
+++ b/stack/services/src/main/java/org/apache/usergrid/management/importer/ImportServiceImpl.java
@@ -756,6 +756,9 @@ public class ImportServiceImpl implements ImportService {
// observable that parses JSON and emits write events
JsonParser jp = getJsonParserForFile(file);
+
+ //TODO, move the json parser into the observable creation so that open/close happens automatcially within the stream
+
final JsonEntityParserObservable jsonObservableEntities =
new JsonEntityParserObservable(jp, em, rootEm, fileImport, entitiesOnly);
final Observable<WriteEvent> entityEventObservable = Observable.create(jsonObservableEntities);
@@ -768,16 +771,27 @@ public class ImportServiceImpl implements ImportService {
// function to execute for each write event
+ /**
+ * Function that invokes the work of the event.
+ */
final Action1<WriteEvent> doWork = new Action1<WriteEvent>() {
@Override
- public void call(WriteEvent writeEvent) {
- writeEvent.doWrite(em,fileImport, statistics);
+ public void call( WriteEvent writeEvent ) {
+ writeEvent.doWrite( em, fileImport, statistics );
}
};
+
+
// start parsing JSON
+ //only take while our stats tell us we should continue processing
//potentially skip the first n if this is a resume operation
- entityEventObservable.skip( entityNumSkip ).parallel( new Func1<Observable<WriteEvent>, Observable<WriteEvent>>() {
+ entityEventObservable.takeWhile( new Func1<WriteEvent, Boolean>() {
+ @Override
+ public Boolean call( final WriteEvent writeEvent ) {
+ return !statistics.shouldStopProcessingEntities();
+ }
+ } ).skip( entityNumSkip ).parallel( new Func1<Observable<WriteEvent>, Observable<WriteEvent>>() {
@Override
public Observable<WriteEvent> call( Observable<WriteEvent> entityWrapperObservable ) {
@@ -785,6 +799,8 @@ public class ImportServiceImpl implements ImportService {
return entityWrapperObservable.doOnNext( doWork );
}
}, Schedulers.io() ).toBlocking().last();
+
+
jp.close();
logger.debug("\n\nWrote entities\n");
@@ -797,16 +813,22 @@ public class ImportServiceImpl implements ImportService {
final JsonEntityParserObservable jsonObservableOther =
- new JsonEntityParserObservable(jp, em, rootEm, fileImport, entitiesOnly);
- final Observable<WriteEvent> otherEventObservable = Observable.create(jsonObservableOther);
+ new JsonEntityParserObservable( jp, em, rootEm, fileImport, entitiesOnly );
+ final Observable<WriteEvent> otherEventObservable = Observable.create( jsonObservableOther );
- otherEventObservable.skip( connectionNumSkip ).parallel(new Func1<Observable<WriteEvent>, Observable<WriteEvent>>() {
+ //only take while our stats tell us we should continue processing
+ //potentially skip the first n if this is a resume operation
+ otherEventObservable.takeWhile( new Func1<WriteEvent, Boolean>() {
@Override
- public Observable<WriteEvent> call(Observable<WriteEvent> entityWrapperObservable) {
- return entityWrapperObservable.doOnNext(doWork);
-
+ public Boolean call( final WriteEvent writeEvent ) {
+ return !statistics.shouldStopProcessingConnections();
}
- }, Schedulers.io()).toBlocking().last();
+ } ).skip( connectionNumSkip ).parallel( new Func1<Observable<WriteEvent>, Observable<WriteEvent>>() {
+ @Override
+ public Observable<WriteEvent> call( Observable<WriteEvent> entityWrapperObservable ) {
+ return entityWrapperObservable.doOnNext( doWork );
+ }
+ }, Schedulers.io() ).toBlocking().last();
jp.close();