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 23:48:04 UTC
incubator-usergrid git commit: Updated filename to more accurately
reflect it's function.
Repository: incubator-usergrid
Updated Branches:
refs/heads/two-dot-o-import aafa3d47d -> 57ae8164b
Updated filename to more accurately reflect it's function.
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/57ae8164
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/57ae8164
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/57ae8164
Branch: refs/heads/two-dot-o-import
Commit: 57ae8164b6ea290ec2d77c04d93d73a8cd2ead32
Parents: aafa3d4
Author: Todd Nine <tn...@apigee.com>
Authored: Wed Feb 4 15:48:01 2015 -0700
Committer: Todd Nine <tn...@apigee.com>
Committed: Wed Feb 4 15:48:01 2015 -0700
----------------------------------------------------------------------
.../importer/FileImportStatistics.java | 339 ----------------
.../management/importer/FileImportTracker.java | 333 ++++++++++++++++
.../management/importer/ImportServiceImpl.java | 32 +-
.../importer/FileImportStatisticsTest.java | 358 -----------------
.../importer/FileImportTrackerTest.java | 393 +++++++++++++++++++
5 files changed, 742 insertions(+), 713 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/57ae8164/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
deleted file mode 100644
index 04ddcdf..0000000
--- a/stack/services/src/main/java/org/apache/usergrid/management/importer/FileImportStatistics.java
+++ /dev/null
@@ -1,339 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.usergrid.management.importer;
-
-
-import java.util.UUID;
-import java.util.concurrent.Semaphore;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
-
-import org.apache.usergrid.persistence.EntityManager;
-import org.apache.usergrid.persistence.entities.FailedImportConnection;
-import org.apache.usergrid.persistence.entities.FailedImportEntity;
-import org.apache.usergrid.persistence.entities.FileImport;
-import org.apache.usergrid.persistence.exceptions.EntityNotFoundException;
-import org.apache.usergrid.persistence.exceptions.PersistenceException;
-
-
-/**
- * Statistics used to track a file import. Only 1 instance of this class should exist per file imported in the cluster.
- * There is a direct 1-1 mapping of the statistics provided here and the file import status. This class is threadsafe to
- * be used across multiple threads.
- */
-public class FileImportStatistics {
-
- private static final String ERROR_MESSAGE = "Failed to import some data. See the import counters and errors.";
-
-
- private static final String ERRORS_CONNECTION_NAME = "errors";
-
- private final AtomicLong entitiesWritten = new AtomicLong( 0 );
- private final AtomicLong entitiesFailed = new AtomicLong( 0 );
- private final AtomicLong connectionsWritten = new AtomicLong( 0 );
- private final AtomicLong connectionsFailed = new AtomicLong( 0 );
- private final AtomicInteger cachedOperations = new AtomicInteger( 0 );
-
- private final Semaphore writeSemaphore = new Semaphore( 1 );
-
- private final FileImport fileImport;
- private final EntityManager entityManager;
- private final int flushCount;
-
-
- /**
- * Create an instance to track counters. Note that when this instance is created, it will attempt to load it's state
- * from the entity manager. In the case of using this when resuming, be sure you begin processing where the system thinks
- * it has left off.
- *
- * @param entityManager The entity manager that will hold these entities.
- * @param fileImportId The uuid of the fileImport
- * @param flushCount The number of success + failures to accumulate before flushing
- */
- public FileImportStatistics( final EntityManager entityManager, final UUID fileImportId, final int flushCount ) {
- this.entityManager = entityManager;
- this.flushCount = flushCount;
- this.fileImport = getFileImport( fileImportId );
-
- this.entitiesWritten.addAndGet( fileImport.getImportedEntityCount() );
- this.entitiesFailed.addAndGet( fileImport.getFailedEntityCount() );
-
- this.connectionsWritten.addAndGet( fileImport.getImportedConnectionCount() );
- this.connectionsFailed.addAndGet( fileImport.getFailedConnectionCount() );
- }
-
-
- /**
- * Invoke when an entity has been successfully written
- */
- public void entityWritten() {
- entitiesWritten.incrementAndGet();
- maybeFlush();
- }
-
-
-
-
- /**
- * Invoke when an entity fails to write correctly
- */
-
- public void entityFailed( final String message ) {
- entitiesFailed.incrementAndGet();
-
-
- FailedImportEntity failedImportEntity = new FailedImportEntity();
- failedImportEntity.setErrorMessage( message );
-
- try {
- failedImportEntity = entityManager.create( failedImportEntity );
- entityManager.createConnection( fileImport, ERRORS_CONNECTION_NAME, failedImportEntity );
- }
- catch ( Exception e ) {
- throw new PersistenceException( "Unable to save failed entity import message", e );
- }
- maybeFlush();
- }
-
-
- /**
- * Invoked when a connection is written
- */
- public void connectionWritten() {
- connectionsWritten.incrementAndGet();
- maybeFlush();
- }
-
-
- /**
- * Invoked when a connection cannot be written
- */
- public void connectionFailed( final String message ) {
- connectionsFailed.incrementAndGet();
-
-
- FailedImportConnection failedImportConnection = new FailedImportConnection();
- failedImportConnection.setErrorMessage( message );
-
- try {
- failedImportConnection = entityManager.create( failedImportConnection );
- entityManager.createConnection( fileImport, ERRORS_CONNECTION_NAME, failedImportConnection );
- }
- catch ( Exception e ) {
- throw new PersistenceException( "Unable to save failed entity import message", e );
- }
- maybeFlush();
- }
-
-
- /**
- * Invoke when the file is completed processing
- */
- public void complete() {
-
- final long failed = entitiesFailed.get() + connectionsFailed.get();
-
- final FileImport.State state;
- final String message;
-
- if ( failed > 0 ) {
- state = FileImport.State.FAILED;
- message = ERROR_MESSAGE;
- }
- else {
- state = FileImport.State.FINISHED;
- message = null;
- }
-
- updateFileImport( state, message );
- }
-
-
- /**
- * Invoke when we halt the import with a fatal error that cannot be recovered.
- */
- public void fatal( final String message ) {
-
- updateFileImport( FileImport.State.FAILED, message );
- }
-
-
- /**
- * Return the total number of successful imports + failed imports. Can be used in resume. Note that this reflects
- * the counts last written to cassandra when this instance was created + any processing
- */
- public long getTotalEntityCount() {
- return getEntitiesWritten() + getEntitiesFailed();
- }
-
-
- /**
- * Get the total number of failed + successful connections
- * @return
- */
- public long getTotalConnectionCount(){
- 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 shouldStopProcessingEntities() {
-
- //TODO Dave, George. What algorithm should we use here?
- return false;
- }
-
-
- /**
- * 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
- */
- public long getEntitiesWritten() {
- return entitiesWritten.get();
- }
-
-
- /**
- * Get the number of failed entities
- * @return
- */
- public long getEntitiesFailed() {
- return entitiesFailed.get();
- }
-
-
- /**
- * Get the number of connections written
- * @return
- */
- public long getConnectionsWritten() {
- return connectionsWritten.get();
- }
-
-
- /**
- * Get the number of connections failed
- * @return
- */
- public long getConnectionsFailed() {
- return connectionsFailed.get();
- }
-
- private void maybeFlush() {
- final int count = cachedOperations.incrementAndGet();
-
- //no op
- if ( count < flushCount ) {
- return;
- }
-
- //another thread is writing, no op, just return
- if ( !writeSemaphore.tryAcquire() ) {
- return;
- }
-
- final long failed = entitiesFailed.get();
- final long written = entitiesWritten.get();
- final String message;
-
- if ( failed > 0 ) {
- message = "Failed to import " + failed + " entities. Successfully imported " + written + " entities";
- }
- else {
- message = "Successfully imported " + written + " entities";
- }
-
- updateFileImport( FileImport.State.STARTED, message );
- cachedOperations.addAndGet( flushCount * -1 );
- writeSemaphore.release();
- }
-
-
- /**
- * Update the file import status with the provided messages
- *
- * @param state The state to set into the import
- * @param message The message to set
- */
- private void updateFileImport( final FileImport.State state, final String message ) {
-
- try {
-
-
- final long writtenEntities = entitiesWritten.get();
- final long failedEntities = entitiesFailed.get();
-
- final long writtenConnections = connectionsFailed.get();
- final long failedConnections = connectionsFailed.get();
-
-
- fileImport.setImportedEntityCount( writtenEntities );
- fileImport.setFailedEntityCount( failedEntities );
- fileImport.setImportedConnectionCount( writtenConnections );
- fileImport.setFailedConnectionCount( failedConnections );
- fileImport.setState( state );
- fileImport.setErrorMessage( message );
-
- entityManager.update( fileImport );
- }
- catch ( Exception e ) {
- throw new RuntimeException( "Unable to persist complete state", e );
- }
- }
-
-
- /**
- * Get the FileImport by uuid and return it
- *
- * @throws EntityNotFoundException if we can't find the file import with the given uuid
- */
- private FileImport getFileImport( final UUID fileImportId ) {
-
- final FileImport fileImport;
-
- try {
- fileImport = entityManager.get( fileImportId, FileImport.class );
- }
- catch ( Exception e ) {
- throw new RuntimeException( "Unable to load fileImport with id " + fileImportId, e );
- }
-
- if ( fileImport == null ) {
- throw new EntityNotFoundException( "Could not file FileImport with id " + fileImportId );
- }
-
- return fileImport;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/57ae8164/stack/services/src/main/java/org/apache/usergrid/management/importer/FileImportTracker.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/management/importer/FileImportTracker.java b/stack/services/src/main/java/org/apache/usergrid/management/importer/FileImportTracker.java
new file mode 100644
index 0000000..8a01259
--- /dev/null
+++ b/stack/services/src/main/java/org/apache/usergrid/management/importer/FileImportTracker.java
@@ -0,0 +1,333 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.usergrid.management.importer;
+
+
+import java.util.UUID;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.usergrid.persistence.EntityManager;
+import org.apache.usergrid.persistence.entities.FailedImportConnection;
+import org.apache.usergrid.persistence.entities.FailedImportEntity;
+import org.apache.usergrid.persistence.entities.FileImport;
+import org.apache.usergrid.persistence.exceptions.EntityNotFoundException;
+import org.apache.usergrid.persistence.exceptions.PersistenceException;
+
+
+/**
+ * Statistics used to track a file import. Only 1 instance of this class should exist per file imported in the cluster.
+ * There is a direct 1-1 mapping of the statistics provided here and the file import status. This class is threadsafe to
+ * be used across multiple threads.
+ */
+public class FileImportTracker {
+
+ private static final String ERROR_MESSAGE = "Failed to import some data. See the import counters and errors.";
+
+
+ private static final String ERRORS_CONNECTION_NAME = "errors";
+
+ private final AtomicLong entitiesWritten = new AtomicLong( 0 );
+ private final AtomicLong entitiesFailed = new AtomicLong( 0 );
+ private final AtomicLong connectionsWritten = new AtomicLong( 0 );
+ private final AtomicLong connectionsFailed = new AtomicLong( 0 );
+ private final AtomicInteger cachedOperations = new AtomicInteger( 0 );
+
+ private final Semaphore writeSemaphore = new Semaphore( 1 );
+
+ private final FileImport fileImport;
+ private final EntityManager entityManager;
+ private final int flushCount;
+
+
+ /**
+ * Create an instance to track counters. Note that when this instance is created, it will attempt to load it's state
+ * from the entity manager. In the case of using this when resuming, be sure you begin processing where the system thinks
+ * it has left off.
+ *
+ * @param entityManager The entity manager that will hold these entities.
+ * @param fileImportId The uuid of the fileImport
+ * @param flushCount The number of success + failures to accumulate before flushing
+ */
+ public FileImportTracker( final EntityManager entityManager, final UUID fileImportId, final int flushCount ) {
+ this.entityManager = entityManager;
+ this.flushCount = flushCount;
+ this.fileImport = getFileImport( fileImportId );
+
+ this.entitiesWritten.addAndGet( fileImport.getImportedEntityCount() );
+ this.entitiesFailed.addAndGet( fileImport.getFailedEntityCount() );
+
+ this.connectionsWritten.addAndGet( fileImport.getImportedConnectionCount() );
+ this.connectionsFailed.addAndGet( fileImport.getFailedConnectionCount() );
+ }
+
+
+ /**
+ * Invoke when an entity has been successfully written
+ */
+ public void entityWritten() {
+ entitiesWritten.incrementAndGet();
+ maybeFlush();
+ }
+
+
+
+
+ /**
+ * Invoke when an entity fails to write correctly
+ */
+
+ public void entityFailed( final String message ) {
+ entitiesFailed.incrementAndGet();
+
+
+ FailedImportEntity failedImportEntity = new FailedImportEntity();
+ failedImportEntity.setErrorMessage( message );
+
+ try {
+ failedImportEntity = entityManager.create( failedImportEntity );
+ entityManager.createConnection( fileImport, ERRORS_CONNECTION_NAME, failedImportEntity );
+ }
+ catch ( Exception e ) {
+ throw new PersistenceException( "Unable to save failed entity import message", e );
+ }
+ maybeFlush();
+ }
+
+
+ /**
+ * Invoked when a connection is written
+ */
+ public void connectionWritten() {
+ connectionsWritten.incrementAndGet();
+ maybeFlush();
+ }
+
+
+ /**
+ * Invoked when a connection cannot be written
+ */
+ public void connectionFailed( final String message ) {
+ connectionsFailed.incrementAndGet();
+
+
+ FailedImportConnection failedImportConnection = new FailedImportConnection();
+ failedImportConnection.setErrorMessage( message );
+
+ try {
+ failedImportConnection = entityManager.create( failedImportConnection );
+ entityManager.createConnection( fileImport, ERRORS_CONNECTION_NAME, failedImportConnection );
+ }
+ catch ( Exception e ) {
+ throw new PersistenceException( "Unable to save failed entity import message", e );
+ }
+ maybeFlush();
+ }
+
+
+ /**
+ * Invoke when the file is completed processing
+ */
+ public void complete() {
+
+ final long failed = entitiesFailed.get() + connectionsFailed.get();
+
+ final FileImport.State state;
+ final String message;
+
+ if ( failed > 0 ) {
+ state = FileImport.State.FAILED;
+ message = ERROR_MESSAGE;
+ }
+ else {
+ state = FileImport.State.FINISHED;
+ message = null;
+ }
+
+ updateFileImport( state, message );
+ }
+
+
+ /**
+ * Invoke when we halt the import with a fatal error that cannot be recovered.
+ */
+ public void fatal( final String message ) {
+
+ updateFileImport( FileImport.State.FAILED, message );
+ }
+
+
+ /**
+ * Return the total number of successful imports + failed imports. Can be used in resume. Note that this reflects
+ * the counts last written to cassandra when this instance was created + any processing
+ */
+ public long getTotalEntityCount() {
+ return getEntitiesWritten() + getEntitiesFailed();
+ }
+
+
+ /**
+ * Get the total number of failed + successful connections
+ * @return
+ */
+ public long getTotalConnectionCount(){
+ return getConnectionsFailed() + getConnectionsWritten();
+ }
+
+
+ /**
+ * Returns true if we should stop processing. We use fail fast logic, so after the first
+ * failure this will return true.
+ */
+ public boolean shouldStopProcessingEntities() {
+ return entitiesFailed.get() > 0;
+ }
+
+
+ /**
+ * Returns true if we should stop processing. We use fail fast logic, so after the first
+ * failure this will return true.
+ */
+ public boolean shouldStopProcessingConnections() {
+ return connectionsFailed.get() > 0;
+ }
+
+ /**
+ * Get the number of entities written
+ * @return
+ */
+ public long getEntitiesWritten() {
+ return entitiesWritten.get();
+ }
+
+
+ /**
+ * Get the number of failed entities
+ * @return
+ */
+ public long getEntitiesFailed() {
+ return entitiesFailed.get();
+ }
+
+
+ /**
+ * Get the number of connections written
+ * @return
+ */
+ public long getConnectionsWritten() {
+ return connectionsWritten.get();
+ }
+
+
+ /**
+ * Get the number of connections failed
+ * @return
+ */
+ public long getConnectionsFailed() {
+ return connectionsFailed.get();
+ }
+
+ private void maybeFlush() {
+ final int count = cachedOperations.incrementAndGet();
+
+ //no op
+ if ( count < flushCount ) {
+ return;
+ }
+
+ //another thread is writing, no op, just return
+ if ( !writeSemaphore.tryAcquire() ) {
+ return;
+ }
+
+ final long failed = entitiesFailed.get();
+ final long written = entitiesWritten.get();
+ final String message;
+
+ if ( failed > 0 ) {
+ message = "Failed to import " + failed + " entities. Successfully imported " + written + " entities";
+ }
+ else {
+ message = "Successfully imported " + written + " entities";
+ }
+
+ updateFileImport( FileImport.State.STARTED, message );
+ cachedOperations.addAndGet( flushCount * -1 );
+ writeSemaphore.release();
+ }
+
+
+ /**
+ * Update the file import status with the provided messages
+ *
+ * @param state The state to set into the import
+ * @param message The message to set
+ */
+ private void updateFileImport( final FileImport.State state, final String message ) {
+
+ try {
+
+
+ final long writtenEntities = entitiesWritten.get();
+ final long failedEntities = entitiesFailed.get();
+
+ final long writtenConnections = connectionsFailed.get();
+ final long failedConnections = connectionsFailed.get();
+
+
+ fileImport.setImportedEntityCount( writtenEntities );
+ fileImport.setFailedEntityCount( failedEntities );
+ fileImport.setImportedConnectionCount( writtenConnections );
+ fileImport.setFailedConnectionCount( failedConnections );
+ fileImport.setState( state );
+ fileImport.setErrorMessage( message );
+
+ entityManager.update( fileImport );
+ }
+ catch ( Exception e ) {
+ throw new RuntimeException( "Unable to persist complete state", e );
+ }
+ }
+
+
+ /**
+ * Get the FileImport by uuid and return it
+ *
+ * @throws EntityNotFoundException if we can't find the file import with the given uuid
+ */
+ private FileImport getFileImport( final UUID fileImportId ) {
+
+ final FileImport fileImport;
+
+ try {
+ fileImport = entityManager.get( fileImportId, FileImport.class );
+ }
+ catch ( Exception e ) {
+ throw new RuntimeException( "Unable to load fileImport with id " + fileImportId, e );
+ }
+
+ if ( fileImport == null ) {
+ throw new EntityNotFoundException( "Could not file FileImport with id " + fileImportId );
+ }
+
+ return fileImport;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/57ae8164/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 238c0ab..5646f46 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
@@ -764,10 +764,10 @@ public class ImportServiceImpl implements ImportService {
final Observable<WriteEvent> entityEventObservable = Observable.create(jsonObservableEntities);
//flush every 100 entities
- final FileImportStatistics statistics = new FileImportStatistics( em, fileImport.getUuid(), 100 );
+ final FileImportTracker tracker = new FileImportTracker( em, fileImport.getUuid(), 100 );
//truncate due to RX api
- final int entityNumSkip = (int)statistics.getTotalEntityCount();
- final int connectionNumSkip = (int)statistics.getTotalConnectionCount();
+ final int entityNumSkip = (int)tracker.getTotalEntityCount();
+ final int connectionNumSkip = (int)tracker.getTotalConnectionCount();
// function to execute for each write event
@@ -777,7 +777,7 @@ public class ImportServiceImpl implements ImportService {
final Action1<WriteEvent> doWork = new Action1<WriteEvent>() {
@Override
public void call( WriteEvent writeEvent ) {
- writeEvent.doWrite( em, fileImport, statistics );
+ writeEvent.doWrite( em, fileImport, tracker );
}
};
@@ -789,7 +789,7 @@ public class ImportServiceImpl implements ImportService {
entityEventObservable.takeWhile( new Func1<WriteEvent, Boolean>() {
@Override
public Boolean call( final WriteEvent writeEvent ) {
- return !statistics.shouldStopProcessingEntities();
+ return !tracker.shouldStopProcessingEntities();
}
} ).skip( entityNumSkip ).parallel( new Func1<Observable<WriteEvent>, Observable<WriteEvent>>() {
@Override
@@ -803,7 +803,7 @@ public class ImportServiceImpl implements ImportService {
jp.close();
- logger.debug("\n\nWrote entities\n");
+ logger.debug( "\n\nWrote entities\n" );
// now do other stuff: connections and dictionaries
entitiesOnly = false;
@@ -821,7 +821,7 @@ public class ImportServiceImpl implements ImportService {
otherEventObservable.takeWhile( new Func1<WriteEvent, Boolean>() {
@Override
public Boolean call( final WriteEvent writeEvent ) {
- return !statistics.shouldStopProcessingConnections();
+ return !tracker.shouldStopProcessingConnections();
}
} ).skip( connectionNumSkip ).parallel( new Func1<Observable<WriteEvent>, Observable<WriteEvent>>() {
@Override
@@ -833,14 +833,14 @@ public class ImportServiceImpl implements ImportService {
jp.close();
//flush the job statistics
- statistics.complete();
+ tracker.complete();
logger.debug("\n\nWrote others\n");
}
private interface WriteEvent {
- public void doWrite(EntityManager em, FileImport fileImport, FileImportStatistics stats);
+ public void doWrite(EntityManager em, FileImport fileImport, FileImportTracker tracker);
}
@@ -861,19 +861,19 @@ public class ImportServiceImpl implements ImportService {
// Creates entities
@Override
- public void doWrite(EntityManager em, FileImport fileImport, FileImportStatistics stats) {
+ public void doWrite(EntityManager em, FileImport fileImport, FileImportTracker tracker) {
try {
logger.debug("Writing imported entity {}:{} into app {}",
new Object[]{entityType, entityUuid, em.getApplication().getUuid()});
em.create(entityUuid, entityType, properties);
- stats.entityWritten();
+ tracker.entityWritten();
} catch (Exception e) {
logger.error("Error writing entity", e);
- stats.entityFailed( e.getMessage() );
+ tracker.entityFailed( e.getMessage() );
}
}
}
@@ -893,7 +893,7 @@ public class ImportServiceImpl implements ImportService {
// creates connections between entities
@Override
- public void doWrite(EntityManager em, FileImport fileImport, FileImportStatistics stats) {
+ public void doWrite(EntityManager em, FileImport fileImport, FileImportTracker tracker) {
try {
// TODO: do we need to ensure that all Entity events happen first?
@@ -911,11 +911,11 @@ public class ImportServiceImpl implements ImportService {
em.createConnection(ownerEntityRef, connectionType, entityRef);
- stats.connectionWritten();
+ tracker.connectionWritten();
} catch (Exception e) {
logger.error("Error writing connection", e);
- stats.connectionFailed( e.getMessage() );
+ tracker.connectionFailed( e.getMessage() );
}
}
}
@@ -935,7 +935,7 @@ public class ImportServiceImpl implements ImportService {
// adds map to the dictionary
@Override
- public void doWrite(EntityManager em, FileImport fileImport, FileImportStatistics stats) {
+ public void doWrite(EntityManager em, FileImport fileImport, FileImportTracker stats) {
EntityManager rootEm = emf.getEntityManager(CpNamingUtils.MANAGEMENT_APPLICATION_ID);
try {
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/57ae8164/stack/services/src/test/java/org/apache/usergrid/management/importer/FileImportStatisticsTest.java
----------------------------------------------------------------------
diff --git a/stack/services/src/test/java/org/apache/usergrid/management/importer/FileImportStatisticsTest.java b/stack/services/src/test/java/org/apache/usergrid/management/importer/FileImportStatisticsTest.java
deleted file mode 100644
index 510302a..0000000
--- a/stack/services/src/test/java/org/apache/usergrid/management/importer/FileImportStatisticsTest.java
+++ /dev/null
@@ -1,358 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.usergrid.management.importer;
-
-
-import java.util.List;
-import java.util.UUID;
-
-import org.junit.Test;
-import org.mockito.ArgumentCaptor;
-import org.mockito.invocation.InvocationOnMock;
-import org.mockito.stubbing.Answer;
-
-import org.apache.usergrid.persistence.EntityManager;
-import org.apache.usergrid.persistence.entities.FailedImport;
-import org.apache.usergrid.persistence.entities.FailedImportConnection;
-import org.apache.usergrid.persistence.entities.FailedImportEntity;
-import org.apache.usergrid.persistence.entities.FileImport;
-import org.apache.usergrid.persistence.model.util.UUIDGenerator;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertSame;
-import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.eq;
-import static org.mockito.Matchers.same;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
-
-public class FileImportStatisticsTest {
-
- @Test
- public void testSuccess() throws Exception {
-
- final EntityManager em = mock( EntityManager.class );
-
- final UUID importFileId = UUIDGenerator.newTimeUUID();
-
-
- final FileImport fileImport = new FileImport();
- fileImport.setUuid( importFileId );
-
- when( em.get( importFileId, FileImport.class ) ).thenReturn( fileImport );
-
-
- final FileImportStatistics fileImportStatistics = new FileImportStatistics( em, importFileId, 1000 );
-
- final long expectedCount = 100;
-
- for ( long i = 0; i < expectedCount; i++ ) {
- fileImportStatistics.entityWritten();
- }
-
-
- fileImportStatistics.complete();
-
-
- ArgumentCaptor<FileImport> savedFileImport = ArgumentCaptor.forClass( FileImport.class );
-
- verify( em ).update( savedFileImport.capture() );
-
- final FileImport updated = savedFileImport.getValue();
-
- assertSame( "Same instance should be updated", fileImport, updated );
-
-
- assertEquals( "Same count expected", expectedCount, updated.getImportedEntityCount() );
-
- assertNull( updated.getErrorMessage() );
- }
-
-
- @Test
- public void testBoth() throws Exception {
-
- final EntityManager em = mock( EntityManager.class );
-
- final UUID importFileId = UUIDGenerator.newTimeUUID();
-
-
- final FileImport fileImport = new FileImport();
- fileImport.setUuid( importFileId );
-
- when( em.get( importFileId, FileImport.class ) ).thenReturn( fileImport );
-
- //mock up returning the FailedEntityImport instance after save is invoked.
-
- when( em.create( any( FailedImportEntity.class ) ) ).thenAnswer( new Answer<FailedImportEntity>() {
- @Override
- public FailedImportEntity answer( final InvocationOnMock invocation ) throws Throwable {
- return ( FailedImportEntity ) invocation.getArguments()[0];
- }
- } );
-
- final FileImportStatistics fileImportStatistics = new FileImportStatistics( em, importFileId, 1000 );
-
- final long expectedSuccess = 100;
-
- for ( long i = 0; i < expectedSuccess; i++ ) {
- fileImportStatistics.entityWritten();
- }
-
- final int expectedFails = 10;
-
- for ( int i = 0; i < expectedFails; i++ ) {
- fileImportStatistics.entityFailed( "Failed to write entity " + i );
- }
-
-
- fileImportStatistics.complete();
-
-
- ArgumentCaptor<FileImport> savedFileImport = ArgumentCaptor.forClass( FileImport.class );
-
- verify( em ).update( savedFileImport.capture() );
-
- final FileImport updated = savedFileImport.getValue();
-
- assertSame( "Same instance should be updated", fileImport, updated );
-
- assertEquals( "Same count expected", expectedSuccess, updated.getImportedEntityCount() );
-
- assertEquals( "Same fail expected", expectedFails, updated.getFailedEntityCount() );
-
- assertEquals( "Correct error message", "Failed to import some data. See the import counters and errors.",
- updated.getErrorMessage() );
-
- //TODO get the connections from the file import
-
- ArgumentCaptor<FailedImportEntity> failedEntities = ArgumentCaptor.forClass( FailedImportEntity.class );
-
- verify( em, times( expectedFails ) )
- .createConnection( same( fileImport ), eq( "errors" ), failedEntities.capture() );
-
- //now check all our arguments
-
- final List<FailedImportEntity> args = failedEntities.getAllValues();
-
- assertEquals( "Same number of error connections created", expectedFails, args.size() );
-
-
- for ( int i = 0; i < expectedFails; i++ ) {
-
- final FailedImportEntity failedImport = args.get( i );
-
- assertEquals( "Same message expected", "Failed to write entity " + i, failedImport.getErrorMessage() );
- }
- }
-
-
- @Test
- public void explicitFail() throws Exception {
-
- final EntityManager em = mock( EntityManager.class );
-
- final UUID importFileId = UUIDGenerator.newTimeUUID();
-
-
- final FileImport fileImport = new FileImport();
- fileImport.setUuid( importFileId );
-
- when( em.get( importFileId, FileImport.class ) ).thenReturn( fileImport );
-
-
- final FileImportStatistics fileImportStatistics = new FileImportStatistics( em, importFileId, 1000 );
-
- final long expectedCount = 100;
-
- for ( long i = 0; i < expectedCount; i++ ) {
- fileImportStatistics.entityWritten();
- }
-
-
- fileImportStatistics.fatal( "Something bad happened" );
-
-
- ArgumentCaptor<FileImport> savedFileImport = ArgumentCaptor.forClass( FileImport.class );
-
- verify( em ).update( savedFileImport.capture() );
-
- final FileImport updated = savedFileImport.getValue();
-
- assertSame( "Same instance should be updated", fileImport, updated );
-
-
- assertEquals( "Same count expected", expectedCount, updated.getImportedEntityCount() );
-
- assertEquals( "Fail count is 0", 0, updated.getFailedEntityCount() );
-
- assertEquals( "Correct expected message", "Something bad happened", updated.getErrorMessage() );
-
- assertEquals( "Expected failed state", FileImport.State.FAILED, updated.getState() );
- }
-
-
- @Test
- public void testAutoFlushSuccess() throws Exception {
-
- final EntityManager em = mock( EntityManager.class );
-
- final UUID importFileId = UUIDGenerator.newTimeUUID();
-
-
- final FileImport fileImport = new FileImport();
- fileImport.setUuid( importFileId );
-
- when( em.get( importFileId, FileImport.class ) ).thenReturn( fileImport );
-
- //mock up returning the FailedEntityImport instance after save is invoked.
-
-
- when( em.create( any( FailedImportConnection.class ) ) ).thenAnswer( new Answer<Object>() {
- @Override
- public Object answer( final InvocationOnMock invocation ) throws Throwable {
- return invocation.getArguments()[0];
- }
- } );
-
- final int expectedSuccess = 100;
- final int expectedFails = 100;
- final int expectedConnectionSuccess = 100;
- final int expectedConnectionFails = 100;
-
- final int expectedFlushCount = 2;
- final int flushSize = ( expectedFails + expectedFails + expectedConnectionSuccess + expectedConnectionFails )
- / expectedFlushCount;
-
- //set this to 1/2, so that we get saved twice
- final FileImportStatistics fileImportStatistics = new FileImportStatistics( em, importFileId, flushSize );
-
-
- for ( long i = 0; i < expectedSuccess; i++ ) {
- fileImportStatistics.entityWritten();
- }
-
-
- for ( int i = 0; i < expectedFails; i++ ) {
- fileImportStatistics.entityFailed( "Failed to write entity " + i );
- }
-
-
- for ( long i = 0; i < expectedConnectionSuccess; i++ ) {
- fileImportStatistics.connectionWritten();
- }
-
-
- for ( int i = 0; i < expectedConnectionFails; i++ ) {
- fileImportStatistics.connectionFailed( "Failed to write connection " + i );
- }
-
-
- fileImportStatistics.complete();
-
-
- ArgumentCaptor<FileImport> savedFileImport = ArgumentCaptor.forClass( FileImport.class );
-
- verify( em, times( expectedFlushCount + 1 ) ).update( savedFileImport.capture() );
-
- final FileImport updated = savedFileImport.getAllValues().get( 2 );
-
- assertSame( "Same instance should be updated", fileImport, updated );
-
- assertEquals( "Same count expected", expectedSuccess, updated.getImportedEntityCount() );
-
- assertEquals( "Same fail expected", expectedFails, updated.getFailedEntityCount() );
-
- assertEquals( "Same connection count expected", expectedConnectionSuccess,
- updated.getImportedConnectionCount() );
-
- assertEquals( "Same connection error count expected", expectedConnectionFails,
- updated.getFailedConnectionCount() );
-
- assertEquals( "Correct error message", "Failed to import some data. See the import counters and errors.",
- updated.getErrorMessage() );
-
- //TODO get the connections from the file import
-
- ArgumentCaptor<FailedImport> failedEntities = ArgumentCaptor.forClass( FailedImport.class );
-
- verify( em, times( expectedFails + expectedConnectionFails ) )
- .createConnection( same( fileImport ), eq( "errors" ), failedEntities.capture() );
-
- //now check all our arguments
-
- final List<FailedImport> args = failedEntities.getAllValues();
-
- assertEquals( "Same number of error connections created", expectedFails+expectedConnectionFails, args.size() );
-
-
- for ( int i = 0; i < expectedFails; i++ ) {
-
- final FailedImport failedImport = args.get( i );
-
- assertEquals( "Same message expected", "Failed to write entity " + i, failedImport.getErrorMessage() );
- }
-
- for ( int i = expectedFails; i < expectedConnectionFails; i++ ) {
-
- final FailedImport failedImport = args.get( i );
-
- assertEquals( "Same message expected", "Failed to write connection " + i, failedImport.getErrorMessage() );
- }
- }
-
-
- @Test
- public void loadingExistingState() throws Exception {
-
- final EntityManager em = mock( EntityManager.class );
-
- final UUID importFileId = UUIDGenerator.newTimeUUID();
-
-
- final FileImport fileImport = new FileImport();
- fileImport.setUuid( importFileId );
- fileImport.setImportedEntityCount( 1 );
- fileImport.setFailedEntityCount( 2 );
- fileImport.setImportedConnectionCount( 3 );
- fileImport.setFailedConnectionCount( 4 );
-
- when( em.get( importFileId, FileImport.class ) ).thenReturn( fileImport );
-
- //mock up returning the FailedEntityImport instance after save is invoked.
-
- FileImportStatistics statistics = new FileImportStatistics( em, importFileId, 100 );
-
- assertEquals( 1, statistics.getEntitiesWritten() );
- assertEquals( 2, statistics.getEntitiesFailed() );
-
- assertEquals( 3, statistics.getTotalEntityCount() );
-
- assertEquals( 3, statistics.getConnectionsWritten() );
- assertEquals( 4, statistics.getConnectionsFailed() );
-
- assertEquals( 7, statistics.getTotalConnectionCount() );
- }
-}
-
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/57ae8164/stack/services/src/test/java/org/apache/usergrid/management/importer/FileImportTrackerTest.java
----------------------------------------------------------------------
diff --git a/stack/services/src/test/java/org/apache/usergrid/management/importer/FileImportTrackerTest.java b/stack/services/src/test/java/org/apache/usergrid/management/importer/FileImportTrackerTest.java
new file mode 100644
index 0000000..0a58602
--- /dev/null
+++ b/stack/services/src/test/java/org/apache/usergrid/management/importer/FileImportTrackerTest.java
@@ -0,0 +1,393 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.usergrid.management.importer;
+
+
+import java.util.List;
+import java.util.UUID;
+
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+import org.apache.usergrid.persistence.EntityManager;
+import org.apache.usergrid.persistence.entities.FailedImport;
+import org.apache.usergrid.persistence.entities.FailedImportConnection;
+import org.apache.usergrid.persistence.entities.FailedImportEntity;
+import org.apache.usergrid.persistence.entities.FileImport;
+import org.apache.usergrid.persistence.model.util.UUIDGenerator;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Matchers.same;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+
+public class FileImportTrackerTest {
+
+ @Test
+ public void testSuccess() throws Exception {
+
+ final EntityManager em = mock( EntityManager.class );
+
+ final UUID importFileId = UUIDGenerator.newTimeUUID();
+
+
+ final FileImport fileImport = new FileImport();
+ fileImport.setUuid( importFileId );
+
+ when( em.get( importFileId, FileImport.class ) ).thenReturn( fileImport );
+
+
+ final FileImportTracker fileImportTracker = new FileImportTracker( em, importFileId, 1000 );
+
+ final long expectedCount = 100;
+
+ for ( long i = 0; i < expectedCount; i++ ) {
+ fileImportTracker.entityWritten();
+ }
+
+
+ fileImportTracker.complete();
+
+
+ ArgumentCaptor<FileImport> savedFileImport = ArgumentCaptor.forClass( FileImport.class );
+
+ verify( em ).update( savedFileImport.capture() );
+
+ final FileImport updated = savedFileImport.getValue();
+
+ assertSame( "Same instance should be updated", fileImport, updated );
+
+
+ assertEquals( "Same count expected", expectedCount, updated.getImportedEntityCount() );
+
+ assertNull( updated.getErrorMessage() );
+ }
+
+
+ @Test
+ public void testBoth() throws Exception {
+
+ final EntityManager em = mock( EntityManager.class );
+
+ final UUID importFileId = UUIDGenerator.newTimeUUID();
+
+
+ final FileImport fileImport = new FileImport();
+ fileImport.setUuid( importFileId );
+
+ when( em.get( importFileId, FileImport.class ) ).thenReturn( fileImport );
+
+ //mock up returning the FailedEntityImport instance after save is invoked.
+
+ when( em.create( any( FailedImportEntity.class ) ) ).thenAnswer( new Answer<FailedImportEntity>() {
+ @Override
+ public FailedImportEntity answer( final InvocationOnMock invocation ) throws Throwable {
+ return ( FailedImportEntity ) invocation.getArguments()[0];
+ }
+ } );
+
+ final FileImportTracker fileImportTracker = new FileImportTracker( em, importFileId, 1000 );
+
+ final long expectedSuccess = 100;
+
+ for ( long i = 0; i < expectedSuccess; i++ ) {
+ fileImportTracker.entityWritten();
+ }
+
+ final int expectedFails = 10;
+
+ for ( int i = 0; i < expectedFails; i++ ) {
+ fileImportTracker.entityFailed( "Failed to write entity " + i );
+ }
+
+
+ fileImportTracker.complete();
+
+
+ ArgumentCaptor<FileImport> savedFileImport = ArgumentCaptor.forClass( FileImport.class );
+
+ verify( em ).update( savedFileImport.capture() );
+
+ final FileImport updated = savedFileImport.getValue();
+
+ assertSame( "Same instance should be updated", fileImport, updated );
+
+ assertEquals( "Same count expected", expectedSuccess, updated.getImportedEntityCount() );
+
+ assertEquals( "Same fail expected", expectedFails, updated.getFailedEntityCount() );
+
+ assertEquals( "Correct error message", "Failed to import some data. See the import counters and errors.",
+ updated.getErrorMessage() );
+
+ //TODO get the connections from the file import
+
+ ArgumentCaptor<FailedImportEntity> failedEntities = ArgumentCaptor.forClass( FailedImportEntity.class );
+
+ verify( em, times( expectedFails ) )
+ .createConnection( same( fileImport ), eq( "errors" ), failedEntities.capture() );
+
+ //now check all our arguments
+
+ final List<FailedImportEntity> args = failedEntities.getAllValues();
+
+ assertEquals( "Same number of error connections created", expectedFails, args.size() );
+
+
+ for ( int i = 0; i < expectedFails; i++ ) {
+
+ final FailedImportEntity failedImport = args.get( i );
+
+ assertEquals( "Same message expected", "Failed to write entity " + i, failedImport.getErrorMessage() );
+ }
+ }
+
+
+ @Test
+ public void explicitFail() throws Exception {
+
+ final EntityManager em = mock( EntityManager.class );
+
+ final UUID importFileId = UUIDGenerator.newTimeUUID();
+
+
+ final FileImport fileImport = new FileImport();
+ fileImport.setUuid( importFileId );
+
+ when( em.get( importFileId, FileImport.class ) ).thenReturn( fileImport );
+
+
+ final FileImportTracker fileImportTracker = new FileImportTracker( em, importFileId, 1000 );
+
+ final long expectedCount = 100;
+
+ for ( long i = 0; i < expectedCount; i++ ) {
+ fileImportTracker.entityWritten();
+ }
+
+
+ fileImportTracker.fatal( "Something bad happened" );
+
+
+ ArgumentCaptor<FileImport> savedFileImport = ArgumentCaptor.forClass( FileImport.class );
+
+ verify( em ).update( savedFileImport.capture() );
+
+ final FileImport updated = savedFileImport.getValue();
+
+ assertSame( "Same instance should be updated", fileImport, updated );
+
+
+ assertEquals( "Same count expected", expectedCount, updated.getImportedEntityCount() );
+
+ assertEquals( "Fail count is 0", 0, updated.getFailedEntityCount() );
+
+ assertEquals( "Correct expected message", "Something bad happened", updated.getErrorMessage() );
+
+ assertEquals( "Expected failed state", FileImport.State.FAILED, updated.getState() );
+ }
+
+
+ @Test
+ public void testAutoFlushSuccess() throws Exception {
+
+ final EntityManager em = mock( EntityManager.class );
+
+ final UUID importFileId = UUIDGenerator.newTimeUUID();
+
+
+ final FileImport fileImport = new FileImport();
+ fileImport.setUuid( importFileId );
+
+ when( em.get( importFileId, FileImport.class ) ).thenReturn( fileImport );
+
+ //mock up returning the FailedEntityImport instance after save is invoked.
+
+
+ when( em.create( any( FailedImportConnection.class ) ) ).thenAnswer( new Answer<Object>() {
+ @Override
+ public Object answer( final InvocationOnMock invocation ) throws Throwable {
+ return invocation.getArguments()[0];
+ }
+ } );
+
+ final int expectedSuccess = 100;
+ final int expectedFails = 100;
+ final int expectedConnectionSuccess = 100;
+ final int expectedConnectionFails = 100;
+
+ final int expectedFlushCount = 2;
+ final int flushSize = ( expectedFails + expectedFails + expectedConnectionSuccess + expectedConnectionFails )
+ / expectedFlushCount;
+
+ //set this to 1/2, so that we get saved twice
+ final FileImportTracker fileImportTracker = new FileImportTracker( em, importFileId, flushSize );
+
+
+ for ( long i = 0; i < expectedSuccess; i++ ) {
+ fileImportTracker.entityWritten();
+ }
+
+
+ for ( int i = 0; i < expectedFails; i++ ) {
+ fileImportTracker.entityFailed( "Failed to write entity " + i );
+ }
+
+
+ for ( long i = 0; i < expectedConnectionSuccess; i++ ) {
+ fileImportTracker.connectionWritten();
+ }
+
+
+ for ( int i = 0; i < expectedConnectionFails; i++ ) {
+ fileImportTracker.connectionFailed( "Failed to write connection " + i );
+ }
+
+
+ fileImportTracker.complete();
+
+
+ ArgumentCaptor<FileImport> savedFileImport = ArgumentCaptor.forClass( FileImport.class );
+
+ verify( em, times( expectedFlushCount + 1 ) ).update( savedFileImport.capture() );
+
+ final FileImport updated = savedFileImport.getAllValues().get( 2 );
+
+ assertSame( "Same instance should be updated", fileImport, updated );
+
+ assertEquals( "Same count expected", expectedSuccess, updated.getImportedEntityCount() );
+
+ assertEquals( "Same fail expected", expectedFails, updated.getFailedEntityCount() );
+
+ assertEquals( "Same connection count expected", expectedConnectionSuccess,
+ updated.getImportedConnectionCount() );
+
+ assertEquals( "Same connection error count expected", expectedConnectionFails,
+ updated.getFailedConnectionCount() );
+
+ assertEquals( "Correct error message", "Failed to import some data. See the import counters and errors.",
+ updated.getErrorMessage() );
+
+ //TODO get the connections from the file import
+
+ ArgumentCaptor<FailedImport> failedEntities = ArgumentCaptor.forClass( FailedImport.class );
+
+ verify( em, times( expectedFails + expectedConnectionFails ) )
+ .createConnection( same( fileImport ), eq( "errors" ), failedEntities.capture() );
+
+ //now check all our arguments
+
+ final List<FailedImport> args = failedEntities.getAllValues();
+
+ assertEquals( "Same number of error connections created", expectedFails + expectedConnectionFails,
+ args.size() );
+
+
+ for ( int i = 0; i < expectedFails; i++ ) {
+
+ final FailedImport failedImport = args.get( i );
+
+ assertEquals( "Same message expected", "Failed to write entity " + i, failedImport.getErrorMessage() );
+ }
+
+ for ( int i = expectedFails; i < expectedConnectionFails; i++ ) {
+
+ final FailedImport failedImport = args.get( i );
+
+ assertEquals( "Same message expected", "Failed to write connection " + i, failedImport.getErrorMessage() );
+ }
+ }
+
+
+ @Test
+ public void loadingExistingState() throws Exception {
+
+ final EntityManager em = mock( EntityManager.class );
+
+ final UUID importFileId = UUIDGenerator.newTimeUUID();
+
+
+ final FileImport fileImport = new FileImport();
+ fileImport.setUuid( importFileId );
+ fileImport.setImportedEntityCount( 1 );
+ fileImport.setFailedEntityCount( 2 );
+ fileImport.setImportedConnectionCount( 3 );
+ fileImport.setFailedConnectionCount( 4 );
+
+ when( em.get( importFileId, FileImport.class ) ).thenReturn( fileImport );
+
+ //mock up returning the FailedEntityImport instance after save is invoked.
+
+ FileImportTracker statistics = new FileImportTracker( em, importFileId, 100 );
+
+ assertEquals( 1, statistics.getEntitiesWritten() );
+ assertEquals( 2, statistics.getEntitiesFailed() );
+
+ assertEquals( 3, statistics.getTotalEntityCount() );
+
+ assertEquals( 3, statistics.getConnectionsWritten() );
+ assertEquals( 4, statistics.getConnectionsFailed() );
+
+ assertEquals( 7, statistics.getTotalConnectionCount() );
+ }
+
+
+ @Test
+ public void failFast() throws Exception {
+
+ final EntityManager em = mock( EntityManager.class );
+
+ final UUID importFileId = UUIDGenerator.newTimeUUID();
+
+
+ final FileImport fileImport = new FileImport();
+
+ when( em.get( importFileId, FileImport.class ) ).thenReturn( fileImport );
+
+ //mock up returning the FailedEntityImport instance after save is invoked.
+
+ FileImportTracker statistics = new FileImportTracker( em, importFileId, 100 );
+
+
+ assertFalse( statistics.shouldStopProcessingEntities() );
+
+ assertFalse( statistics.shouldStopProcessingConnections() );
+
+
+ statistics.entityFailed( "test fail" );
+
+ assertTrue("We shouldn't process after a failure", statistics.shouldStopProcessingEntities());
+
+ statistics.connectionFailed( "test fail" );
+
+ assertTrue( "We shouldn't process after a failure", statistics.shouldStopProcessingConnections() );
+ }
+}
+