You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by mr...@apache.org on 2016/07/11 15:35:29 UTC
[19/50] [abbrv] usergrid git commit: Unique values test passes with
new "actor system" module
http://git-wip-us.apache.org/repos/asf/usergrid/blob/ee18950f/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerImpl.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerImpl.java
index 3cc4a07..dea0b0c 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerImpl.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerImpl.java
@@ -27,6 +27,7 @@ import java.util.List;
import java.util.UUID;
import com.netflix.astyanax.model.ConsistencyLevel;
+import org.apache.usergrid.persistence.actorsystem.ActorSystemManager;
import org.apache.usergrid.persistence.collection.uniquevalues.UniqueValuesService;
import org.apache.usergrid.persistence.collection.serialization.impl.LogEntryIterator;
import org.slf4j.Logger;
@@ -125,6 +126,7 @@ public class EntityCollectionManagerImpl implements EntityCollectionManager {
private final RxTaskScheduler rxTaskScheduler;
private final UniqueValuesService uniqueValuesService;
+ private final ActorSystemManager actorSystemManager;
@Inject
@@ -144,6 +146,7 @@ public class EntityCollectionManagerImpl implements EntityCollectionManager {
final MetricsFactory metricsFactory,
final SerializationFig serializationFig,
final RxTaskScheduler rxTaskScheduler,
+ ActorSystemManager actorSystemManager,
UniqueValuesService uniqueValuesService,
@Assisted final ApplicationScope applicationScope ) {
@@ -154,6 +157,7 @@ public class EntityCollectionManagerImpl implements EntityCollectionManager {
this.serializationFig = serializationFig;
this.rxTaskScheduler = rxTaskScheduler;
+ this.actorSystemManager = actorSystemManager;
this.uniqueValuesService = uniqueValuesService;
ValidationUtils.validateApplicationScope( applicationScope );
@@ -255,7 +259,8 @@ public class EntityCollectionManagerImpl implements EntityCollectionManager {
Preconditions.checkNotNull( entityIds, "entityIds cannot be null" );
- final Observable<EntitySet> entitySetObservable = Observable.create( new Observable.OnSubscribe<EntitySet>() {
+ final Observable<EntitySet> entitySetObservable =
+ Observable.create( new Observable.OnSubscribe<EntitySet>() {
@Override
public void call( final Subscriber<? super EntitySet> subscriber ) {
@@ -461,7 +466,8 @@ public class EntityCollectionManagerImpl implements EntityCollectionManager {
public Observable<VersionSet> getLatestVersion( final Collection<Id> entityIds ) {
- final Observable<VersionSet> observable = Observable.create( new Observable.OnSubscribe<VersionSet>() {
+ final Observable<VersionSet> observable =
+ Observable.create( new Observable.OnSubscribe<VersionSet>() {
@Override
public void call( final Subscriber<? super VersionSet> subscriber ) {
@@ -506,17 +512,4 @@ public class EntityCollectionManagerImpl implements EntityCollectionManager {
return Health.RED;
}
-
-
- @Override
- public void startAkkaForTesting( String hostname, int port, String region ) {
- try {
- uniqueValuesService.start( hostname, port, region );
- uniqueValuesService.waitForRequestActors();
-
- } catch (Throwable t) {
- logger.error("Error starting Akka", t);
- throw t;
- }
- }
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/ee18950f/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/WriteCommit.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/WriteCommit.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/WriteCommit.java
index 44028ae..267e793 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/WriteCommit.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/WriteCommit.java
@@ -22,8 +22,8 @@ import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
+import org.apache.usergrid.persistence.actorsystem.ActorSystemFig;
import org.apache.usergrid.persistence.collection.exception.WriteUniqueVerifyException;
-import org.apache.usergrid.persistence.collection.uniquevalues.AkkaFig;
import org.apache.usergrid.persistence.collection.uniquevalues.UniqueValueException;
import org.apache.usergrid.persistence.collection.uniquevalues.UniqueValuesService;
import org.slf4j.Logger;
@@ -66,7 +66,7 @@ public class WriteCommit implements Func1<CollectionIoEvent<MvccEntity>, Collect
private static final Logger logger = LoggerFactory.getLogger( WriteCommit.class );
- AkkaFig akkaFig;
+ ActorSystemFig actorSystemFig;
UniqueValuesService akkaUvService;
@Inject
@@ -81,7 +81,7 @@ public class WriteCommit implements Func1<CollectionIoEvent<MvccEntity>, Collect
public WriteCommit( final MvccLogEntrySerializationStrategy logStrat,
final MvccEntitySerializationStrategy entryStrat,
final UniqueValueSerializationStrategy uniqueValueStrat,
- final AkkaFig akkaFig,
+ final ActorSystemFig actorSystemFig,
final UniqueValuesService akkaUvService ) {
Preconditions.checkNotNull( logStrat, "MvccLogEntrySerializationStrategy is required" );
@@ -91,7 +91,7 @@ public class WriteCommit implements Func1<CollectionIoEvent<MvccEntity>, Collect
this.logEntryStrat = logStrat;
this.entityStrat = entryStrat;
this.uniqueValueStrat = uniqueValueStrat;
- this.akkaFig = akkaFig;
+ this.actorSystemFig = actorSystemFig;
this.akkaUvService = akkaUvService;
}
@@ -130,10 +130,10 @@ public class WriteCommit implements Func1<CollectionIoEvent<MvccEntity>, Collect
logMutation.mergeShallow( entityMutation );
// akkaFig may be null when this is called from JUnit tests
- if ( akkaFig != null && akkaFig.getAkkaEnabled() ) {
+ if ( actorSystemFig != null && actorSystemFig.getAkkaEnabled() ) {
String region = ioEvent.getRegion();
if ( region == null ) {
- region = akkaFig.getAkkaAuthoritativeRegion();
+ region = actorSystemFig.getAkkaAuthoritativeRegion();
}
confirmUniqueFieldsAkka( mvccEntity, version, applicationScope, region );
} else {
http://git-wip-us.apache.org/repos/asf/usergrid/blob/ee18950f/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/WriteUniqueVerify.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/WriteUniqueVerify.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/WriteUniqueVerify.java
index da394f7..848ed2e 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/WriteUniqueVerify.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/WriteUniqueVerify.java
@@ -29,6 +29,7 @@ import com.netflix.hystrix.HystrixCommand;
import com.netflix.hystrix.HystrixCommandGroupKey;
import com.netflix.hystrix.HystrixCommandProperties;
import com.netflix.hystrix.HystrixThreadPoolProperties;
+import org.apache.usergrid.persistence.actorsystem.ActorSystemFig;
import org.apache.usergrid.persistence.collection.MvccEntity;
import org.apache.usergrid.persistence.collection.exception.WriteUniqueVerifyException;
import org.apache.usergrid.persistence.collection.mvcc.entity.MvccValidationUtils;
@@ -38,7 +39,6 @@ import org.apache.usergrid.persistence.collection.serialization.UniqueValue;
import org.apache.usergrid.persistence.collection.serialization.UniqueValueSerializationStrategy;
import org.apache.usergrid.persistence.collection.serialization.UniqueValueSet;
import org.apache.usergrid.persistence.collection.serialization.impl.UniqueValueImpl;
-import org.apache.usergrid.persistence.collection.uniquevalues.AkkaFig;
import org.apache.usergrid.persistence.collection.uniquevalues.UniqueValueException;
import org.apache.usergrid.persistence.collection.uniquevalues.UniqueValuesService;
import org.apache.usergrid.persistence.core.astyanax.CassandraConfig;
@@ -65,7 +65,7 @@ public class WriteUniqueVerify implements Action1<CollectionIoEvent<MvccEntity>>
private static final Logger logger = LoggerFactory.getLogger( WriteUniqueVerify.class );
- AkkaFig akkaFig;
+ ActorSystemFig actorSystemFig;
UniqueValuesService akkaUvService;
private final UniqueValueSerializationStrategy uniqueValueStrat;
@@ -85,12 +85,12 @@ public class WriteUniqueVerify implements Action1<CollectionIoEvent<MvccEntity>>
final SerializationFig serializationFig,
final Keyspace keyspace,
final CassandraConfig cassandraFig,
- final AkkaFig akkaFig,
+ final ActorSystemFig actorSystemFig,
final UniqueValuesService akkaUvService ) {
this.keyspace = keyspace;
this.cassandraFig = cassandraFig;
- this.akkaFig = akkaFig;
+ this.actorSystemFig = actorSystemFig;
this.akkaUvService = akkaUvService;
Preconditions.checkNotNull( uniqueValueSerializiationStrategy, "uniqueValueSerializationStrategy is required" );
@@ -105,7 +105,7 @@ public class WriteUniqueVerify implements Action1<CollectionIoEvent<MvccEntity>>
@Override
public void call( final CollectionIoEvent<MvccEntity> ioevent ) {
- if ( akkaFig != null && akkaFig.getAkkaEnabled() ) {
+ if ( actorSystemFig != null && actorSystemFig.getAkkaEnabled() ) {
verifyUniqueFieldsAkka( ioevent );
} else {
verifyUniqueFields( ioevent );
@@ -124,7 +124,7 @@ public class WriteUniqueVerify implements Action1<CollectionIoEvent<MvccEntity>>
String region = ioevent.getRegion();
if ( region == null ) {
- region = akkaFig.getAkkaAuthoritativeRegion();
+ region = actorSystemFig.getAkkaAuthoritativeRegion();
}
try {
akkaUvService.reserveUniqueValues( applicationScope, entity, mvccEntity.getVersion(), region );
http://git-wip-us.apache.org/repos/asf/usergrid/blob/ee18950f/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/AkkaFig.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/AkkaFig.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/AkkaFig.java
deleted file mode 100644
index 0f97403..0000000
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/AkkaFig.java
+++ /dev/null
@@ -1,116 +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.persistence.collection.uniquevalues;
-
-
-import org.safehaus.guicyfig.Default;
-import org.safehaus.guicyfig.FigSingleton;
-import org.safehaus.guicyfig.GuicyFig;
-import org.safehaus.guicyfig.Key;
-
-import java.io.Serializable;
-
-@FigSingleton
-public interface AkkaFig extends GuicyFig, Serializable {
-
- String AKKA_ENABLED = "collection.akka.enabled";
-
- String AKKA_HOSTNAME = "collection.akka.hostname";
-
- String AKKA_PORT = "collection.akka.port";
-
- String AKKA_REGION = "collection.akka.region";
-
- String AKKA_REGION_LIST = "usergrid.queue.regionList"; // same region list used by queues
-
- String AKKA_REGION_SEEDS = "collection.akka.region.seeds";
-
- String AKKA_UNIQUEVALUE_ACTORS = "collection.akka.uniquevalue.actors";
-
- String AKKA_UNIQUEVALUE_CACHE_TTL = "collection.akka.uniquevalue.cache.ttl";
-
- String AKKA_UNIQUEVALUE_RESERVATION_TTL= "collection.akka.uniquevalue.reservation.ttl";
-
- String AKKA_AUTHORITATIVE_REGION = "collection.akka.uniquevalue.authoritative.region";
-
- /**
- * Use Akka or nah
- */
- @Key(AKKA_ENABLED)
- @Default("true")
- boolean getAkkaEnabled();
-
- /**
- * Hostname to be used in Akka configuration.
- */
- @Key(AKKA_HOSTNAME)
- String getHostname();
-
- /**
- * local port to be used in Akka configuration.
- */
- @Key(AKKA_PORT)
- int getPort();
-
- /**
- * Local region to be used in Akka configuration.
- */
- @Key(AKKA_REGION)
- String getRegion();
-
- /**
- * Comma separated list of regions known to cluster.
- */
- @Key(AKKA_REGION_LIST)
- String getRegionList();
-
- /**
- * Number of UniqueValueActors to be started on each node
- */
- @Key(AKKA_UNIQUEVALUE_ACTORS)
- @Default("300")
- int getUniqueValueActors();
-
- /**
- * Comma-separated lists of seeds each with format {region}:{hostname}:{port}.
- * Regions MUST be listed in the 'usergrid.queue.regionList'
- */
- @Key(AKKA_REGION_SEEDS)
- String getRegionSeeds();
-
- /**
- * If no region specified for type, use the authoritative region
- */
- @Key(AKKA_AUTHORITATIVE_REGION)
- String getAkkaAuthoritativeRegion();
-
- /**
- * Unique Value cache TTL in seconds.
- */
- @Key(AKKA_UNIQUEVALUE_CACHE_TTL)
- @Default("10")
- int getUniqueValueCacheTtl();
-
- /**
- * Unique Value Reservation TTL in seconds.
- */
- @Key(AKKA_UNIQUEVALUE_RESERVATION_TTL)
- @Default("10")
- int getUniqueValueReservationTtl();
-}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/ee18950f/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/ClusterSingletonRouter.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/ClusterSingletonRouter.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/ClusterSingletonRouter.java
deleted file mode 100644
index f2f80bf..0000000
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/ClusterSingletonRouter.java
+++ /dev/null
@@ -1,70 +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.persistence.collection.uniquevalues;
-
-import akka.actor.ActorRef;
-import akka.actor.Props;
-import akka.actor.UntypedActor;
-import akka.routing.ConsistentHashingRouter;
-import akka.routing.FromConfig;
-import com.google.inject.Inject;
-import com.google.inject.Injector;
-import org.apache.commons.lang.RandomStringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-
-/**
- * Uses a consistent hash to route Unique Value requests to UniqueValueActors.
- */
-public class ClusterSingletonRouter extends UntypedActor {
- private static final Logger logger = LoggerFactory.getLogger( UniqueValueActor.class );
-
- private final String name = RandomStringUtils.randomAlphanumeric( 4 );
-
- private final ActorRef router;
-
- @Inject
- public ClusterSingletonRouter( Injector injector ) {
-
- router = getContext().actorOf(
- FromConfig.getInstance().props(Props.create(UniqueValueActor.class)), "router");
-
- // TODO: is there some way to pass the injector here without getting this exception:
- // NotSerializableException: No configured serialization-bindings for class [InjectorImpl]
- //router = getContext().actorOf(
- //FromConfig.getInstance().props( Props.create( GuiceActorProducer.class, injector, UniqueValueActor.class)),
- //"router" );
-
- logger.info("ClusterSingletonRouter {} is live with injector {}", name, injector);
- }
-
- @Override
- public void onReceive(Object message) {
-
- if ( message instanceof UniqueValueActor.Request) {
- UniqueValueActor.Request request = (UniqueValueActor.Request)message;
-
- ConsistentHashingRouter.ConsistentHashableEnvelope envelope =
- new ConsistentHashingRouter.ConsistentHashableEnvelope( message, request.getConsistentHashKey() );
- router.tell( envelope, getSender());
-
- } else {
- unhandled(message);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/ee18950f/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/GuiceActorProducer.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/GuiceActorProducer.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/GuiceActorProducer.java
deleted file mode 100644
index 0e99bca..0000000
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/GuiceActorProducer.java
+++ /dev/null
@@ -1,46 +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.persistence.collection.uniquevalues;
-
-import akka.actor.Actor;
-import akka.actor.IndirectActorProducer;
-import com.google.inject.Injector;
-
-
-public class GuiceActorProducer implements IndirectActorProducer {
-
- final Injector injector;
- final Class<? extends Actor> actorClass;
-
- public GuiceActorProducer(Injector injector, Class<? extends Actor> actorClass) {
- this.injector = injector;
- this.actorClass = actorClass;
- }
-
- @Override
- public Class<? extends Actor> actorClass() {
- return actorClass;
- }
-
- @Override
- public Actor produce() {
- return injector.getInstance( actorClass );
- }
-}
-
http://git-wip-us.apache.org/repos/asf/usergrid/blob/ee18950f/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/RequestActor.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/RequestActor.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/RequestActor.java
deleted file mode 100644
index c451c61..0000000
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/RequestActor.java
+++ /dev/null
@@ -1,186 +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.persistence.collection.uniquevalues;
-
-import akka.actor.ActorSelection;
-import akka.actor.Address;
-import akka.actor.UntypedActor;
-import akka.cluster.Cluster;
-import akka.cluster.ClusterEvent;
-import akka.cluster.Member;
-import akka.cluster.MemberStatus;
-import org.apache.commons.lang3.RandomStringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.Serializable;
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-import java.util.concurrent.ThreadLocalRandom;
-
-
-/**
- * Once notified of nodes, sends unique propertyValue requests to ClusterSingletonRouter via it's local proxy.
- */
-class RequestActor extends UntypedActor {
- private static final Logger logger = LoggerFactory.getLogger( RequestActor.class );
-
- private final String name = RandomStringUtils.randomAlphanumeric( 4 );
-
- private final Set<Address> nodes = new HashSet<>();
-
- private final Cluster cluster = Cluster.get(getContext().system());
- private final String routerProxyPath;
-
- private boolean ready = false;
-
-
- public RequestActor(String routerProxyPath ) {
- this.routerProxyPath = routerProxyPath;
- }
-
- // subscribe to cluster changes, MemberEvent
- @Override
- public void preStart() {
- logger.debug("{} role {} address {}:{} starting up, subscribing to cluster events...", name,
- cluster.getSelfRoles().iterator().next(),
- cluster.readView().selfAddress().host(),
- cluster.readView().selfAddress().hostPort());
- cluster.subscribe(getSelf(), ClusterEvent.MemberEvent.class, ClusterEvent.ReachabilityEvent.class);
- }
-
- // re-subscribe when restart
- @Override
- public void postStop() {
- cluster.unsubscribe(getSelf());
- }
-
- @Override
- public void onReceive(Object message) {
-
- int startSize = nodes.size();
-
- if ( message instanceof UniqueValueActor.Request && ready ) {
-
- // just pick any node, the ClusterSingletonRouter will do the consistent hash routing
- List<Address> nodesList = new ArrayList<>( nodes );
- Address address = nodesList.get( ThreadLocalRandom.current().nextInt( nodesList.size() ) );
- ActorSelection service = getContext().actorSelection( address + routerProxyPath );
- service.tell( message, getSender() );
-
- } else if ( message instanceof UniqueValueActor.Request && !ready ) {
- logger.debug("{} responding with status unknown", name);
-
- getSender().tell( new UniqueValueActor.Response(
- UniqueValueActor.Response.Status.ERROR ) , getSender() );
-
- } else if ( message instanceof StatusRequest ) {
- if ( ready ) {
- getSender().tell( new StatusMessage( name, StatusMessage.Status.READY ), getSender() );
- } else {
- getSender().tell( new StatusMessage( name, StatusMessage.Status.INITIALIZING), getSender() );
- }
- return;
-
- } else {
- processAsClusterEvent( message );
- }
-
- if ( logger.isDebugEnabled() && startSize != nodes.size() ) {
- logger.debug( "{} now knows {} nodes", name, nodes.size() );
- }
-
- if (!nodes.isEmpty() && !ready) {
- logger.debug( name + " is ready" );
- ready = true;
-
- } else if (nodes.isEmpty() && ready) {
- ready = false;
- }
- }
-
- /**
- * Process messages about nodes up, down, reachable and unreachable.
- */
- private void processAsClusterEvent(Object message) {
-
- if (message instanceof ClusterEvent.CurrentClusterState) {
- ClusterEvent.CurrentClusterState state = (ClusterEvent.CurrentClusterState) message;
- nodes.clear();
- for (Member member : state.getMembers()) {
- if (member.hasRole("io") && member.status().equals( MemberStatus.up())) {
- nodes.add(member.address());
- logger.debug("RequestActor {} received cluster-state member-up for {}", name, member.address());
- }
- }
-
- } else if (message instanceof ClusterEvent.MemberUp) {
- ClusterEvent.MemberUp mUp = (ClusterEvent.MemberUp) message;
- if (mUp.member().hasRole("io")) {
- nodes.add( mUp.member().address() );
- }
- logger.debug("{} received member-up for {}", name, mUp.member().address());
-
- } else if (message instanceof ClusterEvent.MemberEvent) {
- ClusterEvent.MemberEvent other = (ClusterEvent.MemberEvent) message;
- nodes.remove(other.member().address());
-
- } else if (message instanceof ClusterEvent.UnreachableMember) {
- ClusterEvent.UnreachableMember unreachable = (ClusterEvent.UnreachableMember) message;
- nodes.remove(unreachable.member().address());
- logger.debug("{} received un-reachable for {}", name, unreachable.member().address());
-
- } else if (message instanceof ClusterEvent.ReachableMember) {
- ClusterEvent.ReachableMember reachable = (ClusterEvent.ReachableMember) message;
- if (reachable.member().hasRole("io")) {
- nodes.add( reachable.member().address() );
- }
- logger.debug("{} received reachable for {}", name, reachable.member().address());
-
- } else {
- logger.error("{}: unhandled message: {}", name, message.toString());
- unhandled(message);
- }
- }
-
- /**
- * RequestAction responds to StatusRequests.
- */
- static class StatusRequest implements Serializable { }
-
- /**
- * RequestActor responds with, and some times unilaterally sends StatusMessages.
- */
- static class StatusMessage implements Serializable {
- final String name;
- public enum Status { INITIALIZING, READY }
- final Status status;
- public StatusMessage(String name, Status status) {
- this.name = name;
- this.status = status;
- }
- public String getName() {
- return name;
- }
- public boolean isReady() {
- return status.equals( Status.READY );
- }
- }
-}
-
http://git-wip-us.apache.org/repos/asf/usergrid/blob/ee18950f/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/ReservationCache.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/ReservationCache.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/ReservationCache.java
index 24b7f6e..f1e68b2 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/ReservationCache.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/ReservationCache.java
@@ -19,6 +19,7 @@ package org.apache.usergrid.persistence.collection.uniquevalues;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheStats;
+import org.apache.usergrid.persistence.actorsystem.ClientActor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -26,7 +27,7 @@ import java.util.concurrent.TimeUnit;
public class ReservationCache {
- private static final Logger logger = LoggerFactory.getLogger( RequestActor.class );
+ private static final Logger logger = LoggerFactory.getLogger( ClientActor.class );
Cache<String, UniqueValueActor.Reservation> cache;
long ttl;
http://git-wip-us.apache.org/repos/asf/usergrid/blob/ee18950f/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValueActor.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValueActor.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValueActor.java
index 1e7879a..bb30b92 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValueActor.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValueActor.java
@@ -48,7 +48,7 @@ public class UniqueValueActor extends UntypedActor {
// TODO: is there a way to avoid this ugly kludge? see also: ClusterSingletonRouter
this.table = UniqueValuesServiceImpl.injector.getInstance( UniqueValuesTable.class );
- logger.info("UniqueValueActor {} is live with table {}", name, table);
+ //logger.info("UniqueValueActor {} is live with table {}", name, table);
}
@Override
http://git-wip-us.apache.org/repos/asf/usergrid/blob/ee18950f/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValuesRouter.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValuesRouter.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValuesRouter.java
new file mode 100644
index 0000000..376af66
--- /dev/null
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValuesRouter.java
@@ -0,0 +1,70 @@
+/*
+ * 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.persistence.collection.uniquevalues;
+
+import akka.actor.ActorRef;
+import akka.actor.Props;
+import akka.actor.UntypedActor;
+import akka.routing.ConsistentHashingRouter;
+import akka.routing.FromConfig;
+import com.google.inject.Inject;
+import com.google.inject.Injector;
+import org.apache.commons.lang.RandomStringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Uses a consistent hash to route Unique Value requests to UniqueValueActors.
+ */
+public class UniqueValuesRouter extends UntypedActor {
+ private static final Logger logger = LoggerFactory.getLogger( UniqueValueActor.class );
+
+ private final String name = RandomStringUtils.randomAlphanumeric( 4 );
+
+ private final ActorRef router;
+
+ @Inject
+ public UniqueValuesRouter(Injector injector ) {
+
+ router = getContext().actorOf(
+ FromConfig.getInstance().props(Props.create(UniqueValueActor.class)), "router");
+
+ // TODO: is there some way to pass the injector here without getting this exception:
+ // NotSerializableException: No configured serialization-bindings for class [InjectorImpl]
+ //router = getContext().actorOf(
+ //FromConfig.getInstance().props( Props.create( GuiceActorProducer.class, injector, UniqueValueActor.class)),
+ //"router" );
+
+ //logger.info("UniqueValuesRouter {} is live with injector {}", name, injector);
+ }
+
+ @Override
+ public void onReceive(Object message) {
+
+ if ( message instanceof UniqueValueActor.Request) {
+ UniqueValueActor.Request request = (UniqueValueActor.Request)message;
+
+ ConsistentHashingRouter.ConsistentHashableEnvelope envelope =
+ new ConsistentHashingRouter.ConsistentHashableEnvelope( message, request.getConsistentHashKey() );
+ router.tell( envelope, getSender());
+
+ } else {
+ unhandled(message);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/ee18950f/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValuesService.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValuesService.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValuesService.java
index 744c5b9..e9a9f50 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValuesService.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValuesService.java
@@ -19,20 +19,17 @@
package org.apache.usergrid.persistence.collection.uniquevalues;
+import org.apache.usergrid.persistence.actorsystem.RouterProducer;
import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.model.entity.Entity;
import java.util.UUID;
+
/**
* Service that reserves and confirms unique values.
*/
-public interface UniqueValuesService {
-
- /**
- * Initialize and start service.
- */
- void start();
+public interface UniqueValuesService extends RouterProducer {
/**
* Check that unique values are unique and reserve them for a limited time.
@@ -58,15 +55,4 @@ public interface UniqueValuesService {
*/
void confirmUniqueValues( ApplicationScope scope, Entity entity, UUID version , String region )
throws UniqueValueException;
-
-
- /**
- * For test purposes only.
- */
- void start( String hostname, Integer port, String region );
-
- /**
- * For test purposes only.
- */
- void waitForRequestActors();
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/ee18950f/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValuesServiceImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValuesServiceImpl.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValuesServiceImpl.java
index 670fffd..85b9d1a 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValuesServiceImpl.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValuesServiceImpl.java
@@ -28,15 +28,12 @@ import akka.cluster.singleton.ClusterSingletonProxy;
import akka.cluster.singleton.ClusterSingletonProxySettings;
import akka.pattern.Patterns;
import akka.util.Timeout;
-import com.google.common.collect.ArrayListMultimap;
-import com.google.common.collect.ListMultimap;
import com.google.inject.Inject;
import com.google.inject.Injector;
import com.google.inject.Singleton;
-import com.typesafe.config.Config;
-import com.typesafe.config.ConfigFactory;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.usergrid.persistence.collection.exception.CollectionRuntimeException;
+import org.apache.usergrid.persistence.actorsystem.ActorSystemFig;
+import org.apache.usergrid.persistence.actorsystem.ActorSystemManager;
+import org.apache.usergrid.persistence.actorsystem.GuiceActorProducer;
import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.model.entity.Entity;
import org.apache.usergrid.persistence.model.field.Field;
@@ -45,7 +42,8 @@ import org.slf4j.LoggerFactory;
import scala.concurrent.Await;
import scala.concurrent.Future;
-import java.util.*;
+import java.util.Map;
+import java.util.UUID;
import java.util.concurrent.TimeUnit;
@@ -53,157 +51,30 @@ import java.util.concurrent.TimeUnit;
public class UniqueValuesServiceImpl implements UniqueValuesService {
private static final Logger logger = LoggerFactory.getLogger( UniqueValuesServiceImpl.class );
- static Injector injector;
-
- AkkaFig akkaFig;
- UniqueValuesTable table;
- private String hostname;
- private Integer port = null;
- private String currentRegion;
-
- private Map<String, ActorRef> requestActorsByRegion;
-
- //private Map<String, String> regionsByType = new HashMap<>();
-
-// private final MetricRegistry metrics = new MetricRegistry();
-//
-// private final Timer getTimer = metrics.timer( "get" );
-// private final Timer saveTimer = metrics.timer( "save" );
-//
-// private final Counter cacheCounter = metrics.counter( "cache" );
-// private final Counter dupCounter = metrics.counter( "duplicates" );
-//
-// private final Timer reservationTimer = metrics.timer( "reservation" );
-// private final Timer commitmentTimer = metrics.timer( "commitment" );
-
+ static Injector injector;
+ ActorSystemFig actorSystemFig;
+ ActorSystemManager actorSystemManager;
+ UniqueValuesTable table;
private ReservationCache reservationCache;
@Inject
- public UniqueValuesServiceImpl(Injector inj, AkkaFig akkaFig, UniqueValuesTable table ) {
+ public UniqueValuesServiceImpl(
+ Injector inj,
+ ActorSystemFig actorSystemFig,
+ ActorSystemManager actorSystemManager,
+ UniqueValuesTable table ) {
+
injector = inj;
- this.akkaFig = akkaFig;
+ this.actorSystemManager = actorSystemManager;
+ this.actorSystemFig = actorSystemFig;
this.table = table;
- ReservationCache.init( akkaFig.getUniqueValueCacheTtl() );
+ ReservationCache.init( actorSystemFig.getUniqueValueCacheTtl() );
this.reservationCache = ReservationCache.getInstance();
}
- /**
- * Init Akka ActorSystems and wait for request actors to start.
- */
- public void start() {
-
- this.hostname = akkaFig.getHostname();
- this.currentRegion = akkaFig.getRegion();
- this.port = null;
-
- initAkka();
- waitForRequestActors();
- }
-
-
- /**
- * For testing purposes only; does not wait for request actors to start.
- */
- public void start( String hostname, Integer port, String currentRegion ) {
-
- this.hostname = hostname;
- this.currentRegion = currentRegion;
- this.port = port;
-
- initAkka();
- }
-
-
- private Map<String, ActorRef> getRequestActorsByRegion() {
- return requestActorsByRegion;
- }
-
-
-// private Map<String, String> getRegionsByType() {
-// return regionsByType;
-// }
-
-// public Counter getDupCounter() {
-// return dupCounter;
-// }
-//
-// public Counter getCacheCounter() {
-// return cacheCounter;
-// }
-//
-// public Timer getReservationTimer() {
-// return reservationTimer;
-// }
-//
-// public Timer getCommitmentTimer() {
-// return commitmentTimer;
-// }
-//
-// public Timer getSaveTimer() {
-// return saveTimer;
-// }
-//
-// public Timer getGetTimer() {
-// return getTimer;
-// }
-
- private void initAkka() {
- logger.info("Initializing Akka");
-
- // Create one actor system with request actor for each region
-
- if ( StringUtils.isEmpty( hostname )) {
- throw new RuntimeException( "No value specified for " + AkkaFig.AKKA_HOSTNAME );
- }
-
- if ( StringUtils.isEmpty( currentRegion )) {
- throw new RuntimeException( "No value specified for " + AkkaFig.AKKA_REGION );
- }
-
- if ( StringUtils.isEmpty( akkaFig.getRegionList() )) {
- throw new RuntimeException( "No value specified for " + AkkaFig.AKKA_REGION_LIST );
- }
-
- if ( StringUtils.isEmpty( akkaFig.getRegionSeeds() )) {
- throw new RuntimeException( "No value specified for " + AkkaFig.AKKA_REGION_SEEDS);
- }
-
- if ( StringUtils.isEmpty( akkaFig.getAkkaAuthoritativeRegion() )) {
- throw new RuntimeException( "No value specified for " + AkkaFig.AKKA_AUTHORITATIVE_REGION);
- }
-
- List regionList = Arrays.asList( akkaFig.getRegionList().toLowerCase().split(",") );
-
- logger.info("Initializing Akka for hostname {} region {} regionList {} seeds {}",
- hostname, currentRegion, regionList, akkaFig.getRegionSeeds() );
-
-// String typesValue = akkaFig.getRegionTypes();
-// String[] regionTypes = StringUtils.isEmpty( typesValue ) ? new String[0] : typesValue.split(",");
-// for ( String regionType : regionTypes ) {
-// String[] parts = regionType.toLowerCase().split(":");
-// String typeRegion = parts[0];
-// String type = parts[1];
-//
-// if ( !regionList.contains( typeRegion) ) {
-// throw new RuntimeException(
-// "'collection.akka.region.seeds' references unknown region: " + typeRegion );
-// }
-// this.regionsByType.put( type, typeRegion );
-// }
-
- final Map<String, ActorSystem> systemMap = new HashMap<>();
-
- ActorSystem localSystem = createClusterSingletonProxies( readClusterSingletonConfigs(), systemMap );
-
- createRequestActors( systemMap );
-
- subscribeToReservations( localSystem, systemMap );
- }
-
-
private void subscribeToReservations( ActorSystem localSystem, Map<String, ActorSystem> systemMap ) {
for ( String region : systemMap.keySet() ) {
@@ -216,246 +87,11 @@ public class UniqueValuesServiceImpl implements UniqueValuesService {
}
- /**
- * Create ActorSystem and ClusterSingletonProxy for every region.
- * Create ClusterSingletonManager for the current region.
- *
- * @param configMap Configurations to be used to create ActorSystems
- * @param systemMap Map of ActorSystems created by this method
- *
- * @return ActorSystem for this region.
- */
- private ActorSystem createClusterSingletonProxies(
- Map<String, Config> configMap, Map<String, ActorSystem> systemMap ) {
-
- ActorSystem localSystem = null;
-
- for ( String region : configMap.keySet() ) {
- Config config = configMap.get( region );
-
- ActorSystem system = ActorSystem.create( "ClusterSystem", config );
- systemMap.put( region, system );
-
- // cluster singletons only run role "io" nodes and NOT on "client" nodes of other regions
- if ( currentRegion.equals( region ) ) {
-
- localSystem = system;
-
- // create cluster singleton supervisor for actor system
- ClusterSingletonManagerSettings settings =
- ClusterSingletonManagerSettings.create( system ).withRole("io");
-
- // Akka.system().actorOf(Props.create(GuiceInjectedActor.class, INJECTOR,Retreiver.class))
-
- system.actorOf( ClusterSingletonManager.props(
- //Props.create( ClusterSingletonRouter.class, table ),
- Props.create( GuiceActorProducer.class, injector, ClusterSingletonRouter.class),
- PoisonPill.getInstance(), settings ), "uvRouter");
- }
-
- // create proxy for sending messages to singleton
- ClusterSingletonProxySettings proxySettings =
- ClusterSingletonProxySettings.create( system ).withRole("io");
- system.actorOf( ClusterSingletonProxy.props( "/user/uvRouter", proxySettings ), "uvProxy" );
- }
-
- return localSystem;
- }
-
-
- /**
- * Create RequestActor for each region.
- *
- * @param systemMap Map of regions to ActorSystems.
- */
- private void createRequestActors( Map<String, ActorSystem> systemMap ) {
-
- requestActorsByRegion = new HashMap<>();
-
- for ( String region : systemMap.keySet() ) {
-
- logger.info("Creating request actor for region {}", region);
-
- // Each RequestActor needs to know path to ClusterSingletonProxy and region
- ActorRef requestActor = systemMap.get( region ).actorOf(
- Props.create( RequestActor.class, "/user/uvProxy" ), "requestActor" );
-
- requestActorsByRegion.put( region, requestActor );
- }
- }
-
-
- public void waitForRequestActors() {
-
- for ( String region : requestActorsByRegion.keySet() ) {
- ActorRef ra = requestActorsByRegion.get( region );
- waitForRequestActor( ra );
- }
- }
-
-
- private void waitForRequestActor( ActorRef ra ) {
-
- logger.info( "Waiting on request actor {}...", ra.path() );
-
- boolean started = false;
- int retries = 0;
- int maxRetries = 60;
- while (retries < maxRetries) {
- Timeout t = new Timeout( 10, TimeUnit.SECONDS );
-
- Future<Object> fut = Patterns.ask( ra, new RequestActor.StatusRequest(), t );
- try {
- RequestActor.StatusMessage result = (RequestActor.StatusMessage) Await.result( fut, t.duration() );
-
- if (result.status.equals( RequestActor.StatusMessage.Status.READY )) {
- started = true;
- break;
- }
- logger.info( "Waiting for request actor {} region {} ({}s)", ra.path(), currentRegion, retries );
- Thread.sleep( 1000 );
-
- } catch (Exception e) {
- logger.error( "Error: Timeout waiting for requestActor" );
- }
- retries++;
- }
-
- if (started) {
- logger.info( "RequestActor has started" );
- } else {
- throw new RuntimeException( "RequestActor did not start in time" );
- }
- }
-
-
- /**
- * Read configuration and create a Config for each region.
- *
- * @return Map of regions to Configs.
- */
- private Map<String, Config> readClusterSingletonConfigs() {
-
- Map<String, Config> configs = new HashMap<>();
-
- ListMultimap<String, String> seedsByRegion = ArrayListMultimap.create();
-
- String[] regionSeeds = akkaFig.getRegionSeeds().split( "," );
-
- logger.info("Found region {} seeds {}", regionSeeds.length, regionSeeds);
-
- try {
-
- if ( port != null ) {
-
- // we are testing
- String seed = "akka.tcp://ClusterSystem@" + hostname + ":" + port;
- seedsByRegion.put( currentRegion, seed );
- logger.info("Akka testing, only starting one seed");
-
- } else {
-
- for (String regionSeed : regionSeeds) {
-
- String[] parts = regionSeed.split( ":" );
- String region = parts[0];
- String hostname = parts[1];
- String regionPortString = parts[2];
-
- // all seeds in same region must use same port
- // we assume 0th seed has the right port
- final Integer regionPort;
-
- if (port == null) {
- regionPort = Integer.parseInt( regionPortString );
- } else {
- regionPort = port; // unless we are testing
- }
-
- String seed = "akka.tcp://ClusterSystem@" + hostname + ":" + regionPort;
-
- logger.info("Adding seed {} for region {}", seed, region );
-
- seedsByRegion.put( region, seed );
- }
-
- if (seedsByRegion.keySet().isEmpty()) {
- throw new RuntimeException(
- "No seeds listed in 'parsing collection.akka.region.seeds' property." );
- }
- }
-
- int numInstancesPerNode = akkaFig.getUniqueValueActors();
-
- for ( String region : seedsByRegion.keySet() ) {
-
- List<String> seeds = seedsByRegion.get( region );
- int lastColon = seeds.get(0).lastIndexOf(":") + 1;
- final Integer regionPort = Integer.parseInt( seeds.get(0).substring( lastColon ));
-
- // cluster singletons only run role "io" nodes and NOT on "client" nodes of other regions
- String clusterRole = currentRegion.equals( region ) ? "io" : "client";
-
- logger.info( "Config for region {} is:\n" +
- " AkkaUV Hostname {}\n" +
- " AkkaUV Seeds {}\n" +
- " AkkaUV Port {}\n" +
- " AkkaUV UniqueValueActors per node {}\n" +
- " AkkaUV Authoritative Region {}",
- region, hostname, seeds, port, numInstancesPerNode, akkaFig.getAkkaAuthoritativeRegion() );
-
- Map<String, Object> configMap = new HashMap<String, Object>() {{
- put( "akka", new HashMap<String, Object>() {{
- put( "remote", new HashMap<String, Object>() {{
- put( "netty.tcp", new HashMap<String, Object>() {{
- put( "hostname", hostname );
- put( "bind-hostname", hostname );
- put( "port", regionPort );
- }} );
- }} );
- put( "cluster", new HashMap<String, Object>() {{
- put( "max-nr-of-instances-per-node", numInstancesPerNode );
- put( "roles", Collections.singletonList(clusterRole) );
- put( "seed-nodes", new ArrayList<String>() {{
- for (String seed : seeds) {
- add( seed );
- }
- }} );
- }} );
- put( "actor", new HashMap<String, Object>() {{
- put( "deployment", new HashMap<String, Object>() {{
- put( "/uvRouter/singleton/router", new HashMap<String, Object>() {{
- put( "cluster", new HashMap<String, Object>() {{
- //put( "roles", Collections.singletonList(role) );
- put( "max-nr-of-instances-per-node", numInstancesPerNode );
- }} );
- }} );
- }} );
- }} );
- }} );
- }};
-
- Config config = ConfigFactory
- .parseMap( configMap )
- .withFallback( ConfigFactory.parseString( "akka.cluster.roles = [io]" ) )
- .withFallback( ConfigFactory.load( "cluster-singleton" ) );
-
- configs.put( region, config );
- }
-
- } catch ( Exception e ) {
- throw new RuntimeException("Error 'parsing collection.akka.region.seeds' property", e );
- }
-
- return configs;
- }
-
-
@Override
public void reserveUniqueValues(
ApplicationScope scope, Entity entity, UUID version, String region ) throws UniqueValueException {
- if ( this.getRequestActorsByRegion().isEmpty() ) {
+ if ( !actorSystemManager.isReady() ) {
throw new RuntimeException("Unique values service not initialized, no request actors ready");
}
@@ -487,7 +123,7 @@ public class UniqueValuesServiceImpl implements UniqueValuesService {
public void confirmUniqueValues(
ApplicationScope scope, Entity entity, UUID version, String region ) throws UniqueValueException {
- if ( this.getRequestActorsByRegion().isEmpty() ) {
+ if ( !actorSystemManager.isReady() ) {
throw new RuntimeException("Unique values service not initialized, no request actors ready");
}
@@ -518,7 +154,7 @@ public class UniqueValuesServiceImpl implements UniqueValuesService {
private void reserveUniqueField(
ApplicationScope scope, Entity entity, UUID version, Field field, String region ) throws UniqueValueException {
- final ActorRef requestActor = getRequestActorsByRegion().get( region );
+ final ActorRef requestActor = actorSystemManager.getClientActor( region );
if ( requestActor == null ) {
throw new RuntimeException( "No request actor for region " + region);
@@ -542,7 +178,7 @@ public class UniqueValuesServiceImpl implements UniqueValuesService {
private void confirmUniqueField(
ApplicationScope scope, Entity entity, UUID version, Field field, String region) throws UniqueValueException {
- final ActorRef requestActor = getRequestActorsByRegion().get( region );
+ final ActorRef requestActor = actorSystemManager.getClientActor( region );
if ( requestActor == null ) {
throw new RuntimeException( "No request actor for type, cannot verify unique fields!" );
@@ -558,7 +194,7 @@ public class UniqueValuesServiceImpl implements UniqueValuesService {
private void cancelUniqueField(
ApplicationScope scope, Entity entity, UUID version, Field field, String region ) throws UniqueValueException {
- final ActorRef requestActor = getRequestActorsByRegion().get( region );
+ final ActorRef requestActor = actorSystemManager.getClientActor( region );
if ( requestActor == null ) {
throw new RuntimeException( "No request actor for type, cannot verify unique fields!" );
@@ -616,7 +252,7 @@ public class UniqueValuesServiceImpl implements UniqueValuesService {
} catch ( Exception e ) {
logger.debug("{} caused retry {} for entity {} rowkey {}",
- e.getClass().getSimpleName(), retries, entity.getId().getUuid(), request.getConsistentHashKey());
+ e.getClass().getSimpleName(), retries, entity.getId().getUuid(), request.getConsistentHashKey());
}
}
@@ -634,4 +270,34 @@ public class UniqueValuesServiceImpl implements UniqueValuesService {
throw new UniqueValueException( "Error property not unique", request.getField() );
}
}
+
+
+ @Override
+ public void createClusterSingletonManager(ActorSystem system) {
+
+ // create cluster singleton supervisor for actor system
+ ClusterSingletonManagerSettings settings =
+ ClusterSingletonManagerSettings.create( system ).withRole("io");
+
+ system.actorOf( ClusterSingletonManager.props(
+ //Props.create( ClusterSingletonRouter.class, table ),
+ Props.create( GuiceActorProducer.class, injector, UniqueValuesRouter.class),
+ PoisonPill.getInstance(), settings ), "uvRouter");
+ }
+
+
+ @Override
+ public void createClusterSingletonProxy(ActorSystem system) {
+
+ ClusterSingletonProxySettings proxySettings =
+ ClusterSingletonProxySettings.create( system ).withRole("io");
+
+ system.actorOf( ClusterSingletonProxy.props( "/user/uvRouter", proxySettings ), "uvProxy" );
+ }
+
+
+ @Override
+ public void createLocalSystemActors( ActorSystem localSystem, Map<String, ActorSystem> systemMap ) {
+ subscribeToReservations( localSystem, systemMap );
+ }
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/ee18950f/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValuesTableImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValuesTableImpl.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValuesTableImpl.java
index c0fa390..de326dd 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValuesTableImpl.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValuesTableImpl.java
@@ -22,6 +22,7 @@ import com.google.inject.Inject;
import com.google.inject.Singleton;
import com.netflix.astyanax.MutationBatch;
import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
+import org.apache.usergrid.persistence.actorsystem.ActorSystemFig;
import org.apache.usergrid.persistence.collection.serialization.UniqueValue;
import org.apache.usergrid.persistence.collection.serialization.UniqueValueSerializationStrategy;
import org.apache.usergrid.persistence.collection.serialization.UniqueValueSet;
@@ -41,12 +42,12 @@ public class UniqueValuesTableImpl implements UniqueValuesTable {
private static final Logger logger = LoggerFactory.getLogger( UniqueValuesTableImpl.class );
final UniqueValueSerializationStrategy strat;
- final AkkaFig akkaFig;
+ final ActorSystemFig actorSystemFig;
@Inject
- public UniqueValuesTableImpl( final UniqueValueSerializationStrategy strat, AkkaFig akkaFig ) {
+ public UniqueValuesTableImpl( final UniqueValueSerializationStrategy strat, ActorSystemFig actorSystemFig) {
this.strat = strat;
- this.akkaFig = akkaFig;
+ this.actorSystemFig = actorSystemFig;
}
@@ -62,7 +63,7 @@ public class UniqueValuesTableImpl implements UniqueValuesTable {
public void reserve( ApplicationScope scope, Id owner, UUID version, Field field ) throws ConnectionException {
UniqueValue uv = new UniqueValueImpl( field, owner, version);
- final MutationBatch write = strat.write( scope, uv, akkaFig.getUniqueValueReservationTtl() );
+ final MutationBatch write = strat.write( scope, uv, actorSystemFig.getUniqueValueReservationTtl() );
write.execute();
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/ee18950f/stack/corepersistence/collection/src/main/resources/application.conf
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/resources/application.conf b/stack/corepersistence/collection/src/main/resources/application.conf
deleted file mode 100644
index 93854f9..0000000
--- a/stack/corepersistence/collection/src/main/resources/application.conf
+++ /dev/null
@@ -1,28 +0,0 @@
-akka {
-
- loggers = ["akka.event.slf4j.Slf4jLogger"]
- loglevel = "ERROR"
- logging-filter = "akka.event.slf4j.Slf4jLoggingFilter"
-
- actor {
- provider = "akka.cluster.ClusterActorRefProvider"
- }
-
- remote {
- log-remote-lifecycle-events = off
- netty.tcp {
- hostname = "127.0.0.1"
- port = 0
- }
- }
-}
-
-# Disable legacy metrics in akka-cluster.
-akka.cluster.metrics.enabled=off
-
-# Enable metrics extension in akka-cluster-metrics.
-akka.extensions=["akka.cluster.metrics.ClusterMetricsExtension", "akka.cluster.pubsub.DistributedPubSub"]
-
-# Sigar native library extract location during tests.
-# Note: use per-jvm-instance folder when running multiple jvm on one host.
-akka.cluster.metrics.native-library-extract-folder=${user.dir}/target/native
http://git-wip-us.apache.org/repos/asf/usergrid/blob/ee18950f/stack/corepersistence/collection/src/main/resources/cluster-singleton.conf
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/resources/cluster-singleton.conf b/stack/corepersistence/collection/src/main/resources/cluster-singleton.conf
deleted file mode 100644
index 907aebb..0000000
--- a/stack/corepersistence/collection/src/main/resources/cluster-singleton.conf
+++ /dev/null
@@ -1,25 +0,0 @@
-include "application"
-
-akka.actor.deployment {
- /uvRouter/singleton/router {
- router = consistent-hashing-pool
- cluster {
- enabled = on
- allow-local-routees = on
-
- # singleton will only run on nodes with role "io"
- use-role = io
-
- # more forgiving failure detector
- failure-detector {
- threshold = 10
- acceptable-heartbeat-pause = 3 s
- heartbeat-interval = 1 s
- heartbeat-request {
- expected-response-after = 3 s
- }
- }
-
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/ee18950f/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/EntityCollectionManagerIT.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/EntityCollectionManagerIT.java b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/EntityCollectionManagerIT.java
index d94b7b5..6c3cfe7 100644
--- a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/EntityCollectionManagerIT.java
+++ b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/EntityCollectionManagerIT.java
@@ -21,6 +21,7 @@ package org.apache.usergrid.persistence.collection;
import com.fasterxml.uuid.UUIDComparator;
import com.google.inject.Inject;
import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
+import org.apache.usergrid.persistence.actorsystem.ActorSystemManager;
import org.apache.usergrid.persistence.collection.exception.WriteUniqueVerifyException;
import org.apache.usergrid.persistence.collection.guice.TestCollectionModule;
import org.apache.usergrid.persistence.collection.mvcc.entity.Stage;
@@ -28,6 +29,8 @@ import org.apache.usergrid.persistence.collection.serialization.MvccEntitySerial
import org.apache.usergrid.persistence.collection.serialization.SerializationFig;
import org.apache.usergrid.persistence.collection.serialization.UniqueValueSerializationStrategy;
import org.apache.usergrid.persistence.collection.serialization.UniqueValueSet;
+import org.apache.usergrid.persistence.collection.uniquevalues.UniqueValueActor;
+import org.apache.usergrid.persistence.collection.uniquevalues.UniqueValuesService;
import org.apache.usergrid.persistence.collection.util.EntityHelper;
import org.apache.usergrid.persistence.core.guice.MigrationManagerRule;
import org.apache.usergrid.persistence.core.guicyfig.SetConfigTestBypass;
@@ -73,21 +76,30 @@ public class EntityCollectionManagerIT {
@Inject
private SerializationFig serializationFig;
-
@Inject
private UniqueValueSerializationStrategy uniqueValueSerializationStrategy;
@Inject
private MvccEntitySerializationStrategy entitySerializationStrategy;
+ @Inject
+ ActorSystemManager actorSystemManager;
+
+ @Inject
+ UniqueValuesService uniqueValuesService;
+
private static AtomicBoolean startedAkka = new AtomicBoolean( false );
@Before
public void initAkka() {
if ( !startedAkka.getAndSet( true ) ) {
- ApplicationScope context = new ApplicationScopeImpl( new SimpleId( "organization" ) );
- EntityCollectionManager manager = factory.createCollectionManager( context );
- manager.startAkkaForTesting( "127.0.0.1", 2551, "us-east" );
+ actorSystemManager.registerRouterProducer( uniqueValuesService );
+ actorSystemManager.registerMessageType( UniqueValueActor.Request.class, "/user/uvProxy" );
+ actorSystemManager.registerMessageType( UniqueValueActor.Reservation.class, "/user/uvProxy" );
+ actorSystemManager.registerMessageType( UniqueValueActor.Cancellation.class, "/user/uvProxy" );
+ actorSystemManager.registerMessageType( UniqueValueActor.Confirmation.class, "/user/uvProxy" );
+ actorSystemManager.start( "127.0.0.1", 2554, "us-east" );
+ actorSystemManager.waitForRequestActors();
}
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/ee18950f/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/WriteUniqueVerifyIT.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/WriteUniqueVerifyIT.java b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/WriteUniqueVerifyIT.java
index ee610a9..9e29f44 100644
--- a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/WriteUniqueVerifyIT.java
+++ b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/WriteUniqueVerifyIT.java
@@ -18,6 +18,9 @@
package org.apache.usergrid.persistence.collection.mvcc.stage.write;
+import org.apache.usergrid.persistence.actorsystem.ActorSystemManager;
+import org.apache.usergrid.persistence.collection.uniquevalues.UniqueValueActor;
+import org.apache.usergrid.persistence.collection.uniquevalues.UniqueValuesService;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
@@ -69,14 +72,24 @@ public class WriteUniqueVerifyIT {
@Inject
public EntityCollectionManagerFactory cmf;
+ @Inject
+ ActorSystemManager actorSystemManager;
+
+ @Inject
+ UniqueValuesService uniqueValuesService;
+
private static AtomicBoolean startedAkka = new AtomicBoolean( false );
@Before
public void initAkka() {
if ( !startedAkka.getAndSet( true ) ) {
- ApplicationScope context = new ApplicationScopeImpl( new SimpleId( "organization" ) );
- EntityCollectionManager manager = factory.createCollectionManager( context );
- manager.startAkkaForTesting( "127.0.0.1", 2552, "us-east" );
+ actorSystemManager.registerRouterProducer( uniqueValuesService );
+ actorSystemManager.registerMessageType( UniqueValueActor.Request.class, "/user/uvProxy" );
+ actorSystemManager.registerMessageType( UniqueValueActor.Reservation.class, "/user/uvProxy" );
+ actorSystemManager.registerMessageType( UniqueValueActor.Cancellation.class, "/user/uvProxy" );
+ actorSystemManager.registerMessageType( UniqueValueActor.Confirmation.class, "/user/uvProxy" );
+ actorSystemManager.start( "127.0.0.1", 2554, "us-east" );
+ actorSystemManager.waitForRequestActors();
}
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/ee18950f/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/WriteUniqueVerifyTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/WriteUniqueVerifyTest.java b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/WriteUniqueVerifyTest.java
index 196f531..6daef08 100644
--- a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/WriteUniqueVerifyTest.java
+++ b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/WriteUniqueVerifyTest.java
@@ -18,8 +18,11 @@
package org.apache.usergrid.persistence.collection.mvcc.stage.write;
+import org.apache.usergrid.persistence.actorsystem.ActorSystemManager;
import org.apache.usergrid.persistence.collection.EntityCollectionManager;
import org.apache.usergrid.persistence.collection.EntityCollectionManagerFactory;
+import org.apache.usergrid.persistence.collection.uniquevalues.UniqueValueActor;
+import org.apache.usergrid.persistence.collection.uniquevalues.UniqueValuesService;
import org.apache.usergrid.persistence.core.scope.ApplicationScopeImpl;
import org.apache.usergrid.persistence.model.entity.SimpleId;
import org.junit.Before;
@@ -74,15 +77,25 @@ public class WriteUniqueVerifyTest {
@Inject
private CassandraConfig cassandraConfig;
+ @Inject
+ ActorSystemManager actorSystemManager;
+
+ @Inject
+ UniqueValuesService uniqueValuesService;
+
private static AtomicBoolean startedAkka = new AtomicBoolean( false );
@Before
public void initAkka() {
if ( !startedAkka.getAndSet( true ) ) {
- ApplicationScope context = new ApplicationScopeImpl( new SimpleId( "organization" ) );
- EntityCollectionManager manager = factory.createCollectionManager( context );
- manager.startAkkaForTesting( "127.0.0.1", 2553, "us-east" );
+ actorSystemManager.registerRouterProducer( uniqueValuesService );
+ actorSystemManager.registerMessageType( UniqueValueActor.Request.class, "/user/uvProxy" );
+ actorSystemManager.registerMessageType( UniqueValueActor.Reservation.class, "/user/uvProxy" );
+ actorSystemManager.registerMessageType( UniqueValueActor.Cancellation.class, "/user/uvProxy" );
+ actorSystemManager.registerMessageType( UniqueValueActor.Confirmation.class, "/user/uvProxy" );
+ actorSystemManager.start( "127.0.0.1", 2554, "us-east" );
+ actorSystemManager.waitForRequestActors();
}
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/ee18950f/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValuesServiceTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValuesServiceTest.java b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValuesServiceTest.java
index 94d98f5..8db582e 100644
--- a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValuesServiceTest.java
+++ b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/uniquevalues/UniqueValuesServiceTest.java
@@ -1,9 +1,29 @@
+/**
+ * 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.persistence.collection.uniquevalues;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Multimap;
import com.google.common.collect.Multimaps;
import com.google.inject.Inject;
+import org.apache.usergrid.persistence.actorsystem.ActorSystemFig;
+import org.apache.usergrid.persistence.actorsystem.ActorSystemManager;
import org.apache.usergrid.persistence.collection.EntityCollectionManager;
import org.apache.usergrid.persistence.collection.EntityCollectionManagerFactory;
import org.apache.usergrid.persistence.collection.exception.WriteUniqueVerifyException;
@@ -49,7 +69,10 @@ public class UniqueValuesServiceTest {
public MigrationManagerRule migrationManagerRule;
@Inject
- AkkaFig akkaFig;
+ ActorSystemFig actorSystemFig;
+
+ @Inject
+ ActorSystemManager actorSystemManager;
@Inject
UniqueValuesService uniqueValuesService;
@@ -58,14 +81,19 @@ public class UniqueValuesServiceTest {
int numThreads = 6;
int poolSize = 5;
- int numUsers = 100;
+ int numUsers = 1;
+
@Before
public void initAkka() {
if ( !startedAkka.getAndSet( true ) ) {
- ApplicationScope context = new ApplicationScopeImpl( new SimpleId( "organization" ) );
- EntityCollectionManager manager = factory.createCollectionManager( context );
- manager.startAkkaForTesting( "127.0.0.1", 2554, "us-east" );
+ actorSystemManager.registerRouterProducer( uniqueValuesService );
+ actorSystemManager.registerMessageType( UniqueValueActor.Request.class, "/user/uvProxy" );
+ actorSystemManager.registerMessageType( UniqueValueActor.Reservation.class, "/user/uvProxy" );
+ actorSystemManager.registerMessageType( UniqueValueActor.Cancellation.class, "/user/uvProxy" );
+ actorSystemManager.registerMessageType( UniqueValueActor.Confirmation.class, "/user/uvProxy" );
+ actorSystemManager.start( "127.0.0.1", 2554, "us-east" );
+ actorSystemManager.waitForRequestActors();
}
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/ee18950f/stack/corepersistence/common/pom.xml
----------------------------------------------------------------------
diff --git a/stack/corepersistence/common/pom.xml b/stack/corepersistence/common/pom.xml
index 897bb38..a0488b2 100644
--- a/stack/corepersistence/common/pom.xml
+++ b/stack/corepersistence/common/pom.xml
@@ -1,4 +1,20 @@
<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
@@ -83,7 +99,6 @@
<version>${guicyfig.version}</version>
</dependency>
-
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-collections4</artifactId>
http://git-wip-us.apache.org/repos/asf/usergrid/blob/ee18950f/stack/corepersistence/model/pom.xml
----------------------------------------------------------------------
diff --git a/stack/corepersistence/model/pom.xml b/stack/corepersistence/model/pom.xml
index ddf301d..2c2b557 100644
--- a/stack/corepersistence/model/pom.xml
+++ b/stack/corepersistence/model/pom.xml
@@ -1,4 +1,20 @@
<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
http://git-wip-us.apache.org/repos/asf/usergrid/blob/ee18950f/stack/corepersistence/pom.xml
----------------------------------------------------------------------
diff --git a/stack/corepersistence/pom.xml b/stack/corepersistence/pom.xml
index f2bbbcd..5d4ccac 100644
--- a/stack/corepersistence/pom.xml
+++ b/stack/corepersistence/pom.xml
@@ -110,6 +110,7 @@ limitations under the License.
<module>map</module>
<module>queue</module>
<module>cache</module>
+ <module>actorsystem</module>
</modules>
<build>
http://git-wip-us.apache.org/repos/asf/usergrid/blob/ee18950f/stack/corepersistence/queryindex/pom.xml
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/pom.xml b/stack/corepersistence/queryindex/pom.xml
index 003281e..ff032da 100644
--- a/stack/corepersistence/queryindex/pom.xml
+++ b/stack/corepersistence/queryindex/pom.xml
@@ -1,4 +1,20 @@
<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
@@ -13,7 +29,7 @@
<description>Module provates indexing and query of Entities via ElasticSearch</description>
<artifactId>queryindex</artifactId>
- <name>Usergrid Queryindex</name>
+ <name>Usergrid QueryIndex</name>
<build>