usergrid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mdun...@apache.org
Subject [33/44] usergrid git commit: Removes obsolete plugins.
Date Sat, 26 Sep 2015 01:42:38 GMT
Removes obsolete plugins.

Adds tests and implements entity-data plugin.


Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/2689a015
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/2689a015
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/2689a015

Branch: refs/heads/USERGRID-933
Commit: 2689a015938e6b05ceb6f6cc2c228f61cb5e3b17
Parents: acbf3a1
Author: Todd Nine <tnine@apigee.com>
Authored: Thu Sep 17 16:17:43 2015 -0600
Committer: Todd Nine <tnine@apigee.com>
Committed: Thu Sep 17 16:17:43 2015 -0600

----------------------------------------------------------------------
 .../usergrid/corepersistence/CoreModule.java    |  53 +++++--
 .../migration/CoreMigration.java                |  45 ++----
 .../migration/CoreMigrationPlugin.java          |  31 ++--
 .../migration/DeDupConnectionDataMigration.java |  93 ++++++++++++
 .../migration/EntityTypeMappingMigration.java   | 104 --------------
 .../corepersistence/migration/Versions.java     |  31 ----
 .../DeDupConnectionDataMigrationTest.java       | 144 +++++++++++++++++++
 .../migration/EntityTypeMappingMigrationIT.java | 125 ----------------
 .../migration/data/TestProgressObserver.java    |  10 +-
 9 files changed, 314 insertions(+), 322 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/usergrid/blob/2689a015/stack/core/src/main/java/org/apache/usergrid/corepersistence/CoreModule.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CoreModule.java
b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CoreModule.java
index 7ed95e2..11707be 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CoreModule.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CoreModule.java
@@ -16,22 +16,43 @@
 package org.apache.usergrid.corepersistence;
 
 
-import com.google.inject.AbstractModule;
-import com.google.inject.TypeLiteral;
-import com.google.inject.assistedinject.FactoryModuleBuilder;
-import com.google.inject.multibindings.Multibinder;
+import org.safehaus.guicyfig.GuicyFigModule;
+
 import org.apache.usergrid.corepersistence.asyncevents.AsyncEventService;
 import org.apache.usergrid.corepersistence.asyncevents.AsyncIndexProvider;
 import org.apache.usergrid.corepersistence.asyncevents.EventBuilder;
 import org.apache.usergrid.corepersistence.asyncevents.EventBuilderImpl;
-import org.apache.usergrid.corepersistence.index.*;
+import org.apache.usergrid.corepersistence.index.ApplicationIndexBucketLocator;
+import org.apache.usergrid.corepersistence.index.CoreIndexFig;
+import org.apache.usergrid.corepersistence.index.IndexLocationStrategyFactory;
+import org.apache.usergrid.corepersistence.index.IndexLocationStrategyFactoryImpl;
+import org.apache.usergrid.corepersistence.index.IndexProcessorFig;
+import org.apache.usergrid.corepersistence.index.IndexService;
+import org.apache.usergrid.corepersistence.index.IndexServiceImpl;
+import org.apache.usergrid.corepersistence.index.ReIndexService;
+import org.apache.usergrid.corepersistence.index.ReIndexServiceImpl;
 import org.apache.usergrid.corepersistence.migration.CoreMigration;
 import org.apache.usergrid.corepersistence.migration.CoreMigrationPlugin;
-import org.apache.usergrid.corepersistence.migration.EntityTypeMappingMigration;
+import org.apache.usergrid.corepersistence.migration.DeDupConnectionDataMigration;
 import org.apache.usergrid.corepersistence.migration.MigrationModuleVersionPlugin;
 import org.apache.usergrid.corepersistence.pipeline.PipelineModule;
-import org.apache.usergrid.corepersistence.rx.impl.*;
-import org.apache.usergrid.corepersistence.service.*;
+import org.apache.usergrid.corepersistence.rx.impl.AllApplicationsObservable;
+import org.apache.usergrid.corepersistence.rx.impl.AllApplicationsObservableImpl;
+import org.apache.usergrid.corepersistence.rx.impl.AllEntitiesInSystemImpl;
+import org.apache.usergrid.corepersistence.rx.impl.AllEntityIdsObservable;
+import org.apache.usergrid.corepersistence.rx.impl.AllEntityIdsObservableImpl;
+import org.apache.usergrid.corepersistence.rx.impl.AllNodesInGraphImpl;
+import org.apache.usergrid.corepersistence.service.AggregationService;
+import org.apache.usergrid.corepersistence.service.AggregationServiceFactory;
+import org.apache.usergrid.corepersistence.service.AggregationServiceImpl;
+import org.apache.usergrid.corepersistence.service.ApplicationService;
+import org.apache.usergrid.corepersistence.service.ApplicationServiceImpl;
+import org.apache.usergrid.corepersistence.service.CollectionService;
+import org.apache.usergrid.corepersistence.service.CollectionServiceImpl;
+import org.apache.usergrid.corepersistence.service.ConnectionService;
+import org.apache.usergrid.corepersistence.service.ConnectionServiceImpl;
+import org.apache.usergrid.corepersistence.service.StatusService;
+import org.apache.usergrid.corepersistence.service.StatusServiceImpl;
 import org.apache.usergrid.persistence.collection.guice.CollectionModule;
 import org.apache.usergrid.persistence.collection.serialization.impl.migration.EntityIdScope;
 import org.apache.usergrid.persistence.core.guice.CommonModule;
@@ -42,7 +63,11 @@ import org.apache.usergrid.persistence.core.scope.ApplicationScope;
 import org.apache.usergrid.persistence.graph.guice.GraphModule;
 import org.apache.usergrid.persistence.graph.serialization.impl.migration.GraphNode;
 import org.apache.usergrid.persistence.index.guice.IndexModule;
-import org.safehaus.guicyfig.GuicyFigModule;
+
+import com.google.inject.AbstractModule;
+import com.google.inject.TypeLiteral;
+import com.google.inject.assistedinject.FactoryModuleBuilder;
+import com.google.inject.multibindings.Multibinder;
 
 
 /**
@@ -105,7 +130,7 @@ public class CoreModule  extends AbstractModule {
                         new TypeLiteral<DataMigration>() {}, CoreMigration.class );
 
 
-        dataMigrationMultibinder.addBinding().to( EntityTypeMappingMigration.class );
+        dataMigrationMultibinder.addBinding().to( DeDupConnectionDataMigration.class );
 
 
         //wire up the collection migration plugin
@@ -164,6 +189,14 @@ public class CoreModule  extends AbstractModule {
         bind( ApplicationService.class ).to( ApplicationServiceImpl.class );
 
         bind( StatusService.class ).to( StatusServiceImpl.class );
+
+        /**
+         * Install migration services
+         */
+
+          //migrations
+        //we want to make sure our generics are retained, so we use a typeliteral
+
     }
 
 }

http://git-wip-us.apache.org/repos/asf/usergrid/blob/2689a015/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/CoreMigration.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/CoreMigration.java
b/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/CoreMigration.java
index 4df32da..b491e1f 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/CoreMigration.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/CoreMigration.java
@@ -1,40 +1,21 @@
 /*
- * 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
+ * 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
+ *      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.
+ * 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.corepersistence.migration;/*
- * 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.corepersistence.migration;
 
 
 import java.lang.annotation.Retention;

http://git-wip-us.apache.org/repos/asf/usergrid/blob/2689a015/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/CoreMigrationPlugin.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/CoreMigrationPlugin.java
b/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/CoreMigrationPlugin.java
index 810f9fb..b69451a 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/CoreMigrationPlugin.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/CoreMigrationPlugin.java
@@ -1,20 +1,18 @@
 /*
- * 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
+ * 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
+ *      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.
+ * 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.corepersistence.migration;
@@ -22,9 +20,10 @@ package org.apache.usergrid.corepersistence.migration;
 
 import java.util.Set;
 
-import org.apache.usergrid.persistence.core.migration.data.MigrationInfoSerialization;
+import org.apache.usergrid.persistence.collection.serialization.impl.migration.CollectionMigration;
 import org.apache.usergrid.persistence.core.migration.data.AbstractMigrationPlugin;
 import org.apache.usergrid.persistence.core.migration.data.DataMigration;
+import org.apache.usergrid.persistence.core.migration.data.MigrationInfoSerialization;
 import org.apache.usergrid.persistence.core.migration.data.PluginPhase;
 
 import com.google.inject.Inject;
@@ -37,7 +36,7 @@ import com.google.inject.Singleton;
 @Singleton
 public class CoreMigrationPlugin extends AbstractMigrationPlugin {
 
-    public static final String PLUGIN_NAME = "core-data";
+    public static final String PLUGIN_NAME = "collections-entity-data";
 
 
 

http://git-wip-us.apache.org/repos/asf/usergrid/blob/2689a015/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/DeDupConnectionDataMigration.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/DeDupConnectionDataMigration.java
b/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/DeDupConnectionDataMigration.java
new file mode 100644
index 0000000..f65864a
--- /dev/null
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/DeDupConnectionDataMigration.java
@@ -0,0 +1,93 @@
+/*
+ * 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.corepersistence.migration;
+
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.usergrid.corepersistence.rx.impl.AllApplicationsObservable;
+import org.apache.usergrid.corepersistence.service.ConnectionService;
+import org.apache.usergrid.persistence.core.migration.data.DataMigration;
+import org.apache.usergrid.persistence.core.migration.data.ProgressObserver;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+
+
+@Singleton
+public class DeDupConnectionDataMigration implements DataMigration {
+
+    private static final Logger logger = LoggerFactory.getLogger(DeDupConnectionDataMigration.class);
+
+    private static final long UPDATE_COUNT = 1000;
+
+    private final ConnectionService connectionService;
+    private final AllApplicationsObservable allApplicationsObservable;
+
+
+    @Inject
+    public DeDupConnectionDataMigration( final ConnectionService connectionService,
+                                         final AllApplicationsObservable allApplicationsObservable
) {
+        this.connectionService = connectionService;
+        this.allApplicationsObservable = allApplicationsObservable;
+    }
+
+
+    @Override
+    public int migrate( final int currentVersion, final ProgressObserver observer ) {
+
+        final int migrationVersion = getMaxVersion();
+
+        observer.start();
+
+        connectionService.deDupeConnections( allApplicationsObservable.getData() ).reduce(
0l, (count, deDuped) ->{
+
+            final long newCount = count+1;
+
+            /**
+             * Update our progress observer
+             */
+            if(newCount % UPDATE_COUNT == 0){
+                logger.info( "De duped {} edges", newCount );
+                observer.update( migrationVersion, String.format("De duped %d edges", newCount)
);
+            }
+
+            return newCount;
+
+        }).doOnNext( total -> {
+            logger.info("Completed de-duping {} edges", total );
+            observer.complete();;
+        }).subscribe();
+
+        return migrationVersion;
+
+    }
+
+
+    @Override
+    public boolean supports( final int currentVersion ) {
+        return currentVersion == getMaxVersion() - 1;
+    }
+
+
+    @Override
+    public int getMaxVersion() {
+        return 1;
+    }
+}

http://git-wip-us.apache.org/repos/asf/usergrid/blob/2689a015/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/EntityTypeMappingMigration.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/EntityTypeMappingMigration.java
b/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/EntityTypeMappingMigration.java
deleted file mode 100644
index ef65ff2..0000000
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/EntityTypeMappingMigration.java
+++ /dev/null
@@ -1,104 +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.corepersistence.migration;
-
-
-import java.util.UUID;
-import java.util.concurrent.atomic.AtomicLong;
-
-import org.apache.usergrid.corepersistence.ManagerCache;
-import org.apache.usergrid.corepersistence.util.CpNamingUtils;
-import org.apache.usergrid.persistence.collection.serialization.impl.migration.EntityIdScope;
-import org.apache.usergrid.persistence.core.migration.data.DataMigration;
-import org.apache.usergrid.persistence.core.migration.data.MigrationDataProvider;
-import org.apache.usergrid.persistence.core.migration.data.ProgressObserver;
-import org.apache.usergrid.persistence.map.MapManager;
-import org.apache.usergrid.persistence.map.MapScope;
-
-import com.google.inject.Inject;
-
-import rx.Observable;
-import rx.schedulers.Schedulers;
-
-
-/**
- * Migration to ensure that our entity id is written into our map data
- */
-public class EntityTypeMappingMigration implements DataMigration {
-
-    private final ManagerCache managerCache;
-    private final MigrationDataProvider<EntityIdScope> allEntitiesInSystemObservable;
-
-
-    @Inject
-    public EntityTypeMappingMigration( final ManagerCache managerCache,
-                                       final MigrationDataProvider<EntityIdScope> allEntitiesInSystemObservable
) {
-        this.managerCache = managerCache;
-        this.allEntitiesInSystemObservable = allEntitiesInSystemObservable;
-    }
-
-
-    @Override
-    public int migrate( final int currentVersion,
-                        final ProgressObserver observer ) {
-
-        final AtomicLong atomicLong = new AtomicLong();
-
-
-        //migrate up to 100 types simultaneously
-        allEntitiesInSystemObservable.getData().flatMap( entityIdScope -> {
-            return Observable.just( entityIdScope ).doOnNext( entityIdScopeObservable ->
{
-                final MapScope ms = CpNamingUtils
-                                                 .getEntityTypeMapScope( entityIdScope.getApplicationScope().getApplication()
);
-
-                                             final MapManager mapManager = managerCache.getMapManager(
ms );
-
-                                             final UUID entityUuid = entityIdScope.getId().getUuid();
-                                             final String entityType = entityIdScope.getId().getType();
-
-                                             mapManager.putString( entityUuid.toString(),
entityType );
-
-                                             if ( atomicLong.incrementAndGet() % 100 == 0
) {
-                                                 observer.update( getMaxVersion(),
-                                                     String.format( "Updated %d entities",
atomicLong.get() ) );
-                                             }
-
-            } ).subscribeOn( Schedulers.io() );
-        }, 100 ).count().toBlocking().last();
-
-
-        return getMaxVersion();
-
-
-    }
-
-
-    @Override
-    public boolean supports( final int currentVersion ) {
-        //we move from the migration version fix to the current version
-        return CoreDataVersions.INITIAL.getVersion() == currentVersion;
-    }
-
-
-    @Override
-    public int getMaxVersion() {
-        return CoreDataVersions.ID_MAP_FIX.getVersion();
-    }
-}

http://git-wip-us.apache.org/repos/asf/usergrid/blob/2689a015/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/Versions.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/Versions.java
b/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/Versions.java
deleted file mode 100644
index d921353..0000000
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/Versions.java
+++ /dev/null
@@ -1,31 +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.corepersistence.migration;
-
-
-/**
- * Simple class to hold the constants of all versions
- */
-public class Versions {
-
-
-}

http://git-wip-us.apache.org/repos/asf/usergrid/blob/2689a015/stack/core/src/test/java/org/apache/usergrid/corepersistence/migration/DeDupConnectionDataMigrationTest.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/corepersistence/migration/DeDupConnectionDataMigrationTest.java
b/stack/core/src/test/java/org/apache/usergrid/corepersistence/migration/DeDupConnectionDataMigrationTest.java
new file mode 100644
index 0000000..c84ad2b
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/corepersistence/migration/DeDupConnectionDataMigrationTest.java
@@ -0,0 +1,144 @@
+/*
+ * 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.corepersistence.migration;
+
+
+import java.util.List;
+
+import org.junit.Test;
+
+import org.apache.usergrid.corepersistence.rx.impl.AllApplicationsObservable;
+import org.apache.usergrid.corepersistence.service.ConnectionScope;
+import org.apache.usergrid.corepersistence.service.ConnectionService;
+import org.apache.usergrid.corepersistence.util.CpNamingUtils;
+import org.apache.usergrid.persistence.core.migration.data.TestProgressObserver;
+import org.apache.usergrid.persistence.core.scope.ApplicationScope;
+import org.apache.usergrid.persistence.core.scope.ApplicationScopeImpl;
+import org.apache.usergrid.persistence.graph.Edge;
+
+import rx.Observable;
+import rx.Subscriber;
+
+import static org.apache.usergrid.persistence.core.util.IdGenerator.createId;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+
+public class DeDupConnectionDataMigrationTest {
+
+    @Test
+    public void testVersion() {
+
+        //mock up an initial system state
+        final int version = 0;
+
+        final ConnectionService connectionService = mock( ConnectionService.class );
+
+        final AllApplicationsObservable allApplicationsObservable = mock( AllApplicationsObservable.class
);
+
+
+        final DeDupConnectionDataMigration plugin =
+            new DeDupConnectionDataMigration( connectionService, allApplicationsObservable
);
+
+
+        assertTrue( plugin.supports( version ) );
+
+        assertFalse( plugin.supports( plugin.getMaxVersion() ) );
+    }
+
+
+    /**
+     * Test
+     */
+    @Test
+    public void testEmitEdges() {
+
+        //mock up an initial system state
+        final int version = 0;
+
+        final ConnectionService connectionService = mock( ConnectionService.class );
+
+
+        final AllApplicationsObservable allApplicationsObservable = mock( AllApplicationsObservable.class
);
+
+
+        final int count = 3000;
+
+        final Observable<ConnectionScope> edgeEmitter = Observable.create( new DupConnectionEmitter(
count ) );
+
+        when( connectionService.deDupeConnections( any( Observable.class ) ) ).thenReturn(
edgeEmitter );
+
+
+        final DeDupConnectionDataMigration plugin =
+            new DeDupConnectionDataMigration( connectionService, allApplicationsObservable
);
+
+
+        final TestProgressObserver testProgressObserver = new TestProgressObserver();
+
+
+        plugin.migrate( version, testProgressObserver );
+
+        final List<String> updates = testProgressObserver.getUpdates();
+
+        assertEquals( "Expected 3 updates", 3, updates.size() );
+
+
+        assertEquals( "De duped 1000 edges", updates.get( 0 ) );
+        assertEquals( "De duped 2000 edges", updates.get( 1 ) );
+        assertEquals( "De duped 3000 edges", updates.get( 2 ) );
+
+
+        assertTrue( "Should complete", testProgressObserver.isComplete() );
+        assertTrue( "Should start", testProgressObserver.isStarted() );
+        assertFalse( "Should not fail", testProgressObserver.isFailed() );
+    }
+
+
+    private final class DupConnectionEmitter implements Observable.OnSubscribe<ConnectionScope>
{
+
+
+        private final int count;
+
+
+        private DupConnectionEmitter( final int count ) {this.count = count;}
+
+
+        @Override
+        public void call( final Subscriber<? super ConnectionScope> subscriber ) {
+
+            final ApplicationScope applicationScope = new ApplicationScopeImpl( createId(
"application" ) );
+            final Edge edge = CpNamingUtils.createConnectionEdge( createId( "source" ), "test",
createId( "target" ) );
+
+
+            final ConnectionScope scope = new ConnectionScope( applicationScope, edge );
+
+
+            subscriber.onStart();
+
+            for ( int i = 0; i < count; i++ ) {
+                subscriber.onNext( scope );
+            }
+
+            subscriber.onCompleted();
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/usergrid/blob/2689a015/stack/core/src/test/java/org/apache/usergrid/corepersistence/migration/EntityTypeMappingMigrationIT.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/corepersistence/migration/EntityTypeMappingMigrationIT.java
b/stack/core/src/test/java/org/apache/usergrid/corepersistence/migration/EntityTypeMappingMigrationIT.java
deleted file mode 100644
index 7d22abe..0000000
--- a/stack/core/src/test/java/org/apache/usergrid/corepersistence/migration/EntityTypeMappingMigrationIT.java
+++ /dev/null
@@ -1,125 +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.corepersistence.migration;
-
-
-import org.junit.Test;
-
-import org.apache.usergrid.AbstractCoreIT;
-import org.apache.usergrid.corepersistence.ManagerCache;
-import org.apache.usergrid.corepersistence.util.CpNamingUtils;
-import org.apache.usergrid.persistence.collection.serialization.impl.migration.EntityIdScope;
-import org.apache.usergrid.persistence.core.migration.data.TestMigrationDataProvider;
-import org.apache.usergrid.persistence.core.migration.data.TestProgressObserver;
-import org.apache.usergrid.persistence.core.scope.ApplicationScope;
-import org.apache.usergrid.persistence.core.scope.ApplicationScopeImpl;
-import org.apache.usergrid.persistence.map.MapManager;
-import org.apache.usergrid.persistence.map.MapScope;
-import org.apache.usergrid.persistence.map.impl.MapScopeImpl;
-import org.apache.usergrid.persistence.model.entity.Id;
-
-import rx.Observable;
-
-import static org.apache.usergrid.persistence.core.util.IdGenerator.createId;
-import static org.junit.Assert.assertEquals;
-import static org.mockito.Matchers.eq;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
-
-/**
- * Test for our entity type mapping
- */
-public class EntityTypeMappingMigrationIT  {
-
-
-    @Test
-    public void testIdMapping() throws Throwable {
-
-        final Id applicationId = createId("application");
-
-        final ApplicationScope scope1 = new ApplicationScopeImpl( applicationId );
-
-        final Id entityId1 = createId("thing");
-
-        final EntityIdScope idScope1 = new EntityIdScope(scope1, entityId1 );
-
-        final MapScope mapScope1 = new MapScopeImpl(applicationId, CpNamingUtils.TYPES_BY_UUID_MAP
);
-
-
-
-        final ApplicationScope scope2 = new ApplicationScopeImpl( applicationId);
-
-        final Id entityId2 = createId("foo");
-
-        final EntityIdScope idScope2 = new EntityIdScope( scope2, entityId2 );
-
-        final MapScope mapScope2 = new MapScopeImpl(applicationId, CpNamingUtils.TYPES_BY_UUID_MAP
);
-
-
-        final Observable<EntityIdScope> scopes = Observable.just(idScope1, idScope2);
-
-        final TestMigrationDataProvider<EntityIdScope> migrationDataProvider = new
TestMigrationDataProvider<>();
-
-        //set our scopes
-        migrationDataProvider.setObservable( scopes );
-
-
-
-
-        //mock up returning our map manager
-        final MapManager mapManager = mock(MapManager.class);
-        final ManagerCache managerCache = mock(ManagerCache.class);
-
-        when(managerCache.getMapManager( eq( mapScope1 ) )).thenReturn( mapManager );
-
-        when(managerCache.getMapManager( eq( mapScope2 ) )).thenReturn( mapManager );
-
-        final TestProgressObserver progressObserver = new TestProgressObserver();
-
-
-        //wire it up
-        final EntityTypeMappingMigration migration = new EntityTypeMappingMigration( managerCache,
migrationDataProvider );
-
-        //run it
-
-        final int returnedVersion = migration.migrate(CoreDataVersions.INITIAL.getVersion(),
progressObserver );
-
-
-        assertEquals(CoreDataVersions.ID_MAP_FIX.getVersion(), returnedVersion);
-
-        //verify we saved it
-
-        verify(mapManager).putString(entityId1.getUuid().toString(), entityId1.getType()
 );
-
-        verify(mapManager).putString(entityId2.getUuid().toString(), entityId2.getType()
 );
-
-
-
-
-
-
-
-
-
-
-    }
-}

http://git-wip-us.apache.org/repos/asf/usergrid/blob/2689a015/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/migration/data/TestProgressObserver.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/migration/data/TestProgressObserver.java
b/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/migration/data/TestProgressObserver.java
index 7f59151..21d6013 100644
--- a/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/migration/data/TestProgressObserver.java
+++ b/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/migration/data/TestProgressObserver.java
@@ -30,19 +30,21 @@ public class TestProgressObserver implements ProgressObserver {
 
     private boolean started = false;
 
+    private boolean complete = false;
+
 
     private List<String> updates = new ArrayList<>( 100 );
 
 
     @Override
     public void start() {
-         started = true;
+        started = true;
     }
 
 
     @Override
     public void complete() {
-        started = false;
+        complete = true;
     }
 
 
@@ -66,7 +68,6 @@ public class TestProgressObserver implements ProgressObserver {
 
     /**
      * Get if we failed
-     * @return
      */
     public boolean isFailed() {
         return failed;
@@ -78,10 +79,11 @@ public class TestProgressObserver implements ProgressObserver {
     }
 
 
+    public boolean isComplete() { return complete;}
+
 
     /**
      * Get update messages
-     * @return
      */
     public List<String> getUpdates() {
         return updates;


Mime
View raw message