You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by to...@apache.org on 2015/04/24 18:01:09 UTC
[43/52] [abbrv] incubator-usergrid git commit: Updated tests and
providers for testing all async index services and index service itself
Updated tests and providers for testing all async index services and index service itself
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/f4d0d1a9
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/f4d0d1a9
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/f4d0d1a9
Branch: refs/heads/USERGRID-593
Commit: f4d0d1a95d8a0f5fc27eae732e61cab3d4940783
Parents: 2a58270
Author: Todd Nine <tn...@apigee.com>
Authored: Wed Apr 22 14:23:35 2015 -0600
Committer: Todd Nine <tn...@apigee.com>
Committed: Wed Apr 22 14:23:35 2015 -0600
----------------------------------------------------------------------
.../usergrid/corepersistence/CoreModule.java | 8 +-
.../corepersistence/CpEntityManager.java | 6 +-
.../corepersistence/CpEntityManagerFactory.java | 6 +-
.../corepersistence/CpRelationManager.java | 6 +-
.../index/AsyncIndexProvider.java | 24 +-
.../index/AsyncIndexService.java | 42 +++
.../index/AsyncReIndexService.java | 42 ---
.../index/InMemoryAsyncIndexService.java | 88 +++++
.../index/InMemoryAsyncReIndexService.java | 108 ------
.../index/IndexProcessorFig.java | 86 +++++
.../corepersistence/index/IndexService.java | 6 +-
.../corepersistence/index/IndexServiceImpl.java | 9 +-
.../corepersistence/index/QueryFig.java | 103 ------
.../index/ReIndexServiceImpl.java | 12 +-
.../index/SQSAsyncIndexService.java | 325 +++++++++++++++++++
.../index/SQSAsyncReIndexService.java | 269 ---------------
.../corepersistence/util/CpNamingUtils.java | 8 +-
.../util/SerializableMapper.java | 4 +-
.../index/AsyncIndexServiceTest.java | 203 ++++++++++++
.../index/InMemoryAsycIndexServiceTest.java | 64 ++++
.../corepersistence/index/IndexServiceTest.java | 233 ++++++++++++-
.../index/SQSAsyncIndexServiceTest.java | 143 +++-----
.../impl/migration/EntityIdScope.java | 40 ++-
.../migration/data/MigrationRelationship.java | 29 +-
.../migration/data/VersionedMigrationSet.java | 31 +-
.../persistence/core/rx/RxSchedulerFig.java | 4 +-
.../core/scope/ApplicationScope.java | 4 +
.../core/scope/ApplicationScopeImpl.java | 21 +-
.../data/VersionedMigrationSetTest.java | 8 +-
.../usergrid/persistence/model/entity/Id.java | 3 +
.../persistence/model/entity/SimpleId.java | 37 ++-
.../index/impl/IndexOperationMessage.java | 8 -
32 files changed, 1259 insertions(+), 721 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f4d0d1a9/stack/core/src/main/java/org/apache/usergrid/corepersistence/CoreModule.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CoreModule.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CoreModule.java
index 51972d8..06fe058 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CoreModule.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CoreModule.java
@@ -22,10 +22,10 @@ import org.apache.usergrid.corepersistence.events.EntityDeletedHandler;
import org.apache.usergrid.corepersistence.events.EntityVersionCreatedHandler;
import org.apache.usergrid.corepersistence.events.EntityVersionDeletedHandler;
import org.apache.usergrid.corepersistence.index.AsyncIndexProvider;
-import org.apache.usergrid.corepersistence.index.AsyncReIndexService;
+import org.apache.usergrid.corepersistence.index.AsyncIndexService;
import org.apache.usergrid.corepersistence.index.IndexService;
import org.apache.usergrid.corepersistence.index.IndexServiceImpl;
-import org.apache.usergrid.corepersistence.index.QueryFig;
+import org.apache.usergrid.corepersistence.index.IndexProcessorFig;
import org.apache.usergrid.corepersistence.migration.AppInfoMigrationPlugin;
import org.apache.usergrid.corepersistence.migration.CoreMigration;
import org.apache.usergrid.corepersistence.migration.CoreMigrationPlugin;
@@ -151,9 +151,9 @@ public class CoreModule extends AbstractModule {
bind( IndexService.class ).to( IndexServiceImpl.class );
//bind the queue provider
- bind( AsyncReIndexService.class).toProvider( AsyncIndexProvider.class );
+ bind( AsyncIndexService.class).toProvider( AsyncIndexProvider.class );
- install( new GuicyFigModule( QueryFig.class ) );
+ install( new GuicyFigModule( IndexProcessorFig.class ) );
install( new GuicyFigModule( ApplicationIdCacheFig.class ) );
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f4d0d1a9/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManager.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManager.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManager.java
index 72ca955..23514c8 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManager.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManager.java
@@ -37,7 +37,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;
-import org.apache.usergrid.corepersistence.index.AsyncReIndexService;
+import org.apache.usergrid.corepersistence.index.AsyncIndexService;
import org.apache.usergrid.corepersistence.util.CpEntityMapUtils;
import org.apache.usergrid.corepersistence.util.CpNamingUtils;
import org.apache.usergrid.persistence.AggregateCounter;
@@ -175,7 +175,7 @@ public class CpEntityManager implements EntityManager {
private final CounterUtils counterUtils;
- private final AsyncReIndexService indexService;
+ private final AsyncIndexService indexService;
private boolean skipAggregateCounters;
private MetricsFactory metricsFactory;
@@ -215,7 +215,7 @@ public class CpEntityManager implements EntityManager {
* @param metricsFactory
* @param applicationId
*/
- public CpEntityManager(final CassandraService cass, final CounterUtils counterUtils, final AsyncReIndexService indexService, final ManagerCache managerCache, final MetricsFactory metricsFactory, final UUID applicationId ) {
+ public CpEntityManager(final CassandraService cass, final CounterUtils counterUtils, final AsyncIndexService indexService, final ManagerCache managerCache, final MetricsFactory metricsFactory, final UUID applicationId ) {
Preconditions.checkNotNull( cass, "cass must not be null" );
Preconditions.checkNotNull( counterUtils, "counterUtils must not be null" );
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f4d0d1a9/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManagerFactory.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManagerFactory.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManagerFactory.java
index 46db3f8..f08bce4 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManagerFactory.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManagerFactory.java
@@ -33,7 +33,7 @@ import org.springframework.context.ApplicationContextAware;
import org.apache.commons.lang.StringUtils;
-import org.apache.usergrid.corepersistence.index.AsyncReIndexService;
+import org.apache.usergrid.corepersistence.index.AsyncIndexService;
import org.apache.usergrid.corepersistence.index.ReIndexService;
import org.apache.usergrid.corepersistence.util.CpNamingUtils;
import org.apache.usergrid.exception.ConflictException;
@@ -122,7 +122,7 @@ public class CpEntityManagerFactory implements EntityManagerFactory, Application
private Injector injector;
private final EntityIndex entityIndex;
private final MetricsFactory metricsFactory;
- private final AsyncReIndexService indexService;
+ private final AsyncIndexService indexService;
public CpEntityManagerFactory( final CassandraService cassandraService, final CounterUtils counterUtils,
final Injector injector) {
@@ -134,7 +134,7 @@ public class CpEntityManagerFactory implements EntityManagerFactory, Application
this.entityIndexFactory = injector.getInstance(EntityIndexFactory.class);
this.managerCache = injector.getInstance( ManagerCache.class );
this.metricsFactory = injector.getInstance( MetricsFactory.class );
- this.indexService = injector.getInstance( AsyncReIndexService.class );
+ this.indexService = injector.getInstance( AsyncIndexService.class );
this.applicationIdCache = injector.getInstance(ApplicationIdCacheFactory.class).getInstance(
getManagementEntityManager() );
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f4d0d1a9/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
index 8f125ad..ea554a3 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
@@ -30,7 +30,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;
-import org.apache.usergrid.corepersistence.index.AsyncReIndexService;
+import org.apache.usergrid.corepersistence.index.AsyncIndexService;
import org.apache.usergrid.corepersistence.results.CollectionResultsLoaderFactoryImpl;
import org.apache.usergrid.corepersistence.results.ConnectionResultsLoaderFactoryImpl;
import org.apache.usergrid.corepersistence.results.ElasticSearchQueryExecutor;
@@ -118,13 +118,13 @@ public class CpRelationManager implements RelationManager {
private final ApplicationScope applicationScope;
- private final AsyncReIndexService indexService;
+ private final AsyncIndexService indexService;
private MetricsFactory metricsFactory;
private Timer updateCollectionTimer;
- public CpRelationManager( final MetricsFactory metricsFactory, final ManagerCache managerCache, final AsyncReIndexService indexService, final EntityManager em, final UUID applicationId, final EntityRef headEntity) {
+ public CpRelationManager( final MetricsFactory metricsFactory, final ManagerCache managerCache, final AsyncIndexService indexService, final EntityManager em, final UUID applicationId, final EntityRef headEntity) {
Assert.notNull( em, "Entity manager cannot be null" );
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f4d0d1a9/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/AsyncIndexProvider.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/AsyncIndexProvider.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/AsyncIndexProvider.java
index 18df824..0043166 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/AsyncIndexProvider.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/AsyncIndexProvider.java
@@ -20,7 +20,6 @@
package org.apache.usergrid.corepersistence.index;
-import org.apache.usergrid.corepersistence.rx.impl.AllEntityIdsObservable;
import org.apache.usergrid.persistence.collection.EntityCollectionManagerFactory;
import org.apache.usergrid.persistence.core.rx.RxTaskScheduler;
import org.apache.usergrid.persistence.core.metrics.MetricsFactory;
@@ -35,9 +34,9 @@ import com.google.inject.Singleton;
* A provider to allow users to configure their queue impl via properties
*/
@Singleton
-public class AsyncIndexProvider implements Provider<AsyncReIndexService> {
+public class AsyncIndexProvider implements Provider<AsyncIndexService> {
- private final QueryFig queryFig;
+ private final IndexProcessorFig indexProcessorFig;
private final QueueManagerFactory queueManagerFactory;
private final MetricsFactory metricsFactory;
@@ -45,15 +44,15 @@ public class AsyncIndexProvider implements Provider<AsyncReIndexService> {
private final RxTaskScheduler rxTaskScheduler;
private final EntityCollectionManagerFactory entityCollectionManagerFactory;
- private AsyncReIndexService asyncIndexService;
+ private AsyncIndexService asyncIndexService;
@Inject
- public AsyncIndexProvider( final QueryFig queryFig, final QueueManagerFactory queueManagerFactory,
+ public AsyncIndexProvider( final IndexProcessorFig indexProcessorFig, final QueueManagerFactory queueManagerFactory,
final MetricsFactory metricsFactory, final IndexService indexService,
final RxTaskScheduler rxTaskScheduler,
final EntityCollectionManagerFactory entityCollectionManagerFactory ) {
- this.queryFig = queryFig;
+ this.indexProcessorFig = indexProcessorFig;
this.queueManagerFactory = queueManagerFactory;
this.metricsFactory = metricsFactory;
this.indexService = indexService;
@@ -64,7 +63,7 @@ public class AsyncIndexProvider implements Provider<AsyncReIndexService> {
@Override
@Singleton
- public AsyncReIndexService get() {
+ public AsyncIndexService get() {
if ( asyncIndexService == null ) {
asyncIndexService = getIndexService();
}
@@ -74,17 +73,18 @@ public class AsyncIndexProvider implements Provider<AsyncReIndexService> {
}
- private AsyncReIndexService getIndexService() {
- final String value = queryFig.getQueueImplementation();
+ private AsyncIndexService getIndexService() {
+ final String value = indexProcessorFig.getQueueImplementation();
final Implementations impl = Implementations.valueOf( value );
switch ( impl ) {
case LOCAL:
- return new InMemoryAsyncReIndexService( indexService, rxTaskScheduler,
- entityCollectionManagerFactory, metricsFactory );
+ return new InMemoryAsyncIndexService( indexService, rxTaskScheduler,
+ entityCollectionManagerFactory );
case SQS:
- return new SQSAsyncReIndexService( queueManagerFactory, queryFig, metricsFactory );
+ return new SQSAsyncIndexService( queueManagerFactory, indexProcessorFig, metricsFactory, indexService,
+ entityCollectionManagerFactory, rxTaskScheduler );
default:
throw new IllegalArgumentException( "Configuration value of " + getErrorValues() + " are allowed" );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f4d0d1a9/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/AsyncIndexService.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/AsyncIndexService.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/AsyncIndexService.java
new file mode 100644
index 0000000..dac92c8
--- /dev/null
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/AsyncIndexService.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.usergrid.corepersistence.index;
+
+
+import org.apache.usergrid.persistence.core.scope.ApplicationScope;
+import org.apache.usergrid.persistence.model.entity.Entity;
+
+
+/**
+ * Low level queue service for indexing entities
+ */
+public interface AsyncIndexService extends ReIndexAction {
+
+
+ /**
+ * Queue an entity to be indexed. This will start processing immediately. For implementations that are realtime (akka, in memory)
+ * We will return a distributed future. For SQS impls, this will return immediately, and the result will not be available.
+ * After SQS is removed, the tests should be enhanced to ensure that we're processing our queues correctly.
+ * @param applicationScope
+ * @param entity The entity to index
+ */
+ void queueEntityIndexUpdate( final ApplicationScope applicationScope, final Entity entity);
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f4d0d1a9/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/AsyncReIndexService.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/AsyncReIndexService.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/AsyncReIndexService.java
deleted file mode 100644
index c6eedd7..0000000
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/AsyncReIndexService.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.usergrid.corepersistence.index;
-
-
-import org.apache.usergrid.persistence.core.scope.ApplicationScope;
-import org.apache.usergrid.persistence.model.entity.Entity;
-
-
-/**
- * Low level queue service for indexing entities
- */
-public interface AsyncReIndexService extends ReIndexAction {
-
-
- /**
- * Queue an entity to be indexed. This will start processing immediately. For implementations that are realtime (akka, in memory)
- * We will return a distributed future. For SQS impls, this will return immediately, and the result will not be available.
- * After SQS is removed, the tests should be enhanced to ensure that we're processing our queues correctly.
- * @param applicationScope
- * @param entity The entity to index
- */
- void queueEntityIndexUpdate( final ApplicationScope applicationScope, final Entity entity);
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f4d0d1a9/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/InMemoryAsyncIndexService.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/InMemoryAsyncIndexService.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/InMemoryAsyncIndexService.java
new file mode 100644
index 0000000..e8d178c
--- /dev/null
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/InMemoryAsyncIndexService.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.usergrid.corepersistence.index;
+
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.usergrid.persistence.collection.EntityCollectionManagerFactory;
+import org.apache.usergrid.persistence.collection.serialization.impl.migration.EntityIdScope;
+import org.apache.usergrid.persistence.core.rx.RxTaskScheduler;
+import org.apache.usergrid.persistence.core.scope.ApplicationScope;
+import org.apache.usergrid.persistence.index.impl.IndexOperationMessage;
+import org.apache.usergrid.persistence.model.entity.Entity;
+import org.apache.usergrid.persistence.model.entity.Id;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+
+import rx.Observable;
+
+
+@Singleton
+public class InMemoryAsyncIndexService implements AsyncIndexService {
+
+ private static final Logger log = LoggerFactory.getLogger( InMemoryAsyncIndexService.class );
+
+ private final IndexService indexService;
+ private final RxTaskScheduler rxTaskScheduler;
+ private final EntityCollectionManagerFactory entityCollectionManagerFactory;
+
+
+ @Inject
+ public InMemoryAsyncIndexService( final IndexService indexService, final RxTaskScheduler rxTaskScheduler,
+ final EntityCollectionManagerFactory entityCollectionManagerFactory ) {
+ this.indexService = indexService;
+ this.rxTaskScheduler = rxTaskScheduler;
+ this.entityCollectionManagerFactory = entityCollectionManagerFactory;
+ }
+
+
+ @Override
+ public void queueEntityIndexUpdate( final ApplicationScope applicationScope, final Entity entity ) {
+
+ //process the entity immediately
+ //only process the same version, otherwise ignore
+
+
+ log.debug( "Indexing entity {} in app scope {} ", entity, applicationScope );
+
+ final Observable<IndexOperationMessage> edgeObservable = indexService.indexEntity( applicationScope, entity );
+
+ //start it in the background on an i/o thread
+ edgeObservable.subscribeOn( rxTaskScheduler.getAsyncIOScheduler() ).subscribe();
+ }
+
+
+ @Override
+ public void index( final EntityIdScope entityIdScope ) {
+
+ final ApplicationScope applicationScope = entityIdScope.getApplicationScope();
+
+ final Id entityId = entityIdScope.getId();
+
+ //load the entity
+ entityCollectionManagerFactory.createCollectionManager( applicationScope ).load( entityId )
+ //perform indexing on the task scheduler and start it
+ .flatMap( entity -> indexService.indexEntity( applicationScope, entity ) )
+ .subscribeOn( rxTaskScheduler.getAsyncIOScheduler() ).subscribe();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f4d0d1a9/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/InMemoryAsyncReIndexService.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/InMemoryAsyncReIndexService.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/InMemoryAsyncReIndexService.java
deleted file mode 100644
index 4908945..0000000
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/InMemoryAsyncReIndexService.java
+++ /dev/null
@@ -1,108 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.usergrid.corepersistence.index;
-
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.usergrid.persistence.collection.EntityCollectionManagerFactory;
-import org.apache.usergrid.persistence.collection.serialization.impl.migration.EntityIdScope;
-import org.apache.usergrid.persistence.core.metrics.MetricsFactory;
-import org.apache.usergrid.persistence.core.rx.RxTaskScheduler;
-import org.apache.usergrid.persistence.core.scope.ApplicationScope;
-import org.apache.usergrid.persistence.index.impl.IndexOperationMessage;
-import org.apache.usergrid.persistence.model.entity.Entity;
-import org.apache.usergrid.persistence.model.entity.Id;
-
-import com.codahale.metrics.Timer;
-import com.google.inject.Inject;
-import com.google.inject.Singleton;
-
-import rx.Observable;
-
-
-@Singleton
-public class InMemoryAsyncReIndexService implements AsyncReIndexService {
-
- private static final Logger log = LoggerFactory.getLogger( InMemoryAsyncReIndexService.class );
- private final IndexService indexService;
- private final RxTaskScheduler rxTaskScheduler;
- private final EntityCollectionManagerFactory entityCollectionManagerFactory;
- private final Timer timer;
-
-
- @Inject
- public InMemoryAsyncReIndexService( final IndexService indexService, final RxTaskScheduler rxTaskScheduler,
- final EntityCollectionManagerFactory entityCollectionManagerFactory, final
- MetricsFactory metricsFactory ) {
- this.indexService = indexService;
- this.rxTaskScheduler = rxTaskScheduler;
- this.entityCollectionManagerFactory = entityCollectionManagerFactory;
-
- timer = metricsFactory.getTimer( InMemoryAsyncReIndexService.class, "IndexTimer" );
- }
-
-
- @Override
- public void queueEntityIndexUpdate( final ApplicationScope applicationScope, final Entity entity ) {
-
- //process the entity immediately
- //only process the same version, otherwise ignore
-
-
- log.debug( "Indexing entity {} in app scope {} ", entity, applicationScope );
-
- final Observable<IndexOperationMessage> edgeObservable = indexService.indexEntity( applicationScope, entity );
-
-
-
- edgeObservable.subscribeOn( rxTaskScheduler.getAsyncIOScheduler() ).subscribe();
-
- //now start it
-// final Timer.Context time = timer.time();
-//
-// edgeObservable.connect();
-//
-// time.stop();
-
-
- }
-
-
- @Override
- public void index( final EntityIdScope entityIdScope ) {
-
- final ApplicationScope applicationScope = entityIdScope.getApplicationScope();
-
- final Id entityId = entityIdScope.getId();
-
- final Entity entity =
- entityCollectionManagerFactory.createCollectionManager( applicationScope ).load( entityId ).toBlocking()
- .lastOrDefault( null );
-
-
- if ( entity == null ) {
- log.warn( "Could not find entity with id {} in app scope {} ", entityId, applicationScope );
- }
-
- indexService.indexEntity( applicationScope, entity );
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f4d0d1a9/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexProcessorFig.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexProcessorFig.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexProcessorFig.java
new file mode 100644
index 0000000..1e8abff
--- /dev/null
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexProcessorFig.java
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.usergrid.corepersistence.index;
+
+
+import org.safehaus.guicyfig.Default;
+import org.safehaus.guicyfig.GuicyFig;
+import org.safehaus.guicyfig.Key;
+
+
+/**
+ * Application id cache fig
+ */
+public interface IndexProcessorFig extends GuicyFig {
+
+
+ /**
+ * Amount of time in milliseconds to wait when ES rejects our request before retrying. Provides simple
+ * backpressure
+ */
+ String FAILURE_REJECTED_RETRY_WAIT_TIME = "elasticsearch.rejected_retry_wait";
+
+ /**
+ * The number of worker threads to consume from the queue
+ */
+ String ELASTICSEARCH_WORKER_COUNT = "elasticsearch.worker_count";
+
+
+ /**
+ * The queue implementation to use. Values come from <class>QueueProvider.Implementations</class>
+ */
+ String ELASTICSEARCH_QUEUE_IMPL = "elasticsearch.queue_impl";
+
+
+ /**
+ * The queue implementation to use. Values come from <class>QueueProvider.Implementations</class>
+ */
+ String ELASTICSEARCH_QUEUE_OFFER_TIMEOUT = "elasticsearch.queue.offer_timeout";
+
+ /**
+ * Amount of time to wait when reading from the queue
+ */
+ String INDEX_QUEUE_READ_TIMEOUT = "elasticsearch.queue_read_timeout";
+
+
+
+ @Default( "1000" )
+ @Key( FAILURE_REJECTED_RETRY_WAIT_TIME )
+ long getFailureRetryTime();
+
+ //give us 60 seconds to process the message
+ @Default( "60" )
+ @Key( INDEX_QUEUE_READ_TIMEOUT )
+ int getIndexQueueTimeout();
+
+ @Default( "1" )
+ @Key( ELASTICSEARCH_WORKER_COUNT )
+ int getWorkerCount();
+
+ @Default( "LOCAL" )
+ @Key( ELASTICSEARCH_QUEUE_IMPL )
+ String getQueueImplementation();
+
+
+ @Default("30000")
+ @Key("elasticsearch.reindex.sample.interval")
+ long getReIndexSampleInterval();
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f4d0d1a9/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexService.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexService.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexService.java
index 5e9392b..18ab2b7 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexService.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexService.java
@@ -41,9 +41,9 @@ public interface IndexService {
* @param applicationScope The scope of the entity
* @param entity The entity
*
- * @return A ConnectableObservable with every edge in the batch to index the entity. Note that this a cold observable
- * and must be subscribed to, then "connect" in order to perform the operation. This also makes no assumptions on scheduling. It is up to the caller
- * to assign the correct scheduler to the observable
+ * @return An Observable with executed batch future as an observable. Note that this a cold observable
+ * and must be subscribed to in order to perform the index operations. This also makes no assumptions on scheduling. It is up to the caller
+ * to assign the correct scheduler to the observable based on their threading needs
*/
Observable<IndexOperationMessage> indexEntity( final ApplicationScope applicationScope, final Entity entity );
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f4d0d1a9/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexServiceImpl.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexServiceImpl.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexServiceImpl.java
index 87e22f4..81bf6cb 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexServiceImpl.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexServiceImpl.java
@@ -99,7 +99,7 @@ public class IndexServiceImpl implements IndexService {
//we might or might not need to index from target-> source
- final Observable<IndexEdge> targetSizes = getIndexEdgesToTarget( gm, entityId );
+ final Observable<IndexEdge> targetSizes = getIndexEdgesAsTarget( gm, entityId );
//merge the edges together
@@ -129,12 +129,13 @@ public class IndexServiceImpl implements IndexService {
/**
- * Get index edges to the target
+ * Get index edges to the target. Used in only certain entity types, such as roles, users, groups etc
+ * where we doubly index on both directions of the edge
*
* @param graphManager The graph manager
- * @param entityId The entitie's id
+ * @param entityId The entity's id
*/
- private Observable<IndexEdge> getIndexEdgesToTarget( final GraphManager graphManager, final Id entityId ) {
+ private Observable<IndexEdge> getIndexEdgesAsTarget( final GraphManager graphManager, final Id entityId ) {
final String collectionName = InflectionUtils.pluralize( entityId.getType() );
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f4d0d1a9/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/QueryFig.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/QueryFig.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/QueryFig.java
deleted file mode 100644
index 82ed496..0000000
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/QueryFig.java
+++ /dev/null
@@ -1,103 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.usergrid.corepersistence.index;
-
-
-import org.safehaus.guicyfig.Default;
-import org.safehaus.guicyfig.GuicyFig;
-import org.safehaus.guicyfig.Key;
-
-
-/**
- * Application id cache fig
- */
-public interface QueryFig extends GuicyFig {
-
-
- /**
- * Amount of time in milliseconds to wait when ES rejects our request before retrying. Provides simple
- * backpressure
- */
- public static final String FAILURE_REJECTED_RETRY_WAIT_TIME = "elasticsearch.rejected_retry_wait";
-
- /**
- * The number of worker threads to consume from the queue
- */
- public static final String ELASTICSEARCH_WORKER_COUNT = "elasticsearch.worker_count";
-
- /**
- * The queue implementation to use. Values come from <class>QueueProvider.Implementations</class>
- */
- public static final String ELASTICSEARCH_QUEUE_IMPL = "elasticsearch.queue_impl";
-
-
- /**
- * The queue implementation to use. Values come from <class>QueueProvider.Implementations</class>
- */
- public static final String ELASTICSEARCH_QUEUE_OFFER_TIMEOUT = "elasticsearch.queue.offer_timeout";
-
- /**
- * Amount of time to wait when reading from the queue
- */
- public static final String INDEX_QUEUE_READ_TIMEOUT = "elasticsearch.queue_read_timeout";
-
- /**
- * Amount of time to wait when reading from the queue in milliseconds
- */
- public static final String INDEX_QUEUE_TRANSACTION_TIMEOUT = "elasticsearch.queue_transaction_timeout";
-
-
- String INDEX_QUEUE_SIZE = "elasticsearch.queue_size";
-
-
- @Default( "1000" )
- @Key( FAILURE_REJECTED_RETRY_WAIT_TIME )
- long getFailureRetryTime();
-
- //give us 60 seconds to process the message
- @Default( "60" )
- @Key( INDEX_QUEUE_READ_TIMEOUT )
- int getIndexQueueTimeout();
-
- @Default( "2" )
- @Key( ELASTICSEARCH_WORKER_COUNT )
- int getWorkerCount();
-
- @Default( "LOCAL" )
- @Key( ELASTICSEARCH_QUEUE_IMPL )
- String getQueueImplementation();
-
- @Default( "1000" )
- @Key( ELASTICSEARCH_QUEUE_OFFER_TIMEOUT )
- long getQueueOfferTimeout();
-
- /**
- * size of the buffer to build up before you send results
- */
- @Default( "1000" )
- @Key( INDEX_QUEUE_SIZE )
- int getIndexQueueSize();
-
-
- @Default("30000")
- @Key("elasticsearch.reindex.sample.interval")
- long getReIndexSampleInterval();
-
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f4d0d1a9/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/ReIndexServiceImpl.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/ReIndexServiceImpl.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/ReIndexServiceImpl.java
index 3553c87..5bf7957 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/ReIndexServiceImpl.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/ReIndexServiceImpl.java
@@ -60,20 +60,20 @@ public class ReIndexServiceImpl implements ReIndexService {
private final AllApplicationsObservable allApplicationsObservable;
private final AllEntityIdsObservable allEntityIdsObservable;
- private final QueryFig queryFig;
+ private final IndexProcessorFig indexProcessorFig;
private final RxTaskScheduler rxTaskScheduler;
private final MapManager mapManager;
- private final AsyncReIndexService indexService;
+ private final AsyncIndexService indexService;
@Inject
public ReIndexServiceImpl( final AllEntityIdsObservable allEntityIdsObservable,
final MapManagerFactory mapManagerFactory,
- final AllApplicationsObservable allApplicationsObservable, final QueryFig queryFig,
- final RxTaskScheduler rxTaskScheduler, final AsyncReIndexService indexService ) {
+ final AllApplicationsObservable allApplicationsObservable, final IndexProcessorFig indexProcessorFig,
+ final RxTaskScheduler rxTaskScheduler, final AsyncIndexService indexService ) {
this.allEntityIdsObservable = allEntityIdsObservable;
this.allApplicationsObservable = allApplicationsObservable;
- this.queryFig = queryFig;
+ this.indexProcessorFig = indexProcessorFig;
this.rxTaskScheduler = rxTaskScheduler;
this.indexService = indexService;
@@ -108,7 +108,7 @@ public class ReIndexServiceImpl implements ReIndexService {
//start our sampler and state persistence
//take a sample every sample interval to allow us to resume state with minimal loss
- runningReIndex.sample( queryFig.getReIndexSampleInterval(), TimeUnit.MILLISECONDS,
+ runningReIndex.sample( indexProcessorFig.getReIndexSampleInterval(), TimeUnit.MILLISECONDS,
rxTaskScheduler.getAsyncIOScheduler() )
.doOnNext( edge -> {
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f4d0d1a9/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/SQSAsyncIndexService.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/SQSAsyncIndexService.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/SQSAsyncIndexService.java
new file mode 100644
index 0000000..d060123
--- /dev/null
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/SQSAsyncIndexService.java
@@ -0,0 +1,325 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.usergrid.corepersistence.index;
+
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.usergrid.persistence.collection.EntityCollectionManager;
+import org.apache.usergrid.persistence.collection.EntityCollectionManagerFactory;
+import org.apache.usergrid.persistence.collection.serialization.impl.migration.EntityIdScope;
+import org.apache.usergrid.persistence.core.metrics.MetricsFactory;
+import org.apache.usergrid.persistence.core.metrics.ObservableTimer;
+import org.apache.usergrid.persistence.core.rx.RxTaskScheduler;
+import org.apache.usergrid.persistence.core.scope.ApplicationScope;
+import org.apache.usergrid.persistence.index.impl.IndexOperationMessage;
+import org.apache.usergrid.persistence.model.entity.Entity;
+import org.apache.usergrid.persistence.queue.QueueManager;
+import org.apache.usergrid.persistence.queue.QueueManagerFactory;
+import org.apache.usergrid.persistence.queue.QueueMessage;
+import org.apache.usergrid.persistence.queue.QueueScope;
+import org.apache.usergrid.persistence.queue.impl.QueueScopeImpl;
+
+import com.codahale.metrics.Counter;
+import com.codahale.metrics.Gauge;
+import com.codahale.metrics.Timer;
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+
+import rx.Observable;
+import rx.Subscriber;
+import rx.Subscription;
+import rx.schedulers.Schedulers;
+
+
+@Singleton
+public class SQSAsyncIndexService implements AsyncIndexService {
+
+
+ private static final Logger log = LoggerFactory.getLogger( SQSAsyncIndexService.class );
+
+ /**
+ * Set our TTL to 1 month. This is high, but in the event of a bug, we want these entries to get removed
+ */
+ public static final int TTL = 60 * 60 * 24 * 30;
+
+
+ private static final int MAX_TAKE = 10;
+
+ private static final String QUEUE_NAME = "es_queue";
+
+ private final QueueManager queue;
+ private final IndexProcessorFig indexProcessorFig;
+ private final IndexService indexService;
+ private final EntityCollectionManagerFactory entityCollectionManagerFactory;
+ private final RxTaskScheduler rxTaskScheduler;
+
+ private final Timer readTimer;
+ private final Timer writeTimer;
+ private final Timer messageProcessingTimer;
+
+ private final Object mutex = new Object();
+
+
+ private final Counter indexErrorCounter;
+ private final AtomicLong counter = new AtomicLong();
+ private final AtomicLong inFlight = new AtomicLong();
+
+ //the actively running subscription
+ private List<Subscription> subscriptions = new ArrayList<>();
+
+
+ @Inject
+ public SQSAsyncIndexService( final QueueManagerFactory queueManagerFactory, final IndexProcessorFig indexProcessorFig,
+ final MetricsFactory metricsFactory, final IndexService indexService,
+ final EntityCollectionManagerFactory entityCollectionManagerFactory,
+ final RxTaskScheduler rxTaskScheduler ) {
+
+ this.indexService = indexService;
+ this.entityCollectionManagerFactory = entityCollectionManagerFactory;
+ this.rxTaskScheduler = rxTaskScheduler;
+
+ final QueueScope queueScope = new QueueScopeImpl( QUEUE_NAME );
+ this.queue = queueManagerFactory.getQueueManager( queueScope );
+ this.indexProcessorFig = indexProcessorFig;
+
+ this.writeTimer = metricsFactory.getTimer( SQSAsyncIndexService.class, "write" );
+ this.readTimer = metricsFactory.getTimer( SQSAsyncIndexService.class, "read" );
+ this.messageProcessingTimer = metricsFactory.getTimer( SQSAsyncIndexService.class, "message.processing" );
+ this.indexErrorCounter = metricsFactory.getCounter( SQSAsyncIndexService.class, "error" );
+
+
+ //wire up the gauge of inflight messages
+ metricsFactory.addGauge( SQSAsyncIndexService.class, "inflight.meter", new Gauge<Long>() {
+ @Override
+ public Long getValue() {
+ return inFlight.longValue();
+ }
+ } );
+
+ start();
+ }
+
+
+ /**
+ * Offer the EntityIdScope to SQS
+ */
+ private void offer( final EntityIdScope operation ) {
+ final Timer.Context timer = this.writeTimer.time();
+
+ try {
+ //signal to SQS
+ this.queue.sendMessage( operation );
+ }
+ catch ( IOException e ) {
+ throw new RuntimeException( "Unable to queue message", e );
+ }
+ finally {
+ timer.stop();
+ }
+ }
+
+
+ /**
+ * Take messages from SQS
+ */
+ public List<QueueMessage> take() {
+
+ //SQS doesn't support more than 10
+ final Timer.Context timer = this.readTimer.time();
+
+ try {
+ return queue.getMessages( MAX_TAKE, indexProcessorFig.getIndexQueueTimeout(), indexProcessorFig.getIndexQueueTimeout(),
+ EntityIdScope.class );
+ }
+ //stop our timer
+ finally {
+ timer.stop();
+ }
+ }
+
+
+ /**
+ * Ack messages in SQS
+ */
+ public void ack( final List<QueueMessage> messages ) {
+
+ /**
+ * No op
+ */
+ if ( messages.size() == 0 ) {
+ return;
+ }
+
+ queue.commitMessages( messages );
+ }
+
+
+ @Override
+ public void index( final EntityIdScope entityIdScope ) {
+ //queue the re-inex operation
+ offer( entityIdScope );
+ }
+
+
+ @Override
+ public void queueEntityIndexUpdate( final ApplicationScope applicationScope, final Entity entity ) {
+
+ //create our scope
+ final EntityIdScope entityIdScope = new EntityIdScope( applicationScope, entity.getId() );
+
+ //send it to SQS for indexing
+ index( entityIdScope );
+ }
+
+
+ /**
+ * Index an entity and return an observable of the queue message on success
+ */
+ private Observable<IndexOperationMessage> indexEntity( final QueueMessage queueMessage ) {
+ final EntityIdScope entityIdScope = ( EntityIdScope ) queueMessage.getBody();
+ final ApplicationScope applicationScope = entityIdScope.getApplicationScope();
+ final EntityCollectionManager entityCollectionManager =
+ entityCollectionManagerFactory.createCollectionManager( applicationScope );
+
+
+ //run the index operation from the entity
+ return entityCollectionManager.load( entityIdScope.getId() )
+ //invoke the indexing and take the last value
+ .flatMap( entity -> indexService.indexEntity( applicationScope, entity ).last() );
+ }
+
+
+ /**
+ * Do the indexing for a list of queue messages
+ */
+ private void doIndex( final List<QueueMessage> queueMessages ) {
+ //create parallel observables to process all 10 messages
+ final Observable<Long> observable = Observable.from( queueMessages ).flatMap( ( QueueMessage queueMessage ) -> {
+ return indexEntity( queueMessage ).subscribeOn( rxTaskScheduler.getAsyncIOScheduler() );
+ }, MAX_TAKE ).countLong()
+
+ //remove our in flight
+ .doOnNext( count -> inFlight.addAndGet( -1 * count ) )
+
+ //do on completed ack messages. Assumes no expections were thrown
+ .doOnCompleted( () -> ack( queueMessages ) );
+
+ //wrap with our timer and fire
+ ObservableTimer.time( observable, messageProcessingTimer ).subscribe();
+ }
+
+
+ /**
+ * Loop throught and start the workers
+ */
+ public void start() {
+ final int count = indexProcessorFig.getWorkerCount();
+
+ for ( int i = 0; i < count; i++ ) {
+ startWorker();
+ }
+ }
+
+
+ /**
+ * Stop the workers
+ */
+ public void stop() {
+ synchronized ( mutex ) {
+ //stop consuming
+
+ for ( final Subscription subscription : subscriptions ) {
+ subscription.unsubscribe();
+ }
+ }
+ }
+
+
+ private void startWorker() {
+ synchronized ( mutex ) {
+
+ Observable<List<QueueMessage>> consumer =
+ Observable.create( new Observable.OnSubscribe<List<QueueMessage>>() {
+ @Override
+ public void call( final Subscriber<? super List<QueueMessage>> subscriber ) {
+
+ //name our thread so it's easy to see
+ Thread.currentThread().setName( "QueueConsumer_" + counter.incrementAndGet() );
+
+ List<QueueMessage> drainList = null;
+
+ do {
+ Timer.Context timer = readTimer.time();
+
+ try {
+ drainList = take();
+
+ //emit our list in it's entirity to hand off to a worker pool
+ subscriber.onNext( drainList );
+
+ //take since we're in flight
+ inFlight.addAndGet( drainList.size() );
+ }
+
+ catch ( Throwable t ) {
+ final long sleepTime = indexProcessorFig.getFailureRetryTime();
+
+ log.error( "Failed to dequeue. Sleeping for {} milliseconds", sleepTime, t );
+
+ if ( drainList != null ) {
+ inFlight.addAndGet( -1 * drainList.size() );
+ }
+
+
+ try {
+ Thread.sleep( sleepTime );
+ }
+ catch ( InterruptedException ie ) {
+ //swallow
+ }
+
+ indexErrorCounter.inc();
+ }
+
+ finally{
+ timer.stop();
+ }
+ }
+ while ( true );
+ }
+ } )
+ //this won't block our read loop, just reads and proceeds
+ .doOnNext( messages -> doIndex( messages ) ).subscribeOn( Schedulers.newThread() );
+
+ //start in the background
+
+ final Subscription subscription = consumer.subscribe();
+
+ subscriptions.add( subscription );
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f4d0d1a9/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/SQSAsyncReIndexService.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/SQSAsyncReIndexService.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/SQSAsyncReIndexService.java
deleted file mode 100644
index 60a804c..0000000
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/SQSAsyncReIndexService.java
+++ /dev/null
@@ -1,269 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.usergrid.corepersistence.index;
-
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.UUID;
-import java.util.concurrent.TimeUnit;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.usergrid.persistence.collection.serialization.impl.migration.EntityIdScope;
-import org.apache.usergrid.persistence.core.metrics.MetricsFactory;
-import org.apache.usergrid.persistence.core.scope.ApplicationScope;
-import org.apache.usergrid.persistence.model.entity.Entity;
-import org.apache.usergrid.persistence.model.util.UUIDGenerator;
-import org.apache.usergrid.persistence.queue.QueueManager;
-import org.apache.usergrid.persistence.queue.QueueManagerFactory;
-import org.apache.usergrid.persistence.queue.QueueMessage;
-import org.apache.usergrid.persistence.queue.QueueScope;
-import org.apache.usergrid.persistence.queue.impl.QueueScopeImpl;
-
-import com.codahale.metrics.Meter;
-import com.codahale.metrics.Timer;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.dataformat.smile.SmileFactory;
-import com.google.inject.Inject;
-import com.google.inject.Singleton;
-
-
-@Singleton
-public class SQSAsyncReIndexService implements AsyncReIndexService {
-
-
- private static final Logger logger = LoggerFactory.getLogger( SQSAsyncReIndexService.class );
-
- /** Hacky, copied from CPEntityManager b/c we can't access it here */
- public static final UUID MANAGEMENT_APPLICATION_ID = UUID.fromString( "b6768a08-b5d5-11e3-a495-11ddb1de66c8" );
-
-
- /**
- * Set our TTL to 1 month. This is high, but in the event of a bug, we want these entries to get removed
- */
- public static final int TTL = 60 * 60 * 24 * 30;
-
- /**
- * The name to put in the map
- */
- public static final String MAP_NAME = "esqueuedata";
-
-
- private static final String QUEUE_NAME = "es_queue";
-
- private static SmileFactory SMILE_FACTORY = new SmileFactory();
-
- static {
- SMILE_FACTORY.delegateToTextual( true );
- }
-
-
- private final QueueManager queue;
- private final QueryFig queryFig;
- private final ObjectMapper mapper;
- private final Meter readMeter;
- private final Timer readTimer;
- private final Meter writeMeter;
- private final Timer writeTimer;
-
-
- @Inject
- public SQSAsyncReIndexService( final QueueManagerFactory queueManagerFactory, final QueryFig queryFig,
- final MetricsFactory metricsFactory ) {
- final QueueScope queueScope = new QueueScopeImpl( QUEUE_NAME );
-
- this.queue = queueManagerFactory.getQueueManager( queueScope );
- this.queryFig = queryFig;
-
- this.writeTimer = metricsFactory.getTimer( SQSAsyncReIndexService.class, "write.timer" );
- this.writeMeter = metricsFactory.getMeter( SQSAsyncReIndexService.class, "write.meter" );
-
- this.readTimer = metricsFactory.getTimer( SQSAsyncReIndexService.class, "read.timer" );
- this.readMeter = metricsFactory.getMeter( SQSAsyncReIndexService.class, "read.meter" );
-
- this.mapper = new ObjectMapper( SMILE_FACTORY );
- //pretty print, disabling for speed
- // mapper.enable(SerializationFeature.INDENT_OUTPUT);
-
- }
-
-
- public void offer( final IndexEntityEvent operation ) {
- final Timer.Context timer = this.writeTimer.time();
- this.writeMeter.mark();
-
- final UUID identifier = UUIDGenerator.newTimeUUID();
-
- try {
-
- final String payLoad = toString( operation );
-
- //signal to SQS
- this.queue.sendMessage( identifier );
- }
- catch ( IOException e ) {
- throw new RuntimeException( "Unable to queue message", e );
- }
- finally {
- timer.stop();
- }
- }
-
-
- public List<IndexEntityEvent> take( final int takeSize, final long timeout, final TimeUnit timeUnit ) {
-
- //SQS doesn't support more than 10
-
- final int actualTake = Math.min( 10, takeSize );
-
- final Timer.Context timer = this.readTimer.time();
-
- try {
-
- List<QueueMessage> messages = queue
- .getMessages( actualTake, queryFig.getIndexQueueTimeout(), ( int ) timeUnit.toMillis( timeout ),
- String.class );
-
-
- final List<IndexEntityEvent> response = new ArrayList<>( messages.size() );
-
- final List<String> mapEntries = new ArrayList<>( messages.size() );
-
-
- if ( messages.size() == 0 ) {
- return Collections.emptyList();
- }
-
- //add all our keys for a single round trip
- for ( final QueueMessage message : messages ) {
- mapEntries.add( message.getBody().toString() );
- }
-
-
- //load them into our response
- for ( final QueueMessage message : messages ) {
-
- final String payload = getBody( message );
-
- //now see if the key was there
-
-
- final IndexEntityEvent messageBody;
-
- try {
- messageBody = fromString( payload );
- }
- catch ( IOException e ) {
- logger.error( "Unable to deserialize message from string. This is a bug", e );
- throw new RuntimeException( "Unable to deserialize message from string. This is a bug", e );
- }
-
- SqsIndexOperationMessage operation = new SqsIndexOperationMessage( message, messageBody );
-
- response.add( operation );
- }
-
- readMeter.mark( response.size() );
- return response;
- }
- //stop our timer
- finally {
- timer.stop();
- }
- }
-
-
- public void ack( final List<IndexEntityEvent> messages ) {
-
- //nothing to do
- if ( messages.size() == 0 ) {
- return;
- }
-
- List<QueueMessage> toAck = new ArrayList<>( messages.size() );
-
- for ( IndexEntityEvent ioe : messages ) {
-
-
- final SqsIndexOperationMessage sqsIndexOperationMessage = ( SqsIndexOperationMessage ) ioe;
-
- toAck.add( ( ( SqsIndexOperationMessage ) ioe ).getMessage() );
- }
-
- queue.commitMessages( toAck );
- }
-
-
- /** Read the object from Base64 string. */
- private IndexEntityEvent fromString( String s ) throws IOException {
- IndexEntityEvent o = mapper.readValue( s, IndexEntityEvent.class );
- return o;
- }
-
-
- /** Write the object to a Base64 string. */
- private String toString( IndexEntityEvent o ) throws IOException {
- return mapper.writeValueAsString( o );
- }
-
-
- private String getBody( final QueueMessage message ) {
- return message.getBody().toString();
- }
-
-
- @Override
- public void index( final EntityIdScope entityIdScope ) {
-
- }
-
-
- /**
- * The message that subclasses our IndexOperationMessage. holds a pointer to the original message
- */
- public class SqsIndexOperationMessage extends IndexEntityEvent {
-
- private final QueueMessage message;
-
-
- public SqsIndexOperationMessage( final QueueMessage message, final IndexEntityEvent source ) {
- super( source.getApplicationScope(), source.getEntityId(), source.getEntityVersion() );
- this.message = message;
- }
-
-
- /**
- * Get the message from our queue
- */
- public QueueMessage getMessage() {
- return message;
- }
- }
-
-
- @Override
- public void queueEntityIndexUpdate( final ApplicationScope applicationScope, final Entity entity) {
- throw new UnsupportedOperationException( "Implement index rebuild" );
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f4d0d1a9/stack/core/src/main/java/org/apache/usergrid/corepersistence/util/CpNamingUtils.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/util/CpNamingUtils.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/util/CpNamingUtils.java
index c42ad10..364b071 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/util/CpNamingUtils.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/util/CpNamingUtils.java
@@ -99,8 +99,7 @@ public class CpNamingUtils {
* Get the index scope for the edge from the source
*/
public static IndexEdge generateScopeFromSource( final Edge edge ) {
- return new IndexEdgeImpl( edge.getSourceNode(), edge.getType(), SearchEdge.NodeType.SOURCE,
- edge.getTimestamp() );
+ return new IndexEdgeImpl( edge.getSourceNode(), edge.getType(), SearchEdge.NodeType.SOURCE, edge.getTimestamp() );
}
@@ -108,8 +107,7 @@ public class CpNamingUtils {
* Get the index scope for the edge from the source
*/
public static IndexEdge generateScopeToTarget( final Edge edge ) {
- return new IndexEdgeImpl( edge.getTargetNode(), edge.getType(), SearchEdge.NodeType.TARGET,
- edge.getTimestamp() );
+ return new IndexEdgeImpl( edge.getTargetNode(), edge.getType(), SearchEdge.NodeType.TARGET, edge.getTimestamp() );
}
@@ -123,6 +121,8 @@ public class CpNamingUtils {
/**
*
+ * TODO move sourceId to ApplicationScope
+ *
* @param sourceId
* @param collectionName
* @param entityId
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f4d0d1a9/stack/core/src/main/java/org/apache/usergrid/corepersistence/util/SerializableMapper.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/util/SerializableMapper.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/util/SerializableMapper.java
index 1ad4115..19ecf6d 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/util/SerializableMapper.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/util/SerializableMapper.java
@@ -27,6 +27,7 @@ import java.io.Serializable;
import java.nio.charset.StandardCharsets;
import org.apache.usergrid.persistence.collection.serialization.SerializationFig;
+import org.apache.usergrid.persistence.collection.serialization.impl.migration.EntityIdScope;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -46,6 +47,7 @@ public class SerializableMapper {
static{
MAPPER.enableDefaultTypingAsProperty( ObjectMapper.DefaultTyping.JAVA_LANG_OBJECT, "@class" );
+ SMILE_FACTORY.delegateToTextual( true );
}
/**
@@ -66,9 +68,9 @@ public class SerializableMapper {
/**
* Write the value as a string
+ * @param <T>
* @param serialized
* @param clazz
- * @param <T>
* @return
*/
public static <T extends Serializable> T fromString(final String serialized, final Class<T> clazz){
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f4d0d1a9/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/AsyncIndexServiceTest.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/AsyncIndexServiceTest.java b/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/AsyncIndexServiceTest.java
new file mode 100644
index 0000000..faef848
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/AsyncIndexServiceTest.java
@@ -0,0 +1,203 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.usergrid.corepersistence.index;
+
+
+import java.util.UUID;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.apache.usergrid.corepersistence.TestIndexModule;
+import org.apache.usergrid.corepersistence.util.CpNamingUtils;
+import org.apache.usergrid.persistence.collection.EntityCollectionManager;
+import org.apache.usergrid.persistence.collection.EntityCollectionManagerFactory;
+import org.apache.usergrid.persistence.core.aws.NoAWSCredsRule;
+import org.apache.usergrid.persistence.core.guice.MigrationManagerRule;
+import org.apache.usergrid.persistence.core.metrics.MetricsFactory;
+import org.apache.usergrid.persistence.core.rx.RxTaskScheduler;
+import org.apache.usergrid.persistence.core.scope.ApplicationScope;
+import org.apache.usergrid.persistence.core.scope.ApplicationScopeImpl;
+import org.apache.usergrid.persistence.core.test.UseModules;
+import org.apache.usergrid.persistence.graph.Edge;
+import org.apache.usergrid.persistence.graph.GraphManager;
+import org.apache.usergrid.persistence.graph.GraphManagerFactory;
+import org.apache.usergrid.persistence.index.ApplicationEntityIndex;
+import org.apache.usergrid.persistence.index.CandidateResults;
+import org.apache.usergrid.persistence.index.EntityIndex;
+import org.apache.usergrid.persistence.index.EntityIndexFactory;
+import org.apache.usergrid.persistence.index.SearchEdge;
+import org.apache.usergrid.persistence.index.SearchTypes;
+import org.apache.usergrid.persistence.index.impl.EsRunner;
+import org.apache.usergrid.persistence.model.entity.Entity;
+import org.apache.usergrid.persistence.model.entity.Id;
+import org.apache.usergrid.persistence.model.entity.SimpleId;
+import org.apache.usergrid.persistence.model.field.StringField;
+import org.apache.usergrid.persistence.model.util.UUIDGenerator;
+import org.apache.usergrid.persistence.queue.QueueManagerFactory;
+
+import com.google.inject.Inject;
+
+import net.jcip.annotations.NotThreadSafe;
+
+import rx.Observable;
+import rx.schedulers.Schedulers;
+
+import static org.apache.usergrid.persistence.core.util.IdGenerator.createId;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+
+@RunWith( EsRunner.class )
+@UseModules( { TestIndexModule.class } )
+@NotThreadSafe
+public abstract class AsyncIndexServiceTest {
+
+
+ @Inject
+ @Rule
+ public MigrationManagerRule migrationManagerRule;
+
+
+ @Inject
+ public EntityCollectionManagerFactory entityCollectionManagerFactory;
+
+ @Inject
+ public GraphManagerFactory graphManagerFactory;
+
+
+ @Inject
+ public EntityIndexFactory entityIndexFactory;
+
+
+ private AsyncIndexService asyncIndexService;
+
+
+ /**
+ * Get the async index service
+ */
+ protected abstract AsyncIndexService getAsyncIndexService();
+
+
+ @Before
+ public void setup() {
+ asyncIndexService = getAsyncIndexService();
+ }
+
+
+ @Test( timeout = 60000 )
+ public void testMessageIndexing() throws InterruptedException {
+
+
+ ApplicationScope applicationScope =
+ new ApplicationScopeImpl( new SimpleId( UUID.randomUUID(), "application" ) );
+
+
+ final Entity testEntity = new Entity( createId( "thing" ), UUIDGenerator.newTimeUUID() );
+ testEntity.setField( new StringField( "string", "foo" ) );
+
+
+ //write the entity before indexing
+ final EntityCollectionManager collectionManager =
+ entityCollectionManagerFactory.createCollectionManager( applicationScope );
+
+ collectionManager.write( testEntity ).toBlocking().last();
+
+ final GraphManager graphManager = graphManagerFactory.createEdgeManager( applicationScope );
+
+ //create our collection edge
+ final Edge collectionEdge =
+ CpNamingUtils.createCollectionEdge( applicationScope.getApplication(), "things", testEntity.getId() );
+ graphManager.writeEdge( collectionEdge ).toBlocking().last();
+
+
+ /**
+ * Write 10k edges 10 at a time in parallel
+ */
+ final Edge connectionSearch = Observable.range( 0, 10000 ).flatMap( integer -> {
+ final Id connectingId = createId( "connecting" );
+ final Edge edge = CpNamingUtils.createConnectionEdge( connectingId, "likes", testEntity.getId() );
+
+ return graphManager.writeEdge( edge ).subscribeOn( Schedulers.io() );
+ }, 10 ).toBlocking().last();
+
+
+ asyncIndexService.queueEntityIndexUpdate( applicationScope, testEntity );
+
+
+ // Thread.sleep( 1000000000000l );
+
+ final ApplicationEntityIndex applicationEntityIndex =
+ entityIndexFactory.createApplicationEntityIndex( applicationScope );
+
+ final SearchEdge collectionSearchEdge = CpNamingUtils.createSearchEdgeFromSource( collectionEdge );
+
+
+ //query until it's available
+ final CandidateResults collectionResults = getResults( applicationEntityIndex, collectionSearchEdge,
+ SearchTypes.fromTypes( testEntity.getId().getType() ), "select *", 100, 1, 100 );
+
+ assertEquals( 1, collectionResults.size() );
+
+ assertEquals( testEntity.getId(), collectionResults.get( 0 ).getId() );
+
+
+ final SearchEdge connectionSearchEdge = CpNamingUtils.createSearchEdgeFromSource( connectionSearch );
+
+
+ //query until it's available
+ final CandidateResults connectionResults = getResults( applicationEntityIndex, connectionSearchEdge,
+ SearchTypes.fromTypes( testEntity.getId().getType() ), "select *", 100, 1, 100 );
+
+ assertEquals( 1, connectionResults.size() );
+
+ assertEquals( testEntity.getId(), connectionResults.get( 0 ).getId() );
+ }
+
+
+ private CandidateResults getResults( final ApplicationEntityIndex applicationEntityIndex,
+ final SearchEdge searchEdge, final SearchTypes searchTypes, final String ql,
+ final int count, final int expectedSize, final int attempts ) {
+
+
+ for ( int i = 0; i < attempts; i++ ) {
+ final CandidateResults candidateResults =
+ applicationEntityIndex.search( searchEdge, searchTypes, "select *", 100 );
+
+ if ( candidateResults.size() == expectedSize ) {
+ return candidateResults;
+ }
+
+ try {
+ Thread.sleep( 100 );
+ }
+ catch ( InterruptedException e ) {
+ //swallow
+ }
+ }
+
+ fail( "Could not find candidates of size " + expectedSize + "after " + attempts + " attempts" );
+
+ //we'll never reach this, required for compile
+ return null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f4d0d1a9/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/InMemoryAsycIndexServiceTest.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/InMemoryAsycIndexServiceTest.java b/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/InMemoryAsycIndexServiceTest.java
new file mode 100644
index 0000000..43968de
--- /dev/null
+++ b/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/InMemoryAsycIndexServiceTest.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.usergrid.corepersistence.index;
+
+
+import org.junit.Rule;
+import org.junit.runner.RunWith;
+
+import org.apache.usergrid.corepersistence.TestIndexModule;
+import org.apache.usergrid.persistence.core.aws.NoAWSCredsRule;
+import org.apache.usergrid.persistence.core.metrics.MetricsFactory;
+import org.apache.usergrid.persistence.core.rx.RxTaskScheduler;
+import org.apache.usergrid.persistence.core.test.UseModules;
+import org.apache.usergrid.persistence.index.impl.EsRunner;
+import org.apache.usergrid.persistence.queue.QueueManagerFactory;
+
+import com.google.inject.Inject;
+
+import net.jcip.annotations.NotThreadSafe;
+
+
+@RunWith( EsRunner.class )
+@UseModules( { TestIndexModule.class } )
+@NotThreadSafe
+public class InMemoryAsycIndexServiceTest extends AsyncIndexServiceTest {
+
+ @Rule
+ public NoAWSCredsRule noAwsCredsRule = new NoAWSCredsRule();
+
+
+ @Inject
+ public IndexService indexService;
+
+ @Inject
+ public RxTaskScheduler rxTaskScheduler;
+
+
+ @Override
+ protected AsyncIndexService getAsyncIndexService() {
+ return new InMemoryAsyncIndexService( indexService, rxTaskScheduler, entityCollectionManagerFactory );
+ }
+
+
+
+
+
+}