You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by sf...@apache.org on 2015/02/10 19:05:21 UTC
[3/4] incubator-usergrid git commit: moving dependencies to core
persistence
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/c3869371/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/guice/MaxMigrationVersion.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/guice/MaxMigrationVersion.java b/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/guice/MaxMigrationVersion.java
index be45ab5..8184f09 100644
--- a/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/guice/MaxMigrationVersion.java
+++ b/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/guice/MaxMigrationVersion.java
@@ -21,15 +21,19 @@ package org.apache.usergrid.persistence.core.guice;
import org.apache.usergrid.persistence.core.migration.data.DataMigration;
+import org.apache.usergrid.persistence.core.scope.ApplicationEntityGroup;
+import rx.Observable;
/**
* A simple migration that sets the version to max. This way our integration tests always test the latest code
*/
public class MaxMigrationVersion implements DataMigration {
+
@Override
- public void migrate( final ProgressObserver observer ) throws Throwable {
+ public Observable migrate(final ApplicationEntityGroup applicationEntityGroup, final ProgressObserver observer ) throws Throwable {
//no op, just needs to run to be set
+ return Observable.empty();
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/c3869371/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/migration/data/DataMigrationManagerImplTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/migration/data/DataMigrationManagerImplTest.java b/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/migration/data/DataMigrationManagerImplTest.java
index 3ffcd89..0a80cb7 100644
--- a/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/migration/data/DataMigrationManagerImplTest.java
+++ b/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/migration/data/DataMigrationManagerImplTest.java
@@ -25,11 +25,14 @@ package org.apache.usergrid.persistence.core.migration.data;
import java.util.HashSet;
import java.util.Set;
+import org.apache.usergrid.persistence.core.rx.AllEntitiesInSystemObservable;
+import org.apache.usergrid.persistence.core.scope.ApplicationEntityGroup;
import org.junit.Test;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.apache.usergrid.persistence.core.migration.schema.MigrationException;
+import rx.Observable;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.doAnswer;
@@ -46,16 +49,20 @@ import static org.mockito.Mockito.when;
*/
public class DataMigrationManagerImplTest {
+ AllEntitiesInSystemObservable allEntitiesInSystemObservable = new AllEntitiesInSystemObservable() {
+ @Override
+ public Observable<ApplicationEntityGroup> getAllEntitiesInSystem(int bufferSize) {
+ return null;
+ }
+ };
@Test
public void noMigrations() throws MigrationException {
final MigrationInfoSerialization serialization = mock( MigrationInfoSerialization.class );
-
Set<DataMigration> emptyMigration = new HashSet<>();
-
- DataMigrationManagerImpl migrationManager = new DataMigrationManagerImpl( serialization, emptyMigration );
+ DataMigrationManagerImpl migrationManager = new DataMigrationManagerImpl( serialization, emptyMigration, allEntitiesInSystemObservable );
migrationManager.migrate();
@@ -82,13 +89,13 @@ public class DataMigrationManagerImplTest {
migrations.add( v2 );
- DataMigrationManagerImpl migrationManager = new DataMigrationManagerImpl( serialization, migrations );
+ DataMigrationManagerImpl migrationManager = new DataMigrationManagerImpl( serialization, migrations,allEntitiesInSystemObservable );
migrationManager.migrate();
- verify( v1 ).migrate( any( DataMigration.ProgressObserver.class ) );
- verify( v2 ).migrate( any( DataMigration.ProgressObserver.class ) );
+ verify( v1 ).migrate(any(ApplicationEntityGroup.class), any( DataMigration.ProgressObserver.class ) );
+ verify( v2 ).migrate(any(ApplicationEntityGroup.class), any( DataMigration.ProgressObserver.class ) );
//verify we set the running status
verify( serialization, times( 2 ) ).setStatusCode( DataMigrationManagerImpl.StatusCode.RUNNING.status );
@@ -115,8 +122,8 @@ public class DataMigrationManagerImplTest {
when( v1.getVersion() ).thenReturn( 1 );
//throw an exception
- doThrow( new RuntimeException( "Something bad happened" ) ).when( v1 ).migrate(
- any( DataMigration.ProgressObserver.class ) );
+ doThrow( new RuntimeException( "Something bad happened" ) ).when( v1 ).migrate(any(ApplicationEntityGroup.class),
+ any(DataMigration.ProgressObserver.class) );
final DataMigration v2 = mock( DataMigration.class );
when( v2.getVersion() ).thenReturn( 2 );
@@ -127,15 +134,15 @@ public class DataMigrationManagerImplTest {
migrations.add( v2 );
- DataMigrationManagerImpl migrationManager = new DataMigrationManagerImpl( serialization, migrations );
+ DataMigrationManagerImpl migrationManager = new DataMigrationManagerImpl( serialization, migrations,allEntitiesInSystemObservable );
migrationManager.migrate();
- verify( v1 ).migrate( any( DataMigration.ProgressObserver.class ) );
+ verify( v1 ).migrate( any(ApplicationEntityGroup.class),any( DataMigration.ProgressObserver.class ) );
//verify we don't run migration
- verify( v2, never() ).migrate( any( DataMigration.ProgressObserver.class ) );
+ verify( v2, never() ).migrate( any(ApplicationEntityGroup.class),any( DataMigration.ProgressObserver.class ) );
//verify we set the running status
verify( serialization, times( 1 ) ).setStatusCode( DataMigrationManagerImpl.StatusCode.RUNNING.status );
@@ -177,7 +184,7 @@ public class DataMigrationManagerImplTest {
progressObserver.failed( returnedCode, reason );
return null;
}
- } ).when( v1 ).migrate( any( DataMigration.ProgressObserver.class ) );
+ } ).when( v1 ).migrate(any(ApplicationEntityGroup.class), any( DataMigration.ProgressObserver.class ) );
@@ -190,15 +197,15 @@ public class DataMigrationManagerImplTest {
migrations.add( v2 );
- DataMigrationManagerImpl migrationManager = new DataMigrationManagerImpl( serialization, migrations );
+ DataMigrationManagerImpl migrationManager = new DataMigrationManagerImpl( serialization, migrations,allEntitiesInSystemObservable );
migrationManager.migrate();
- verify( v1 ).migrate( any( DataMigration.ProgressObserver.class ) );
+ verify( v1 ).migrate(any(ApplicationEntityGroup.class), any( DataMigration.ProgressObserver.class ) );
//verify we don't run migration
- verify( v2, never() ).migrate( any( DataMigration.ProgressObserver.class ) );
+ verify( v2, never() ).migrate( any(ApplicationEntityGroup.class),any( DataMigration.ProgressObserver.class ) );
//verify we set the running status
verify( serialization, times( 1 ) ).setStatusCode( DataMigrationManagerImpl.StatusCode.RUNNING.status );
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/c3869371/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/guice/GraphModule.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/guice/GraphModule.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/guice/GraphModule.java
index f50c7cc..4786a42 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/guice/GraphModule.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/guice/GraphModule.java
@@ -21,6 +21,7 @@ package org.apache.usergrid.persistence.graph.guice;
import org.apache.usergrid.persistence.core.guice.V1Impl;
import org.apache.usergrid.persistence.core.guice.V2Impl;
+import org.apache.usergrid.persistence.core.migration.data.DataMigration;
import org.apache.usergrid.persistence.graph.serialization.*;
import org.apache.usergrid.persistence.graph.serialization.impl.*;
import org.safehaus.guicyfig.GuicyFigModule;
@@ -90,6 +91,8 @@ public class GraphModule extends AbstractModule {
bind(EdgesFromSourceObservable.class).to(EdgesFromSourceObservableImpl.class);
+ bind(TargetIdObservable.class).to(TargetIdObservableImpl.class);
+
bind(EdgesToTargetObservable.class).to(EdgesToTargetObservableImpl.class);
/**
@@ -112,6 +115,9 @@ public class GraphModule extends AbstractModule {
bind( EdgeMetaRepair.class ).to( EdgeMetaRepairImpl.class );
bind( EdgeDeleteRepair.class ).to( EdgeDeleteRepairImpl.class );
+ Multibinder<DataMigration> dataMigrationMultibinder =
+ Multibinder.newSetBinder( binder(), DataMigration.class );
+ dataMigrationMultibinder.addBinding().to( EdgeMetadataSerializationProxyImpl.class );
/**
* Add our listeners
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/c3869371/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/EdgeMigrationStrategy.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/EdgeMigrationStrategy.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/EdgeMigrationStrategy.java
index 6333f87..ed46e43 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/EdgeMigrationStrategy.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/EdgeMigrationStrategy.java
@@ -27,7 +27,7 @@ import java.util.List;
/**
* Classy class class.
*/
-public interface EdgeMigrationStrategy extends MigrationStrategy<EdgeMetadataSerialization,Edge> {
+public interface EdgeMigrationStrategy extends MigrationStrategy<EdgeMetadataSerialization> {
public static final int MIGRATION_VERSION = 2;
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/c3869371/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/TargetIdObservable.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/TargetIdObservable.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/TargetIdObservable.java
new file mode 100644
index 0000000..31164df
--- /dev/null
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/TargetIdObservable.java
@@ -0,0 +1,38 @@
+/*
+ *
+ * * Licensed to the Apache Software Foundation (ASF) under one or more
+ * * contributor license agreements. 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. For additional information regarding
+ * * copyright in this work, please see the NOTICE file in the top level
+ * * directory of this distribution.
+ *
+ */
+package org.apache.usergrid.persistence.graph.serialization;
+
+import org.apache.usergrid.persistence.graph.GraphManager;
+import org.apache.usergrid.persistence.model.entity.Id;
+import rx.Observable;
+
+/**
+ * Emits the id of all nodes that are target nodes from the given source node
+ */
+public interface TargetIdObservable {
+ /**
+ * Get all nodes that are target nodes from the sourceNode
+ * @param gm
+ * @param sourceNode
+ *
+ * @return
+ */
+ Observable<Id> getTargetNodes(GraphManager gm, Id sourceNode);
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/c3869371/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/EdgeMetadataSerializationProxyImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/EdgeMetadataSerializationProxyImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/EdgeMetadataSerializationProxyImpl.java
index 60b6a5f..6059569 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/EdgeMetadataSerializationProxyImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/EdgeMetadataSerializationProxyImpl.java
@@ -41,6 +41,7 @@ import org.apache.usergrid.persistence.graph.impl.SimpleSearchByEdgeType;
import org.apache.usergrid.persistence.graph.impl.SimpleSearchEdgeType;
import org.apache.usergrid.persistence.graph.serialization.EdgeMetadataSerialization;
import org.apache.usergrid.persistence.graph.serialization.EdgeMigrationStrategy;
+import org.apache.usergrid.persistence.graph.serialization.EdgesFromSourceObservable;
import org.apache.usergrid.persistence.model.entity.Id;
import com.google.inject.Inject;
@@ -61,6 +62,8 @@ public class EdgeMetadataSerializationProxyImpl implements EdgeMetadataSerializa
private final DataMigrationManager dataMigrationManager;
private final Keyspace keyspace;
+ private final GraphManagerFactory graphManagerFactory;
+ private final EdgesFromSourceObservable edgesFromSourceObservable;
private final EdgeMetadataSerialization previous;
private final EdgeMetadataSerialization current;
@@ -70,10 +73,14 @@ public class EdgeMetadataSerializationProxyImpl implements EdgeMetadataSerializa
*/
@Inject
public EdgeMetadataSerializationProxyImpl(final DataMigrationManager dataMigrationManager, final Keyspace keyspace,
+ final GraphManagerFactory graphManagerFactory,
+ final EdgesFromSourceObservable edgesFromSourceObservable,
@V1Impl final EdgeMetadataSerialization previous,
@V2Impl final EdgeMetadataSerialization current) {
this.dataMigrationManager = dataMigrationManager;
this.keyspace = keyspace;
+ this.graphManagerFactory = graphManagerFactory;
+ this.edgesFromSourceObservable = edgesFromSourceObservable;
this.previous = previous;
this.current = current;
}
@@ -291,14 +298,16 @@ public class EdgeMetadataSerializationProxyImpl implements EdgeMetadataSerializa
}
@Override
- public int getMigrationVersion() {
+ public int getVersion() {
return EdgeMigrationStrategy.MIGRATION_VERSION;
}
@Override
- public Observable<Long> executeMigration(final Observable<Edge> edgesFromSource ,final ApplicationEntityGroup applicationEntityGroup,
- final DataMigration.ProgressObserver observer,
- Func1<Id, ? extends ApplicationScope> getScopeFromEntityId) {
+ public Observable<Long> migrate(final ApplicationEntityGroup applicationEntityGroup,
+ final DataMigration.ProgressObserver observer) {
+ final GraphManager gm = graphManagerFactory.createEdgeManager(applicationEntityGroup.applicationScope);
+ final Observable<Edge> edgesFromSource = edgesFromSourceObservable.edgesFromSource(gm, applicationEntityGroup.applicationScope.getApplication());
+
final AtomicLong counter = new AtomicLong();
rx.Observable o =
Observable
@@ -344,7 +353,7 @@ public class EdgeMetadataSerializationProxyImpl implements EdgeMetadataSerializa
final long newCount =
counter.addAndGet(edges.size());
- observer.update(getMigrationVersion(), String.format(
+ observer.update(getVersion(), String.format(
"Currently running. Rewritten %d edge types",
newCount));
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/c3869371/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/TargetIdObservableImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/TargetIdObservableImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/TargetIdObservableImpl.java
new file mode 100644
index 0000000..bf0a03e
--- /dev/null
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/TargetIdObservableImpl.java
@@ -0,0 +1,70 @@
+/*
+ *
+ * * Licensed to the Apache Software Foundation (ASF) under one or more
+ * * contributor license agreements. 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. For additional information regarding
+ * * copyright in this work, please see the NOTICE file in the top level
+ * * directory of this distribution.
+ *
+ */
+package org.apache.usergrid.persistence.graph.serialization.impl;
+
+
+import org.apache.usergrid.persistence.graph.Edge;
+import org.apache.usergrid.persistence.graph.GraphManager;
+import org.apache.usergrid.persistence.graph.serialization.EdgesFromSourceObservable;
+import org.apache.usergrid.persistence.graph.serialization.TargetIdObservable;
+import org.apache.usergrid.persistence.model.entity.Id;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import rx.Observable;
+import rx.functions.Func1;
+
+/**
+ * Emits the id of all nodes that are target nodes from the given source node
+ */
+public class TargetIdObservableImpl implements TargetIdObservable {
+
+ private static final Logger logger = LoggerFactory.getLogger(TargetIdObservableImpl.class);
+ private final EdgesFromSourceObservable edgesFromSourceObservable;
+
+ public TargetIdObservableImpl(final EdgesFromSourceObservable edgesFromSourceObservable){
+ this.edgesFromSourceObservable = edgesFromSourceObservable;
+ }
+
+ /**
+ * Get all nodes that are target nodes from the sourceNode
+ * @param gm
+ * @param sourceNode
+ *
+ * @return
+ */
+ @Override
+ public Observable<Id> getTargetNodes(final GraphManager gm, final Id sourceNode) {
+
+ //only search edge types that start with collections
+ return edgesFromSourceObservable.edgesFromSource(gm, sourceNode ).map( new Func1<Edge, Id>() {
+
+
+ @Override
+ public Id call( final Edge edge ) {
+ final Id targetNode = edge.getTargetNode();
+
+ logger.debug( "Emitting targetId of {}", edge );
+
+
+ return targetNode;
+ }
+ } );
+ }
+}