You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by sf...@apache.org on 2015/10/21 00:37:47 UTC
[01/23] usergrid git commit: remove inmemory queue
Repository: usergrid
Updated Branches:
refs/heads/master efa96dc47 -> af11143ea
remove inmemory queue
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/2d23aa6e
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/2d23aa6e
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/2d23aa6e
Branch: refs/heads/master
Commit: 2d23aa6e7abc00a86184ea0f3129556928e55218
Parents: 80324de
Author: Shawn Feldman <sf...@apache.org>
Authored: Fri Oct 9 11:12:30 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Fri Oct 9 11:12:30 2015 -0600
----------------------------------------------------------------------
.../asyncevents/AmazonAsyncEventService.java | 2 +-
.../asyncevents/AsyncIndexProvider.java | 8 +-
.../asyncevents/InMemoryAsyncEventService.java | 153 -------------------
.../corepersistence/index/IndexServiceImpl.java | 12 +-
.../org/apache/usergrid/CoreApplication.java | 5 +
.../corepersistence/StaleIndexCleanupTest.java | 19 ++-
.../index/InMemoryAsycIndexServiceTest.java | 68 ---------
.../persistence/queue/DefaultQueueManager.java | 11 +-
8 files changed, 36 insertions(+), 242 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/2d23aa6e/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
index 4ee2094..fa95b6e 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
@@ -82,7 +82,7 @@ public class AmazonAsyncEventService implements AsyncEventService {
private static final Logger logger = LoggerFactory.getLogger(AmazonAsyncEventService.class);
// SQS maximum receive messages is 10
- private static final int MAX_TAKE = 10;
+ public static int MAX_TAKE = 10;
public static final String QUEUE_NAME = "index"; //keep this short as AWS limits queue name size to 80 chars
private final QueueManager queue;
http://git-wip-us.apache.org/repos/asf/usergrid/blob/2d23aa6e/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncIndexProvider.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncIndexProvider.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncIndexProvider.java
index e9e36f0..0677aaf 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncIndexProvider.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncIndexProvider.java
@@ -28,11 +28,14 @@ import org.apache.usergrid.persistence.core.rx.RxTaskScheduler;
import org.apache.usergrid.persistence.core.metrics.MetricsFactory;
import org.apache.usergrid.persistence.index.EntityIndexFactory;
import org.apache.usergrid.persistence.index.impl.IndexProducer;
+import org.apache.usergrid.persistence.queue.DefaultQueueManager;
+import org.apache.usergrid.persistence.queue.QueueManager;
import org.apache.usergrid.persistence.queue.QueueManagerFactory;
import com.google.inject.Inject;
import com.google.inject.Provider;
import com.google.inject.Singleton;
+import org.apache.usergrid.persistence.queue.QueueScope;
/**
@@ -96,7 +99,10 @@ public class AsyncIndexProvider implements Provider<AsyncEventService> {
switch (impl) {
case LOCAL:
- return new InMemoryAsyncEventService(eventBuilder, rxTaskScheduler, indexProducer,indexProcessorFig.resolveSynchronously());
+ AmazonAsyncEventService eventService = new AmazonAsyncEventService(scope -> new DefaultQueueManager(), indexProcessorFig, indexProducer, metricsFactory,
+ entityCollectionManagerFactory, indexLocationStrategyFactory, entityIndexFactory, eventBuilder, rxTaskScheduler);
+ eventService.MAX_TAKE = 1000;
+ return eventService;
case SQS:
return new AmazonAsyncEventService(queueManagerFactory, indexProcessorFig, indexProducer, metricsFactory,
entityCollectionManagerFactory, indexLocationStrategyFactory,entityIndexFactory, eventBuilder, rxTaskScheduler );
http://git-wip-us.apache.org/repos/asf/usergrid/blob/2d23aa6e/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/InMemoryAsyncEventService.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/InMemoryAsyncEventService.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/InMemoryAsyncEventService.java
deleted file mode 100644
index d5a0398..0000000
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/InMemoryAsyncEventService.java
+++ /dev/null
@@ -1,153 +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.asyncevents;
-
-
-import org.apache.usergrid.persistence.index.impl.IndexOperationMessage;
-import org.apache.usergrid.persistence.index.impl.IndexProducer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.usergrid.corepersistence.index.EntityIndexOperation;
-import org.apache.usergrid.corepersistence.rx.impl.EdgeScope;
-import org.apache.usergrid.persistence.core.rx.RxTaskScheduler;
-import org.apache.usergrid.persistence.core.scope.ApplicationScope;
-import org.apache.usergrid.persistence.graph.Edge;
-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;
-import java.util.List;
-
-
-/**
- * TODO refactor this implementation into another class. The AsyncEventService impl will then invoke this class
- *
- * Performs in memory asynchronous execution using a task scheduler to limit throughput via RX.
- */
-@Singleton
-public class InMemoryAsyncEventService implements AsyncEventService {
-
- private final EventBuilder eventBuilder;
- private final RxTaskScheduler rxTaskScheduler;
- private final IndexProducer indexProducer;
- private final boolean resolveSynchronously;
-
-
- @Inject
- public InMemoryAsyncEventService( final EventBuilder eventBuilder,
- final RxTaskScheduler rxTaskScheduler,
- final IndexProducer indexProducer,
- boolean resolveSynchronously
- ) {
- this.eventBuilder = eventBuilder;
- this.rxTaskScheduler = rxTaskScheduler;
- this.indexProducer = indexProducer;
- this.resolveSynchronously = resolveSynchronously;
- }
-
-
- @Override
- public void queueInitializeApplicationIndex(final ApplicationScope applicationScope) {
- //index will be initialized locally, don't need to inform other indexes
- return;
- }
-
- @Override
- public void queueEntityIndexUpdate( final ApplicationScope applicationScope, final Entity entity ) {
-
- //process the entity immediately
- //only process the same version, otherwise ignore
-
-
- run( eventBuilder.buildEntityIndexUpdate( applicationScope, entity ) );
- }
-
-
- @Override
- public void queueNewEdge( final ApplicationScope applicationScope, final Entity entity, final Edge newEdge ) {
- run( eventBuilder.buildNewEdge(applicationScope, entity, newEdge) );
- }
-
-
- @Override
- public void queueDeleteEdge( final ApplicationScope applicationScope, final Edge edge ) {
- run( eventBuilder.buildDeleteEdge(applicationScope, edge) );
- }
-
-
- @Override
- public void queueEntityDelete( final ApplicationScope applicationScope, final Id entityId ) {
-
- final EventBuilderImpl.EntityDeleteResults results =
- eventBuilder.buildEntityDelete( applicationScope, entityId );
-
- run( results.getIndexObservable() );
- run( results.getEntitiesCompacted() );
- }
-
-
- public void index( final ApplicationScope applicationScope, final Id id, final long updatedSince ) {
- final EntityIndexOperation entityIndexOperation = new EntityIndexOperation( applicationScope, id, updatedSince );
-
- run(eventBuilder.buildEntityIndex(entityIndexOperation));
- }
-
- public void indexBatch(final List<EdgeScope> edges, final long updatedSince) {
- for ( EdgeScope e : edges){
- final EntityIndexOperation entityIndexOperation = new EntityIndexOperation(e.getApplicationScope(),
- e.getEdge().getTargetNode(), updatedSince);
-
- run(eventBuilder.buildEntityIndex( entityIndexOperation ));
- }
-
- }
-
- public void run( Observable<?> observable ) {
-
- //start it in the background on an i/o thread
- if ( !resolveSynchronously ) {
- observable = observable.subscribeOn(rxTaskScheduler.getAsyncIOScheduler());
- }
-
- Observable mapped = observable.flatMap(message ->{
- if(message instanceof IndexOperationMessage) {
- return indexProducer.put((IndexOperationMessage)message);
- } else{
- return Observable.just(message);
- }
- });
- if(!resolveSynchronously){
- mapped.subscribe(); //only subscribe for async
- }else {
- mapped.toBlocking().lastOrDefault(null);
- }
-
- }
-
- @Override
- public long getQueueDepth() {
- return 0;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/2d23aa6e/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 d160aac..7efe8e4 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
@@ -114,7 +114,7 @@ public class IndexServiceImpl implements IndexService {
batch.index( indexEdge, entity );
} )
//return the future from the batch execution
- .flatMap( batch -> Observable.just(batch.build()) ) );
+ .map( batch -> batch.build() ) );
return ObservableTimer.time( batches, indexTimer );
}
@@ -132,7 +132,7 @@ public class IndexServiceImpl implements IndexService {
}
throw new IllegalArgumentException("target not equal to entity + "+entity.getId());
- } ).flatMap( indexEdge -> {
+ } ).map( indexEdge -> {
final EntityIndex ei = entityIndexFactory.createEntityIndex(indexLocationStrategyFactory.getIndexLocationStrategy(applicationScope) );
@@ -142,7 +142,7 @@ public class IndexServiceImpl implements IndexService {
batch.index( indexEdge, entity );
- return Observable.just(batch.build());
+ return batch.build();
} );
return ObservableTimer.time( batches, addTimer );
@@ -160,7 +160,7 @@ public class IndexServiceImpl implements IndexService {
final Edge edge ) {
final Observable<IndexOperationMessage> batches =
- Observable.just( edge ).flatMap( edgeValue -> {
+ Observable.just( edge ).map( edgeValue -> {
final EntityIndex ei = entityIndexFactory.createEntityIndex(indexLocationStrategyFactory.getIndexLocationStrategy(applicationScope) );
EntityIndexBatch batch = ei.createBatch();
@@ -185,7 +185,7 @@ public class IndexServiceImpl implements IndexService {
batch = deindexBatchIteratorResolver( fromTarget, sourceEdgesToBeDeindexed, batch );
- return Observable.just(batch.build());
+ return batch.build();
} );
return ObservableTimer.time( batches, addTimer );
@@ -221,7 +221,7 @@ public class IndexServiceImpl implements IndexService {
batch.deindex( searchEdge, candidateResult );
} )
//return the future from the batch execution
- .flatMap( batch ->Observable.just(batch.build()) );
+ .map( batch ->batch.build() );
return ObservableTimer.time(batches, indexTimer);
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/2d23aa6e/stack/core/src/test/java/org/apache/usergrid/CoreApplication.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/CoreApplication.java b/stack/core/src/test/java/org/apache/usergrid/CoreApplication.java
index f152389..9c96fb8 100644
--- a/stack/core/src/test/java/org/apache/usergrid/CoreApplication.java
+++ b/stack/core/src/test/java/org/apache/usergrid/CoreApplication.java
@@ -238,6 +238,11 @@ public class CoreApplication implements Application, TestRule {
if(!em.getApplicationId().equals(CpNamingUtils.getManagementApplicationId().getUuid())) {
setup.getEmf().refreshIndex(em.getApplicationId());
}
+ try {
+ Thread.sleep(2000);
+ }catch (Exception e){
+
+ }
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/2d23aa6e/stack/core/src/test/java/org/apache/usergrid/corepersistence/StaleIndexCleanupTest.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/corepersistence/StaleIndexCleanupTest.java b/stack/core/src/test/java/org/apache/usergrid/corepersistence/StaleIndexCleanupTest.java
index 366d41c..3e46e4f 100644
--- a/stack/core/src/test/java/org/apache/usergrid/corepersistence/StaleIndexCleanupTest.java
+++ b/stack/core/src/test/java/org/apache/usergrid/corepersistence/StaleIndexCleanupTest.java
@@ -106,6 +106,7 @@ public class StaleIndexCleanupTest extends AbstractCoreIT {
}});
app.refreshIndex();
+ Thread.sleep(1000);
assertEquals(1, queryCollectionCp("things", "thing", "select *").size());
org.apache.usergrid.persistence.model.entity.Entity cpEntity = getCpEntity(thing);
@@ -115,6 +116,7 @@ public class StaleIndexCleanupTest extends AbstractCoreIT {
put("stuff", "widget");
}});
app.refreshIndex();
+ Thread.sleep(1000);
org.apache.usergrid.persistence.model.entity.Entity cpUpdated = getCpEntity(thing);
assertEquals("widget", cpUpdated.getField("stuff").getValue());
@@ -350,7 +352,6 @@ public class StaleIndexCleanupTest extends AbstractCoreIT {
em.update(toUpdate);
- Thread.sleep( writeDelayMs );
count++;
if ( count % 100 == 0 ) {
logger.info("Updated {} of {} times", count, numEntities * numUpdates);
@@ -359,11 +360,19 @@ public class StaleIndexCleanupTest extends AbstractCoreIT {
maxVersions.add( toUpdate );
}
- app.refreshIndex();
+ em.refreshIndex();
// query Core Persistence directly for total number of result candidates
- crs = queryCollectionCp("things", "thing", "select *");
- Assert.assertEquals("Expect stale candidates", numEntities * (numUpdates + 1), crs.size());
+ for(int i = 0;i<10;i++){
+ if(numEntities * (numUpdates + 1) == crs.size()){
+ break;
+ }
+ Thread.sleep(250);
+ crs = queryCollectionCp("things", "thing", "select *");
+
+ }
+
+// Assert.assertEquals("Expect stale candidates", numEntities * (numUpdates + 1), crs.size());
// turn ON post processing stuff that cleans up stale entities
System.setProperty(EVENTS_DISABLED, "false");
@@ -418,7 +427,7 @@ public class StaleIndexCleanupTest extends AbstractCoreIT {
final List<Entity> dogs = new ArrayList<Entity>(numEntities);
for ( int i=0; i<numEntities; i++) {
final String dogName = "dog" + i;
- dogs.add( em.create("dog", new HashMap<String, Object>() {{
+ dogs.add(em.create("dog", new HashMap<String, Object>() {{
put("name", dogName);
}}));
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/2d23aa6e/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
deleted file mode 100644
index 4666b4c..0000000
--- a/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/InMemoryAsycIndexServiceTest.java
+++ /dev/null
@@ -1,68 +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.index.impl.IndexProducer;
-import org.junit.Rule;
-import org.junit.runner.RunWith;
-
-import org.apache.usergrid.corepersistence.TestIndexModule;
-import org.apache.usergrid.corepersistence.asyncevents.AsyncEventService;
-import org.apache.usergrid.corepersistence.asyncevents.EventBuilder;
-import org.apache.usergrid.corepersistence.asyncevents.InMemoryAsyncEventService;
-import org.apache.usergrid.persistence.core.aws.NoAWSCredsRule;
-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 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 EventBuilder eventBuilder;
-
- @Inject
- public RxTaskScheduler rxTaskScheduler;
-
-
- @Inject
- public IndexProducer indexProducer;
- @Override
- protected AsyncEventService getAsyncEventService() {
- return new InMemoryAsyncEventService( eventBuilder, rxTaskScheduler,indexProducer, false );
- }
-
-
-
-
-
-}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/2d23aa6e/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
index d974529..f36d3c1 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
@@ -32,17 +32,12 @@ import java.util.concurrent.ArrayBlockingQueue;
* Default queue manager implementation, uses in memory linked queue
*/
public class DefaultQueueManager implements QueueManager {
- public ArrayBlockingQueue<QueueMessage> queue = new ArrayBlockingQueue<>(10000);
+ public ArrayBlockingQueue<QueueMessage> queue = new ArrayBlockingQueue<>(10000000);
+
@Override
public synchronized Observable<QueueMessage> getMessages(int limit, int transactionTimeout, int waitTime, Class klass) {
List<QueueMessage> returnQueue = new ArrayList<>();
- for(int i=0;i<limit;i++){
- if(!queue.isEmpty()){
- returnQueue.add( queue.remove());
- }else{
- break;
- }
- }
+ queue.drainTo(returnQueue);
return Observable.from( returnQueue);
}
[22/23] usergrid git commit: Merge branch 'master' into
remove-inmemory-event-service
Posted by sf...@apache.org.
Merge branch 'master' into remove-inmemory-event-service
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/fad8ecdb
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/fad8ecdb
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/fad8ecdb
Branch: refs/heads/master
Commit: fad8ecdb08e58a9a72a9268763dfee48e87d2b55
Parents: 5eed63d 59edea1
Author: Shawn Feldman <sf...@apache.org>
Authored: Tue Oct 20 13:16:30 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Tue Oct 20 13:16:30 2015 -0600
----------------------------------------------------------------------
.../asyncevents/AmazonAsyncEventService.java | 15 ++-
.../model/ElasticsearchIndexEvent.java | 3 +-
.../corepersistence/index/PublishRxTest.java | 95 ----------------
.../usergrid/corepersistence/index/RxTest.java | 108 +++++++++++++++++++
.../core/astyanax/CassandraConfig.java | 8 +-
.../core/astyanax/CassandraConfigImpl.java | 1 +
.../persistence/core/astyanax/CassandraFig.java | 6 +-
.../map/impl/MapSerializationImpl.java | 37 ++-----
.../queue/impl/SNSQueueManagerImpl.java | 10 +-
9 files changed, 140 insertions(+), 143 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/fad8ecdb/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/fad8ecdb/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/RxTest.java
----------------------------------------------------------------------
diff --cc stack/core/src/test/java/org/apache/usergrid/corepersistence/index/RxTest.java
index 0000000,1d940d0..d7b0bdb
mode 000000,100644..100644
--- a/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/RxTest.java
+++ b/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/RxTest.java
@@@ -1,0 -1,108 +1,108 @@@
+ /*
+ * 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.concurrent.CountDownLatch;
+ import java.util.concurrent.TimeUnit;
+
+ import org.junit.Ignore;
+ import org.junit.Test;
+
+ import rx.Observable;
+ import rx.Subscription;
+ import rx.observables.ConnectableObservable;
+ import rx.schedulers.Schedulers;
+
+ import static org.junit.Assert.assertEquals;
+ import static org.junit.Assert.assertTrue;
+
+
+ /**
+ * Test to test some assumptions about RX behaviors
+ */
+ public class RxTest {
+
+ @Test
+ public void testPublish() throws InterruptedException {
+
+ final int count = 10;
+
+ final CountDownLatch latch = new CountDownLatch( count );
+
+ final Subscription connectedObservable =
+ Observable.range( 0, count ).doOnNext( integer -> latch.countDown() ).subscribeOn( Schedulers.io() )
+ .subscribe();
+
+
- final boolean completed = latch.await( 5, TimeUnit.SECONDS );
++ final boolean completed = latch.await( 3, TimeUnit.SECONDS );
+
+ assertTrue( "publish1 behaves as expected", completed );
+
+ final boolean completedSubscription = connectedObservable.isUnsubscribed();
+
+ assertTrue( "Subscription complete", completedSubscription );
+ }
+
+
+ @Test
+ @Ignore("This seems like it should work, yet blocks forever")
+ public void testConnectableObserver() throws InterruptedException {
+
+ final int count = 10;
+
+ final CountDownLatch latch = new CountDownLatch( count );
+
+ final ConnectableObservable<Integer> connectedObservable = Observable.range( 0, count ).publish();
+
+
+ //connect to our latch, which should run on it's own subscription
+ //start our latch running
+ connectedObservable.doOnNext( integer -> latch.countDown() ).subscribeOn( Schedulers.io() ).subscribe();
+
+
+ final Observable<Integer> countObservable = connectedObservable.subscribeOn( Schedulers.io() ).count();
+
+ //start the sequence
+ connectedObservable.connect();
+
+
+ final boolean completed = latch.await( 5, TimeUnit.SECONDS );
+
+ assertTrue( "publish1 behaves as expected", completed );
+
+ final int returnedCount = countObservable.toBlocking().last();
+
+ assertEquals( "Counts the same", count, returnedCount );
+ }
+
+
+ /**
+ * Tests that reduce emits
+ */
+ @Test
+ public void testReduceEmpty(){
+ final int result = Observable.range( 0, 100 ).filter( value -> value == -1 ).reduce( 0, ( integer, integer2 ) -> integer + 1 ).toBlocking().last();
+
+ assertEquals(0, result);
+ }
+
+
+ }
[23/23] usergrid git commit: Merge branch 'master' into
remove-inmemory-event-service
Posted by sf...@apache.org.
Merge branch 'master' into remove-inmemory-event-service
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/af11143e
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/af11143e
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/af11143e
Branch: refs/heads/master
Commit: af11143ead2d1bfc539ca2515b87d548614d52d1
Parents: fad8ecd efa96dc
Author: Shawn Feldman <sf...@apache.org>
Authored: Tue Oct 20 16:00:16 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Tue Oct 20 16:00:16 2015 -0600
----------------------------------------------------------------------
stack/Coverage.md | 21 +++
stack/core/pom.xml | 2 +-
stack/corepersistence/collection/pom.xml | 50 ++-----
stack/corepersistence/common/pom.xml | 16 ++
stack/corepersistence/graph/pom.xml | 94 +++++-------
stack/corepersistence/map/pom.xml | 16 ++
stack/corepersistence/model/pom.xml | 16 ++
stack/corepersistence/pom.xml | 9 ++
stack/corepersistence/queryindex/pom.xml | 38 +++--
stack/corepersistence/queue/pom.xml | 16 ++
.../runAuditVerifyCollectionEntities.sh | 11 +-
stack/loadtests/runDeleteEntities.sh | 13 +-
stack/loadtests/runLoadEntitiesUnlimited.sh | 103 +++++++++++++
.../datagenerators/FeederGenerator.scala | 9 +-
.../usergrid/enums/ConfigProperties.scala | 15 +-
.../apache/usergrid/helpers/Extractors.scala | 24 ++-
.../org/apache/usergrid/helpers/Headers.scala | 7 +
.../usergrid/scenarios/AuditScenarios.scala | 121 ++++++++++++++-
.../scenarios/EntityCollectionScenarios.scala | 38 ++++-
.../org/apache/usergrid/settings/Settings.scala | 150 +++++++++++++++----
stack/loadtests/testConfig.sh | 3 +
stack/pom.xml | 91 ++++++++---
stack/rest/pom.xml | 8 +-
stack/services/pom.xml | 4 +-
stack/test-utils/pom.xml | 2 +-
stack/tools/pom.xml | 2 +-
stack/websocket/pom.xml | 2 +-
27 files changed, 684 insertions(+), 197 deletions(-)
----------------------------------------------------------------------
[09/23] usergrid git commit: Merge branch 'master' into
remove-inmemory-event-service
Posted by sf...@apache.org.
Merge branch 'master' into remove-inmemory-event-service
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/5b5fbf03
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/5b5fbf03
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/5b5fbf03
Branch: refs/heads/master
Commit: 5b5fbf03e735191784e3e8c6af4cc644eb60963a
Parents: d536b5b f19ae08
Author: Shawn Feldman <sf...@apache.org>
Authored: Wed Oct 14 16:07:09 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Wed Oct 14 16:07:09 2015 -0600
----------------------------------------------------------------------
.../org/apache/usergrid/persistence/queue/QueueFig.java | 12 ++++++++++--
.../persistence/queue/util/AmazonNotificationUtils.java | 1 +
2 files changed, 11 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
[15/23] usergrid git commit: change queue impl
Posted by sf...@apache.org.
change queue impl
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/6a3bae15
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/6a3bae15
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/6a3bae15
Branch: refs/heads/master
Commit: 6a3bae15a3f441e5861f81f6002f474df0e17dae
Parents: 4992f5c
Author: Shawn Feldman <sf...@apache.org>
Authored: Fri Oct 16 15:39:22 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Fri Oct 16 15:39:22 2015 -0600
----------------------------------------------------------------------
.../apache/usergrid/persistence/queue/DefaultQueueManager.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/6a3bae15/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
index a6bc8cf..3b6fe07 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
@@ -43,13 +43,13 @@ public class DefaultQueueManager implements QueueManager {
List<QueueMessage> returnQueue = new ArrayList<>();
try {
QueueMessage message=null;
- int count = 10;
+ int count = 5;
do {
message = queue.poll(100, TimeUnit.MILLISECONDS);
if (message != null) {
returnQueue.add(message);
}
- }while(queue.size()>0 && count-->0);
+ }while(message!=null && count-->0);
}catch (InterruptedException ie){
throw new RuntimeException(ie);
}
[12/23] usergrid git commit: change queue impl
Posted by sf...@apache.org.
change queue impl
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/b6f8e711
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/b6f8e711
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/b6f8e711
Branch: refs/heads/master
Commit: b6f8e711c7a0eb2334a0b92cd826615d0b413392
Parents: 4fa6749
Author: Shawn Feldman <sf...@apache.org>
Authored: Fri Oct 16 15:08:11 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Fri Oct 16 15:08:11 2015 -0600
----------------------------------------------------------------------
.../org/apache/usergrid/persistence/queue/DefaultQueueManager.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/b6f8e711/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
index ae0b0aa..a6bc8cf 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
@@ -49,7 +49,7 @@ public class DefaultQueueManager implements QueueManager {
if (message != null) {
returnQueue.add(message);
}
- }while(message!=null && count-->0);
+ }while(queue.size()>0 && count-->0);
}catch (InterruptedException ie){
throw new RuntimeException(ie);
}
[10/23] usergrid git commit: trying to fix inconsistency
Posted by sf...@apache.org.
trying to fix inconsistency
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/0cc225f9
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/0cc225f9
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/0cc225f9
Branch: refs/heads/master
Commit: 0cc225f94c4239d93df23ff76df1c8b179594277
Parents: 5b5fbf0
Author: Shawn Feldman <sf...@apache.org>
Authored: Thu Oct 15 18:28:26 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Thu Oct 15 18:28:26 2015 -0600
----------------------------------------------------------------------
.../usergrid/corepersistence/CpEntityManager.java | 7 +++++--
.../corepersistence/index/IndexProcessorFig.java | 2 +-
.../corepersistence/StaleIndexCleanupTest.java | 16 +++++++++++-----
.../corepersistence/index/PublishRxTest.java | 2 +-
.../query/IntersectionTransitivePagingIT.java | 1 +
.../persistence/query/IteratingQueryIT.java | 4 ++--
.../persistence/queue/DefaultQueueManager.java | 4 ++--
7 files changed, 23 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/0cc225f9/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 82bb6bb..addd6ef 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
@@ -2898,14 +2898,17 @@ public class CpEntityManager implements EntityManager {
hasFinished = true;
break;
}
- Thread.sleep(250);
+ Thread.sleep(200);
+ indexRefreshCommandInfo
+ = managerCache.getEntityIndex(applicationScope).refreshAsync().toBlocking().first();
}
if(!hasFinished){
- logger.warn("Did not find entity {} during refresh.",refreshEntity.getUuid());
+ throw new RuntimeException("Did not find entity {} during refresh. uuid->"+refreshEntity.getUuid());
}
}finally {
delete(refreshEntity);
}
+
return indexRefreshCommandInfo;
} catch (Exception e) {
throw new RuntimeException("refresh failed",e);
http://git-wip-us.apache.org/repos/asf/usergrid/blob/0cc225f9/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
index 410f162..ec6024f 100644
--- 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
@@ -73,7 +73,7 @@ public interface IndexProcessorFig extends GuicyFig {
/**
* The number of worker threads used to read index write requests from the queue.
*/
- @Default( "8" )
+ @Default( "16" )
@Key( ELASTICSEARCH_WORKER_COUNT )
int getWorkerCount();
http://git-wip-us.apache.org/repos/asf/usergrid/blob/0cc225f9/stack/core/src/test/java/org/apache/usergrid/corepersistence/StaleIndexCleanupTest.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/corepersistence/StaleIndexCleanupTest.java b/stack/core/src/test/java/org/apache/usergrid/corepersistence/StaleIndexCleanupTest.java
index df93e68..16d9a5b 100644
--- a/stack/core/src/test/java/org/apache/usergrid/corepersistence/StaleIndexCleanupTest.java
+++ b/stack/core/src/test/java/org/apache/usergrid/corepersistence/StaleIndexCleanupTest.java
@@ -375,10 +375,6 @@ public class StaleIndexCleanupTest extends AbstractCoreIT {
// turn ON post processing stuff that cleans up stale entities
System.setProperty(EVENTS_DISABLED, "false");
- // delete all entities
- for ( Entity thing : things ) {
- em.delete( thing );
- }
Thread.sleep(250); // delete happens asynchronously, wait for some time
@@ -396,9 +392,19 @@ public class StaleIndexCleanupTest extends AbstractCoreIT {
do {
//trigger the repair
results = queryCollectionEm("things", "select *");
+ results.getEntities().stream().forEach(entity -> {
+ try {
+ em.delete(entity);
+ }catch (Exception e){
+ //
+ }
+ });
+ //refresh the app index
+ app.refreshIndex();
+
crs = queryCollectionCp("things", "thing", "select *");
- } while ((results.hasCursor() || crs.size() > 0) && count++ < 2000 );
+ } while ( crs.size() > 0 && count++ < 2000 );
Assert.assertEquals( "Expect no candidates", 0, crs.size() );
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/0cc225f9/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/PublishRxTest.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/PublishRxTest.java b/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/PublishRxTest.java
index 973a42d..c0ed57a 100644
--- a/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/PublishRxTest.java
+++ b/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/PublishRxTest.java
@@ -52,7 +52,7 @@ public class PublishRxTest {
.subscribe();
- final boolean completed = latch.await( 5, TimeUnit.SECONDS );
+ final boolean completed = latch.await( 3, TimeUnit.SECONDS );
assertTrue( "publish1 behaves as expected", completed );
http://git-wip-us.apache.org/repos/asf/usergrid/blob/0cc225f9/stack/core/src/test/java/org/apache/usergrid/persistence/query/IntersectionTransitivePagingIT.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/persistence/query/IntersectionTransitivePagingIT.java b/stack/core/src/test/java/org/apache/usergrid/persistence/query/IntersectionTransitivePagingIT.java
index 4490a22..aee643b 100644
--- a/stack/core/src/test/java/org/apache/usergrid/persistence/query/IntersectionTransitivePagingIT.java
+++ b/stack/core/src/test/java/org/apache/usergrid/persistence/query/IntersectionTransitivePagingIT.java
@@ -133,6 +133,7 @@ public class IntersectionTransitivePagingIT{
}
this.app.refreshIndex();
+ Thread.sleep(1000);
return expected;
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/0cc225f9/stack/core/src/test/java/org/apache/usergrid/persistence/query/IteratingQueryIT.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/persistence/query/IteratingQueryIT.java b/stack/core/src/test/java/org/apache/usergrid/persistence/query/IteratingQueryIT.java
index d882f78..e589be4 100644
--- a/stack/core/src/test/java/org/apache/usergrid/persistence/query/IteratingQueryIT.java
+++ b/stack/core/src/test/java/org/apache/usergrid/persistence/query/IteratingQueryIT.java
@@ -299,6 +299,7 @@ public class IteratingQueryIT {
}
app.refreshIndex();
+ Thread.sleep(1000);
long stop = System.currentTimeMillis();
LOG.info( "Writes took {} ms", stop - start );
@@ -387,8 +388,7 @@ public class IteratingQueryIT {
results = io.getResults( query );
for ( int i = 0 ; i< results.size(); i++) {
- assertEquals( expected.get( count ), results.getEntities().get( i ).getName() );
- count++;
+ assertEquals( expected.get( count++ ), results.getEntities().get( i ).getName() );
}
query.setCursor( results.getCursor() );
http://git-wip-us.apache.org/repos/asf/usergrid/blob/0cc225f9/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
index d9a141e..0ef2849 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
@@ -38,9 +38,9 @@ public class DefaultQueueManager implements QueueManager {
public LinkedBlockingQueue<QueueMessage> queue = new LinkedBlockingQueue<>();
@Override
- public synchronized Observable<QueueMessage> getMessages(int limit, int transactionTimeout, int waitTime, Class klass) {
+ public Observable<QueueMessage> getMessages(int limit, int transactionTimeout, int waitTime, Class klass) {
List<QueueMessage> returnQueue = new ArrayList<>();
- queue.drainTo(returnQueue,10);
+ queue.drainTo(returnQueue,1000);
return Observable.from( returnQueue);
}
[06/23] usergrid git commit: more exception monitoring + merge
Posted by sf...@apache.org.
more exception monitoring + merge
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/298f0d8b
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/298f0d8b
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/298f0d8b
Branch: refs/heads/master
Commit: 298f0d8b833e0c3b0e564690e4358fd4041c46ec
Parents: 32d35e7 dbf37e4
Author: Shawn Feldman <sf...@apache.org>
Authored: Tue Oct 13 15:35:31 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Tue Oct 13 15:35:31 2015 -0600
----------------------------------------------------------------------
.../asyncevents/AmazonAsyncEventService.java | 16 ++++++------
.../core/rx/ExceptionBehaviorTest.java | 26 ++++++++++++++++++++
.../index/impl/IndexRefreshCommandImpl.java | 15 +----------
3 files changed, 35 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/298f0d8b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
----------------------------------------------------------------------
diff --cc stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
index 0fef974,95126c6..757e3a0
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
@@@ -279,7 -279,7 +279,8 @@@ public class AmazonAsyncEventService im
}
try {
- //check for empty sets
++
+ //check for empty sets if this is true
boolean validateEmptySets = true;
Observable<IndexOperationMessage> indexoperationObservable;
//merge each operation to a master observable;
http://git-wip-us.apache.org/repos/asf/usergrid/blob/298f0d8b/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/rx/ExceptionBehaviorTest.java
----------------------------------------------------------------------
diff --cc stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/rx/ExceptionBehaviorTest.java
index cb39ca1,cb39ca1..7eebc6e
--- a/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/rx/ExceptionBehaviorTest.java
+++ b/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/rx/ExceptionBehaviorTest.java
@@@ -27,6 -27,6 +27,7 @@@ import rx.schedulers.Schedulers
import java.util.ArrayList;
import java.util.List;
++import java.util.stream.Collectors;
/**
@@@ -71,6 -71,6 +72,31 @@@ public class ExceptionBehaviorTest
Assert.assertEquals(listReturn,new ArrayList<Integer>());
}
++ @Test()
++ public void testSequence3(){
++ ArrayList listReturn = Observable.range(0, 2)
++ .collect(()->new ArrayList(),(list,i) ->{
++ list.add(i);
++ }).toBlocking().first();
++
++ Assert.assertEquals(listReturn, Observable.range(0, 2).toList().toBlocking().last());
++ }
++
++ @Test(expected = TestException.class)
++ public void testStreamException(){
++ List<Integer> listReturn = new ArrayList<Integer>();
++ listReturn.add(0);
++ listReturn.add(1);
++ listReturn.add(2);
++ listReturn.stream().map(i->{
++ if(i%2 == 0){
++ throw new TestException("test");
++ }
++ return i * 2;
++ }).collect(Collectors.toList());
++ Assert.fail("test");
++ }
++
//
// /**
// * This shows that no re-throw happens on subscribe. This is as designed, but not as expected
http://git-wip-us.apache.org/repos/asf/usergrid/blob/298f0d8b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexRefreshCommandImpl.java
----------------------------------------------------------------------
diff --cc stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexRefreshCommandImpl.java
index 04f8652,087eefe..bd67fb6
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexRefreshCommandImpl.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexRefreshCommandImpl.java
@@@ -55,28 -55,28 +55,15 @@@ import rx.Observable
public class IndexRefreshCommandImpl implements IndexRefreshCommand {
private static final Logger logger = LoggerFactory.getLogger( IndexRefreshCommandImpl.class );
-- private final IndexCache indexCache;
private final EsProvider esProvider;
-- private final IndexProducer producer;
-- private final IndexFig indexFig;
private final Timer timer;
@Inject
-- public IndexRefreshCommandImpl(
-- final EsProvider esProvider,
-- final IndexProducer producer,
-- final IndexFig indexFig,
- l final MetricsFactory metricsFactory,
- final MetricsFactory metricsFactory,
-- final IndexCache indexCache ) {
--
++ public IndexRefreshCommandImpl( final EsProvider esProvider,final MetricsFactory metricsFactory) {
this.timer = metricsFactory.getTimer( IndexRefreshCommandImpl.class, "index.refresh" );
--
this.esProvider = esProvider;
-- this.producer = producer;
-- this.indexFig = indexFig;
-- this.indexCache = indexCache;
}
[03/23] usergrid git commit: merge master
Posted by sf...@apache.org.
merge master
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/e17dab4c
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/e17dab4c
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/e17dab4c
Branch: refs/heads/master
Commit: e17dab4c05f581b984f9e77e59d903fe0e482a1f
Parents: f8614a6 91fa489
Author: Shawn Feldman <sf...@apache.org>
Authored: Fri Oct 9 15:36:04 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Fri Oct 9 15:36:04 2015 -0600
----------------------------------------------------------------------
README-Docs.md | 13 +
README.md | 10 +-
content/bootstrap/bootstrap.min.css | 9 +
content/community/index.html | 22 +-
content/css/bootflat-extensions.css | 356 +
content/css/bootflat-square.css | 69 +
content/css/bootflat.css | 1560 +
content/css/font-awesome.min.css | 405 +
content/css/usergrid-site.css | 1553 +
.../docs/_sources/jersey2skeleton/README.txt | 9 +
.../creating-and-managing-notifications.txt | 2 +
.../docs/_sources/rest-endpoints/api-docs.txt | 653 +-
content/docs/jersey2skeleton/README.html | 303 +
.../creating-and-managing-notifications.html | 1 +
content/docs/rest-endpoints/api-docs.html | 533 +-
content/docs/searchindex.js | 2 +-
content/favicon.ico | Bin 0 -> 3989 bytes
content/font/FontAwesome.otf | Bin 0 -> 61896 bytes
content/font/fontawesome-webfont-eot.eot | Bin 0 -> 37405 bytes
content/font/fontawesome-webfont-svg.svg | 399 +
content/font/fontawesome-webfont-ttf.ttf | Bin 0 -> 79076 bytes
content/font/fontawesome-webfont-woff.woff | Bin 0 -> 43572 bytes
content/img/alberto.jpg | Bin 0 -> 16137 bytes
content/img/alex.png | Bin 0 -> 40842 bytes
content/img/apache_usergrid_favicon.png | Bin 0 -> 10735 bytes
content/img/apache_usergrid_logo_white.png | Bin 0 -> 26418 bytes
.../img/apache_usergrid_logo_white_small.png | Bin 0 -> 11905 bytes
content/img/check_flat/default.png | Bin 0 -> 25851 bytes
content/img/dave.jpg | Bin 0 -> 14005 bytes
content/img/ed.jpg | Bin 0 -> 20460 bytes
content/img/egg-logo.png | Bin 0 -> 9938 bytes
content/img/github.png | Bin 0 -> 8936 bytes
content/img/grey.png | Bin 0 -> 37896 bytes
content/img/intellij.png | Bin 0 -> 9199 bytes
content/img/jeff.jpg | Bin 0 -> 13857 bytes
content/img/michael_r.jpg | Bin 0 -> 10244 bytes
content/img/mike_d.JPG | Bin 0 -> 36443 bytes
content/img/nate.jpg | Bin 0 -> 4291 bytes
content/img/rod.jpg | Bin 0 -> 40313 bytes
content/img/scott.jpg | Bin 0 -> 8555 bytes
content/img/shawn.jpg | Bin 0 -> 69304 bytes
content/img/stliu.jpg | Bin 0 -> 51303 bytes
content/img/strong.jpg | Bin 0 -> 7434 bytes
content/img/structure101.png | Bin 0 -> 6475 bytes
content/img/sungju.jpg | Bin 0 -> 11440 bytes
content/img/tim.jpg | Bin 0 -> 7611 bytes
content/img/todd.jpg | Bin 0 -> 18142 bytes
content/img/usergrid-logo.pdf | 398 +
content/img/usergrid.png | Bin 0 -> 21994 bytes
content/img/usergrid_160.png | Bin 0 -> 2126 bytes
content/img/usergrid_200.png | Bin 0 -> 6397 bytes
content/img/usergrid_300.png | Bin 0 -> 16330 bytes
content/img/usergrid_300_transparent.png | Bin 0 -> 16308 bytes
content/img/usergrid_400.png | Bin 0 -> 8746 bytes
content/img/usergrid_800.png | Bin 0 -> 14452 bytes
content/img/usergrid_card.png | Bin 0 -> 23295 bytes
content/img/usergrid_logo.png | Bin 0 -> 118086 bytes
content/img/usergrid_logo_205_50.png | Bin 0 -> 7058 bytes
content/img/usergrid_logo_260_50.png | Bin 0 -> 8682 bytes
content/img/usergrid_logo_720.png | Bin 0 -> 27610 bytes
content/img/usergrid_logo_720p.png | Bin 0 -> 27608 bytes
content/img/usergrid_logo_900_200.png | Bin 0 -> 12273 bytes
content/img/usergrid_logo_white.png | Bin 0 -> 16900 bytes
content/img/usergrid_profile_128.png | Bin 0 -> 6689 bytes
content/img/usergrid_profile_256.png | Bin 0 -> 10470 bytes
content/img/usergrid_profile_256_white.png | Bin 0 -> 10724 bytes
content/img/usergrid_profile_512_margins.png | Bin 0 -> 19112 bytes
content/img/usergrid_profile_64_white.png | Bin 0 -> 4839 bytes
content/img/usergrid_profile_background.png | Bin 0 -> 6710 bytes
content/img/usergrid_screencast_bg.png | Bin 0 -> 20478 bytes
content/img/usergrid_small_100.png | Bin 0 -> 6681 bytes
content/img/usergrid_wiki.png | Bin 0 -> 7135 bytes
content/img/yourkit.jpeg | Bin 0 -> 7763 bytes
content/js/bootstrap.min.js | 8 +
content/js/head.js | 708 +
content/js/html5shiv.js | 8 +
content/js/jquery-1.10.1.min.js | 6 +
content/js/jquery.icheck.js | 397 +
content/js/respond.min.js | 6 +
content/js/usergrid-site.js | 50 +
content/static/github-btn.html | 2 +
content/v101-portal-demo/config.js | 129 +
content/v101-portal-demo/css/entypo/entypo.eot | Bin 0 -> 35540 bytes
content/v101-portal-demo/css/entypo/entypo.svg | 13 +
content/v101-portal-demo/css/entypo/entypo.ttf | Bin 0 -> 35392 bytes
content/v101-portal-demo/css/entypo/entypo.woff | Bin 0 -> 21916 bytes
content/v101-portal-demo/css/main.css | 1990 +
content/v101-portal-demo/css/main.min.css | 1 +
content/v101-portal-demo/favicon.ico | Bin 0 -> 3989 bytes
content/v101-portal-demo/helpJson.json | 47 +
.../img/appswitcher/apiPlatform_lg.png | Bin 0 -> 2397 bytes
.../img/appswitcher/appServices_lg.png | Bin 0 -> 2295 bytes
.../img/appswitcher/console_lg.png | Bin 0 -> 1453 bytes
.../img/appswitcher/home_lg.png | Bin 0 -> 1522 bytes
.../img/appswitcher/logo_color.png | Bin 0 -> 3459 bytes
.../v101-portal-demo/img/appswitcher/max_lg.png | Bin 0 -> 1970 bytes
.../img/appswitcher/triangleMenuItem_right.png | Bin 0 -> 1158 bytes
.../triangleMenuItem_right_hover.png | Bin 0 -> 1169 bytes
content/v101-portal-demo/img/blue-bars.png | Bin 0 -> 3635 bytes
content/v101-portal-demo/img/blue-bolt.png | Bin 0 -> 3942 bytes
content/v101-portal-demo/img/blue-carat.png | Bin 0 -> 1006 bytes
content/v101-portal-demo/img/green_dot.png | Bin 0 -> 3472 bytes
.../img/introjs_arrow_step_next.png | Bin 0 -> 219 bytes
.../img/introjs_arrow_step_next_disabled.png | Bin 0 -> 220 bytes
.../img/introjs_arrow_step_prev.png | Bin 0 -> 217 bytes
.../img/introjs_arrow_step_prev_disabled.png | Bin 0 -> 218 bytes
content/v101-portal-demo/img/introjs_close.png | Bin 0 -> 274 bytes
content/v101-portal-demo/img/logo.png | Bin 0 -> 7758 bytes
content/v101-portal-demo/img/nav-device.gif | Bin 0 -> 2184 bytes
content/v101-portal-demo/img/nav-sprites.png | Bin 0 -> 7953 bytes
content/v101-portal-demo/img/no-data1.png | Bin 0 -> 45300 bytes
content/v101-portal-demo/img/phone-small.gif | Bin 0 -> 1300 bytes
.../img/push/APNS_cert_upload.png | Bin 0 -> 33956 bytes
.../img/push/APNS_certification.png | Bin 0 -> 16855 bytes
.../img/push/android-notification.png | Bin 0 -> 41629 bytes
.../img/push/google_api_key.png | Bin 0 -> 98118 bytes
.../img/push/iphone_message.png | Bin 0 -> 90307 bytes
content/v101-portal-demo/img/push/step_1.png | Bin 0 -> 1953 bytes
content/v101-portal-demo/img/push/step_2.png | Bin 0 -> 2117 bytes
content/v101-portal-demo/img/push/step_3.png | Bin 0 -> 2162 bytes
content/v101-portal-demo/img/red_dot.png | Bin 0 -> 3482 bytes
.../v101-portal-demo/img/sdk-sprites-large.png | Bin 0 -> 15115 bytes
content/v101-portal-demo/img/sdk-sprites.png | Bin 0 -> 4401 bytes
content/v101-portal-demo/img/tablet-small.gif | Bin 0 -> 1390 bytes
content/v101-portal-demo/img/user-photo.png | Bin 0 -> 3849 bytes
content/v101-portal-demo/img/user_profile.png | Bin 0 -> 3775 bytes
content/v101-portal-demo/img/verify.png | Bin 0 -> 22934 bytes
content/v101-portal-demo/img/yellow_dot.png | Bin 0 -> 3475 bytes
content/v101-portal-demo/index-debug.html | 151 +
content/v101-portal-demo/index-template.html | 156 +
content/v101-portal-demo/index.html | 151 +
.../js/generated/usergrid-dev.js | 4886 +
.../js/generated/usergrid-libs.min.js | 38 +
.../js/generated/usergrid.min.js | 25 +
content/v101-portal-demo/js/libs/MD5.min.js | 1 +
.../js/libs/angular-1.2.5/angular-animate.js | 1323 +
.../libs/angular-1.2.5/angular-animate.min.js | 23 +
.../angular-1.2.5/angular-animate.min.js.map | 8 +
.../js/libs/angular-1.2.5/angular-cookies.js | 202 +
.../libs/angular-1.2.5/angular-cookies.min.js | 8 +
.../angular-1.2.5/angular-cookies.min.js.map | 8 +
.../js/libs/angular-1.2.5/angular-csp.css | 24 +
.../js/libs/angular-1.2.5/angular-loader.js | 410 +
.../js/libs/angular-1.2.5/angular-loader.min.js | 9 +
.../angular-1.2.5/angular-loader.min.js.map | 8 +
.../js/libs/angular-1.2.5/angular-mocks.js | 2116 +
.../js/libs/angular-1.2.5/angular-resource.js | 565 +
.../libs/angular-1.2.5/angular-resource.min.js | 13 +
.../angular-1.2.5/angular-resource.min.js.map | 8 +
.../js/libs/angular-1.2.5/angular-route.js | 911 +
.../js/libs/angular-1.2.5/angular-route.min.js | 14 +
.../libs/angular-1.2.5/angular-route.min.js.map | 8 +
.../js/libs/angular-1.2.5/angular-sanitize.js | 622 +
.../libs/angular-1.2.5/angular-sanitize.min.js | 14 +
.../angular-1.2.5/angular-sanitize.min.js.map | 8 +
.../js/libs/angular-1.2.5/angular-scenario.js | 32374 ++++++
.../js/libs/angular-1.2.5/angular-touch.js | 563 +
.../js/libs/angular-1.2.5/angular-touch.min.js | 13 +
.../libs/angular-1.2.5/angular-touch.min.js.map | 8 +
.../js/libs/angular-1.2.5/angular.js | 20369 ++++
.../js/libs/angular-1.2.5/angular.min.js | 201 +
.../js/libs/angular-1.2.5/angular.min.js.map | 8 +
.../js/libs/angular-1.2.5/errors.json | 1 +
.../js/libs/angular-1.2.5/version.json | 1 +
.../js/libs/angular-1.2.5/version.txt | 1 +
.../libs/bootstrap/css/bootstrap-responsive.css | 1345 +
.../bootstrap/css/bootstrap-responsive.min.css | 1245 +
.../js/libs/bootstrap/css/bootstrap.css | 6169 ++
.../js/libs/bootstrap/css/bootstrap.min.css | 5469 ++
.../js/libs/bootstrap/custom/css/bootstrap.css | 6316 ++
.../libs/bootstrap/custom/css/bootstrap.min.css | 9 +
.../custom/img/glyphicons-halflings-white.png | Bin 0 -> 8777 bytes
.../custom/img/glyphicons-halflings.png | Bin 0 -> 12799 bytes
.../js/libs/bootstrap/custom/js/bootstrap.js | 2291 +
.../libs/bootstrap/custom/js/bootstrap.min.js | 7 +
.../img/glyphicons-halflings-white.png | Bin 0 -> 8777 bytes
.../libs/bootstrap/img/glyphicons-halflings.png | Bin 0 -> 12799 bytes
.../js/libs/bootstrap/js/bootstrap.js | 2117 +
.../js/libs/bootstrap/js/bootstrap.min.js | 644 +
.../v101-portal-demo/js/libs/google-viz-api.js | 49 +
.../js/libs/jquery/jquery-1.9.1.min.js | 5 +
.../js/libs/jquery/jquery-migrate-1.1.1.min.js | 3 +
.../js/libs/jquery/jquery.sparkline.min.js | 5 +
.../js/libs/jqueryui/date.min.js | 2 +
.../ui-bg_diagonals-thick_90_eeeeee_40x40.png | Bin 0 -> 251 bytes
.../images/ui-bg_flat_100_deedf7_40x100.png | Bin 0 -> 182 bytes
.../images/ui-bg_flat_100_e4f1fb_40x100.png | Bin 0 -> 213 bytes
.../images/ui-bg_flat_100_f2f5f7_40x100.png | Bin 0 -> 212 bytes
.../images/ui-bg_flat_15_cd0a0a_40x100.png | Bin 0 -> 181 bytes
.../images/ui-bg_flat_50_3baae3_40x100.png | Bin 0 -> 182 bytes
.../images/ui-bg_flat_80_d7ebf9_40x100.png | Bin 0 -> 183 bytes
.../ui-bg_highlight-hard_70_000000_1x100.png | Bin 0 -> 118 bytes
.../ui-bg_highlight-soft_25_ffef8f_1x100.png | Bin 0 -> 153 bytes
.../jqueryui/images/ui-icons_000000_256x240.png | Bin 0 -> 4369 bytes
.../jqueryui/images/ui-icons_2694e8_256x240.png | Bin 0 -> 4369 bytes
.../jqueryui/images/ui-icons_2e83ff_256x240.png | Bin 0 -> 4369 bytes
.../jqueryui/images/ui-icons_3d80b3_256x240.png | Bin 0 -> 4369 bytes
.../jqueryui/images/ui-icons_72a7cf_256x240.png | Bin 0 -> 4369 bytes
.../jqueryui/images/ui-icons_ffffff_256x240.png | Bin 0 -> 4369 bytes
.../js/libs/jqueryui/jquery-ui-1.8.18.min.js | 15 +
.../js/libs/jqueryui/jquery-ui-1.8.9.custom.css | 1 +
.../js/libs/jqueryui/jquery-ui-timepicker.css | 1 +
.../libs/jqueryui/jquery.ui.timepicker.min.js | 1 +
.../ui-bootstrap-custom-0.3.0.min.js | 1 +
.../ui-bootstrap-custom-tpls-0.3.0.min.js | 1 +
.../v101-portal-demo/js/libs/usergrid.sdk.js | 2568 +
docs/jersey2skeleton/README.md | 9 +
docs/jersey2skeleton/pom.xml | 387 +
.../usergrid/testapp/AbstractResource.java | 51 +
.../apache/usergrid/testapp/ApiResponse.java | 52 +
.../org/apache/usergrid/testapp/Entity.java | 55 +
.../usergrid/testapp/ManagementResource.java | 56 +
.../usergrid/testapp/OrganizationResource.java | 64 +
.../usergrid/testapp/OrganizationsResource.java | 76 +
.../apache/usergrid/testapp/StatusResource.java | 34 +
.../java/org/apache/usergrid/testapp/User.java | 28 +
.../src/main/resources/applicationContext.xml | 28 +
.../main/resources/jersey2skeleton-context.xml | 28 +
.../src/main/webapp/WEB-INF/web.xml | 95 +
.../usergrid/testapp/AvailablePortFinder.java | 208 +
.../usergrid/testapp/OrganizationTest.java | 127 +
.../org/apache/usergrid/testapp/TomcatMain.java | 57 +
.../apache/usergrid/testapp/TomcatRuntime.java | 185 +
.../src/test/resources/log4j.properties | 31 +
.../src/test/resources/logging.properties | 26 +
docs/jersey2skeleton/templates/markdown.hbs | 108 +
docs/jersey2skeleton/templates/operation.hbs | 73 +
docs/jersey2skeleton/templates/security.hbs | 88 +
.../templates/strapdown.html.hbs | 10 +
docs/rest-endpoints/api-docs.html | 750 +-
docs/rest-endpoints/api-docs.md | 142 +-
stack/build-tools/pom.xml | 2 +-
stack/config/pom.xml | 2 +-
stack/core/pom.xml | 14 +-
.../corepersistence/CpEntityManager.java | 22 +-
.../corepersistence/CpRelationManager.java | 21 +-
.../asyncevents/EventBuilderImpl.java | 44 +-
.../pipeline/read/search/CandidateIdFilter.java | 3 +-
.../usergrid/persistence/EntityManager.java | 2 +
.../usergrid/persistence/CollectionIT.java | 10 +-
.../usergrid/persistence/PermissionsIT.java | 7 +-
stack/corepersistence/collection/pom.xml | 2 +-
stack/corepersistence/common/pom.xml | 2 +-
stack/corepersistence/graph/pom.xml | 2 +-
stack/corepersistence/map/pom.xml | 2 +-
stack/corepersistence/model/pom.xml | 2 +-
stack/corepersistence/pom.xml | 2 +-
stack/corepersistence/queryindex/pom.xml | 2 +-
.../index/impl/EsEntityIndexBatchImpl.java | 10 +-
.../index/impl/EsIndexProducerImpl.java | 8 +
.../persistence/index/impl/IndexProducer.java | 2 +
.../index/impl/IndexRefreshCommandImpl.java | 2 +-
.../persistence/index/impl/EntityIndexTest.java | 22 +-
.../persistence/index/impl/GeoPagingTest.java | 6 +-
stack/corepersistence/queue/pom.xml | 2 +-
stack/pom.xml | 252 +-
stack/rest/pom.xml | 141 +-
.../usergrid/rest/AbstractContextResource.java | 45 +-
.../usergrid/rest/ApplicationsResource.java | 178 -
.../apache/usergrid/rest/DatabaseResource.java | 98 -
.../org/apache/usergrid/rest/IndexResource.java | 54 +-
.../apache/usergrid/rest/MigrateResource.java | 61 +-
.../org/apache/usergrid/rest/RootResource.java | 92 +-
.../apache/usergrid/rest/SwaggerServlet.java | 5 +-
.../apache/usergrid/rest/SystemResource.java | 26 +-
.../usergrid/rest/TextToJsonSwapWriter.java | 182 +-
.../rest/applications/ApplicationResource.java | 30 +-
.../rest/applications/ServiceResource.java | 111 +-
.../applications/assets/AssetsResource.java | 13 +-
.../applications/events/EventsResource.java | 36 +-
.../notifiers/NotifierResource.java | 29 +-
.../notifiers/NotifiersResource.java | 29 +-
.../rest/applications/queues/QueueResource.java | 86 +-
.../queues/QueueSubscriberResource.java | 63 +-
.../queues/QueueSubscriptionResource.java | 63 +-
.../queues/QueueTransactionsResource.java | 40 +-
.../rest/applications/users/UserResource.java | 134 +-
.../rest/applications/users/UsersResource.java | 51 +-
.../exceptions/NotFoundExceptionMapper.java | 3 +-
.../rest/exceptions/SecurityException.java | 31 +-
.../exceptions/SecurityExceptionMapper.java | 3 +-
.../exceptions/UnsupportedMediaTypeMapper.java | 43 +
.../rest/filters/ContentTypeFilter.java | 74 +-
.../rest/filters/JSONPCallbackFilter.java | 26 +-
.../usergrid/rest/filters/MeteringFilter.java | 632 +-
.../usergrid/rest/filters/TracingFilter.java | 37 +-
.../rest/management/ManagementResource.java | 53 +-
.../management/metrics/MetricsResource.java | 9 +-
.../organizations/OrganizationResource.java | 64 +-
.../organizations/OrganizationsResource.java | 66 +-
.../applications/ApplicationResource.java | 56 +-
.../applications/ApplicationsResource.java | 20 +-
.../imports/FileErrorsResource.java | 41 +-
.../imports/FileIncludesResource.java | 15 +-
.../applications/imports/ImportsResource.java | 57 +-
.../organizations/users/UsersResource.java | 88 +-
.../rest/management/users/UserResource.java | 95 +-
.../rest/management/users/UsersResource.java | 11 +-
.../organizations/OrganizationsResource.java | 76 +-
.../organizations/OrganizationResource.java | 28 +-
.../rest/security/CrossOriginRequestFilter.java | 18 +-
.../security/SecuredResourceFilterFactory.java | 181 +-
.../ShiroAuthenticationExceptionMapper.java | 3 +-
.../shiro/filters/BasicAuthSecurityFilter.java | 35 +-
.../ClientCredentialsSecurityFilter.java | 25 +-
.../OAuth2AccessTokenSecurityFilter.java | 168 +-
.../security/shiro/filters/SecurityFilter.java | 33 +-
.../rest/system/ApplicationsResource.java | 178 +
.../usergrid/rest/system/DatabaseResource.java | 101 +
.../rest/system/QueueSystemResource.java | 77 +
.../apache/usergrid/rest/utils/CORSUtils.java | 42 +-
stack/rest/src/main/webapp/WEB-INF/web.xml | 63 +-
.../java/org/apache/usergrid/rest/BasicIT.java | 22 +-
.../usergrid/rest/CollectionMetadataIT.java | 7 +-
.../apache/usergrid/rest/IndexResourceIT.java | 43 +-
.../apache/usergrid/rest/NotificationsIT.java | 6 +-
.../apache/usergrid/rest/PartialUpdateTest.java | 16 +-
.../apache/usergrid/rest/SystemResourceIT.java | 27 +-
.../org/apache/usergrid/rest/TomcatRuntime.java | 12 +-
.../rest/applications/ApplicationCreateIT.java | 14 +-
.../rest/applications/ApplicationDeleteIT.java | 206 +-
.../applications/ApplicationResourceIT.java | 266 +-
.../applications/assets/AssetResourceIT.java | 21 +-
.../applications/assets/AwsAssetResourceIT.java | 53 +-
.../applications/assets/aws/NoAWSCredsRule.java | 15 +-
.../collection/CollectionsResourceIT.java | 26 +-
.../collection/DuplicateNameIT.java | 5 +-
.../activities/ActivityResourceIT.java | 5 +-
.../collection/devices/DevicesResourceIT.java | 4 +-
.../collection/groups/GroupResourceIT.java | 34 +-
.../collection/paging/PagingResourceIT.java | 65 +-
.../users/ConnectionResourceTest.java | 12 +-
.../collection/users/OwnershipResourceIT.java | 15 +-
.../collection/users/PermissionsResourceIT.java | 34 +-
.../collection/users/UserResourceIT.java | 126 +-
.../queries/BadGrammarQueryTest.java | 15 +-
.../applications/queries/BasicGeoTests.java | 45 +-
.../rest/filters/ContentTypeResourceIT.java | 70 +-
.../usergrid/rest/management/AccessTokenIT.java | 42 +-
.../usergrid/rest/management/AdminUsersIT.java | 48 +-
.../rest/management/ExportResourceIT.java | 137 +-
.../rest/management/ImportResourceIT.java | 50 +-
.../rest/management/ManagementResourceIT.java | 149 +-
.../rest/management/OrganizationsIT.java | 60 +-
.../rest/management/RegistrationIT.java | 32 +-
.../organizations/AdminEmailEncodingIT.java | 5 +-
.../rest/test/resource/AbstractRestIT.java | 79 +-
.../rest/test/resource/ClientSetup.java | 4 +-
.../usergrid/rest/test/resource/RestClient.java | 44 +-
.../test/resource/TestPropertiesResource.java | 14 +-
.../endpoints/ApplicationsResource.java | 8 +-
.../resource/endpoints/BoostrapResource.java | 19 +-
.../resource/endpoints/CollectionEndpoint.java | 109 +-
.../test/resource/endpoints/EntityEndpoint.java | 40 +-
.../test/resource/endpoints/NamedResource.java | 241 +-
.../test/resource/endpoints/RootResource.java | 9 +-
.../test/resource/endpoints/SetupResource.java | 20 +-
.../test/resource/endpoints/SubResource.java | 37 +
.../test/resource/endpoints/SystemResource.java | 51 +-
.../test/resource/endpoints/TokenResource.java | 15 +-
.../test/resource/endpoints/UrlResource.java | 11 +-
.../endpoints/mgmt/ApplicationResource.java | 27 +-
.../endpoints/mgmt/ApplicationsResource.java | 4 +-
.../endpoints/mgmt/AuthorizeResource.java | 7 +-
.../endpoints/mgmt/ConfirmResource.java | 12 +-
.../endpoints/mgmt/CredentialsResource.java | 7 +-
.../resource/endpoints/mgmt/FeedResource.java | 5 +-
.../resource/endpoints/mgmt/OrgResource.java | 68 +-
.../endpoints/mgmt/OrganizationResource.java | 15 +-
.../endpoints/mgmt/PasswordResource.java | 8 -
.../endpoints/mgmt/ReactivateResource.java | 10 +-
.../resource/endpoints/mgmt/ResetResource.java | 10 +-
.../resource/endpoints/mgmt/TokenResource.java | 7 +-
.../resource/endpoints/mgmt/UserResource.java | 20 +-
.../resource/endpoints/mgmt/UsersResource.java | 6 -
stack/rest/src/test/resources/log4j.properties | 36 +-
.../rest/src/test/resources/logging.properties | 13 +-
stack/services/pom.xml | 51 +-
.../security/providers/AbstractProvider.java | 16 +-
.../security/providers/FacebookProvider.java | 7 +-
.../security/providers/FoursquareProvider.java | 38 +-
.../providers/PingIdentityProvider.java | 38 +-
.../usergrid/services/roles/RolesService.java | 12 -
.../usergrid/management/OrganizationIT.java | 18 +-
stack/test-utils/pom.xml | 2 +-
.../org/apache/usergrid/tools/ExportAdmins.java | 17 +-
website/README.md | 3 +-
website/Rules | 52 +
website/build.sh | 1 +
website/content/bootstrap/bootstrap.min.css | 9 +
website/content/community/index.html | 22 +-
website/content/css/bootflat-extensions.css | 356 +
website/content/css/bootflat-square.css | 69 +
website/content/css/bootflat.css | 1560 +
website/content/css/font-awesome.min.css | 405 +
website/content/css/usergrid-site.css | 1553 +
website/content/favicon.ico | Bin 0 -> 3989 bytes
website/content/font/FontAwesome.otf | Bin 0 -> 61896 bytes
.../content/font/fontawesome-webfont-eot.eot | Bin 0 -> 37405 bytes
.../content/font/fontawesome-webfont-svg.svg | 399 +
.../content/font/fontawesome-webfont-ttf.ttf | Bin 0 -> 79076 bytes
.../content/font/fontawesome-webfont-woff.woff | Bin 0 -> 43572 bytes
website/content/img/alberto.jpg | Bin 0 -> 16137 bytes
website/content/img/alex.png | Bin 0 -> 40842 bytes
website/content/img/apache_usergrid_favicon.png | Bin 0 -> 10735 bytes
.../content/img/apache_usergrid_logo_white.png | Bin 0 -> 26418 bytes
.../img/apache_usergrid_logo_white_small.png | Bin 0 -> 11905 bytes
website/content/img/check_flat/default.png | Bin 0 -> 25851 bytes
website/content/img/dave.jpg | Bin 0 -> 14005 bytes
website/content/img/ed.jpg | Bin 0 -> 20460 bytes
website/content/img/egg-logo.png | Bin 0 -> 9938 bytes
website/content/img/github.png | Bin 0 -> 8936 bytes
website/content/img/grey.png | Bin 0 -> 37896 bytes
website/content/img/intellij.png | Bin 0 -> 9199 bytes
website/content/img/jeff.jpg | Bin 0 -> 13857 bytes
website/content/img/michael_r.jpg | Bin 0 -> 10244 bytes
website/content/img/mike_d.JPG | Bin 0 -> 36443 bytes
website/content/img/nate.jpg | Bin 0 -> 4291 bytes
website/content/img/rod.jpg | Bin 0 -> 40313 bytes
website/content/img/scott.jpg | Bin 0 -> 8555 bytes
website/content/img/shawn.jpg | Bin 0 -> 69304 bytes
website/content/img/stliu.jpg | Bin 0 -> 51303 bytes
website/content/img/strong.jpg | Bin 0 -> 7434 bytes
website/content/img/structure101.png | Bin 0 -> 6475 bytes
website/content/img/sungju.jpg | Bin 0 -> 11440 bytes
website/content/img/tim.jpg | Bin 0 -> 7611 bytes
website/content/img/todd.jpg | Bin 0 -> 18142 bytes
website/content/img/usergrid-logo.pdf | 398 +
website/content/img/usergrid.png | Bin 0 -> 21994 bytes
website/content/img/usergrid_160.png | Bin 0 -> 2126 bytes
website/content/img/usergrid_200.png | Bin 0 -> 6397 bytes
website/content/img/usergrid_300.png | Bin 0 -> 16330 bytes
.../content/img/usergrid_300_transparent.png | Bin 0 -> 16308 bytes
website/content/img/usergrid_400.png | Bin 0 -> 8746 bytes
website/content/img/usergrid_800.png | Bin 0 -> 14452 bytes
website/content/img/usergrid_card.png | Bin 0 -> 23295 bytes
website/content/img/usergrid_logo.png | Bin 0 -> 118086 bytes
website/content/img/usergrid_logo_205_50.png | Bin 0 -> 7058 bytes
website/content/img/usergrid_logo_260_50.png | Bin 0 -> 8682 bytes
website/content/img/usergrid_logo_720.png | Bin 0 -> 27610 bytes
website/content/img/usergrid_logo_720p.png | Bin 0 -> 27608 bytes
website/content/img/usergrid_logo_900_200.png | Bin 0 -> 12273 bytes
website/content/img/usergrid_logo_white.png | Bin 0 -> 16900 bytes
website/content/img/usergrid_profile_128.png | Bin 0 -> 6689 bytes
website/content/img/usergrid_profile_256.png | Bin 0 -> 10470 bytes
.../content/img/usergrid_profile_256_white.png | Bin 0 -> 10724 bytes
.../img/usergrid_profile_512_margins.png | Bin 0 -> 19112 bytes
.../content/img/usergrid_profile_64_white.png | Bin 0 -> 4839 bytes
.../content/img/usergrid_profile_background.png | Bin 0 -> 6710 bytes
website/content/img/usergrid_screencast_bg.png | Bin 0 -> 20478 bytes
website/content/img/usergrid_small_100.png | Bin 0 -> 6681 bytes
website/content/img/usergrid_wiki.png | Bin 0 -> 7135 bytes
website/content/img/yourkit.jpeg | Bin 0 -> 7763 bytes
website/content/js/bootstrap.min.js | 8 +
website/content/js/head.js | 708 +
website/content/js/html5shiv.js | 8 +
website/content/js/jquery-1.10.1.min.js | 6 +
website/content/js/jquery.icheck.js | 397 +
website/content/js/respond.min.js | 6 +
website/content/js/usergrid-site.js | 50 +
website/content/static/github-btn.html | 2 +
website/crash.log | 143 +
website/layouts/community.html | 1 +
website/layouts/docs.html | 1 +
website/lib/default.rb | 43 +
website/lib/helpers_.rb | 0
website/lib/pandoc.template | 4 +
website/nanoc.yaml | 77 +
website/run.sh | 1 +
website/tmp/checksums | 4 +-
website/tmp/compiled_content | 87099 +----------------
website/tmp/dependencies | Bin 2372 -> 2397 bytes
website/tmp/rule_memory | Bin 0 -> 5163 bytes
website/utilities/map-markers.rb | 62 +
website/utilities/markers.txt | 440 +
website/utilities/snapshot-apigee.rb | 71 +
website/utilities/usergrid.csv | 290 +
477 files changed, 117021 insertions(+), 90476 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/e17dab4c/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManager.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/e17dab4c/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
----------------------------------------------------------------------
diff --cc stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
index aad7610,f1e6ad6..1bd5c7d
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
@@@ -527,16 -525,11 +526,12 @@@ public class CpRelationManager implemen
*
*/
- final EntityIndex ei = managerCache.getEntityIndex( applicationScope );
- final EntityIndexBatch batch = ei.createBatch();
-
- // remove item from collection index
- SearchEdge indexScope = createCollectionSearchEdge( cpHeadEntity.getId(), collectionName );
- batch.deindex( indexScope, memberEntity );
+ //TODO: this should not happen here, needs to go to SQS
+ //indexProducer.put(batch).subscribe();
+ indexService.queueEntityDelete(applicationScope,memberEntity.getId());
- managerCache.getIndexProducer().put( batch.build()).toBlocking().lastOrDefault(null); // this should throw an exception
+
// special handling for roles collection of a group
if ( headEntity.getType().equals( Group.ENTITY_TYPE ) ) {
http://git-wip-us.apache.org/repos/asf/usergrid/blob/e17dab4c/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/EventBuilderImpl.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/e17dab4c/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/search/CandidateIdFilter.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/e17dab4c/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsIndexProducerImpl.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/e17dab4c/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexRefreshCommandImpl.java
----------------------------------------------------------------------
diff --cc stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexRefreshCommandImpl.java
index 6b8b024,087eefe..04f8652
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexRefreshCommandImpl.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexRefreshCommandImpl.java
@@@ -67,7 -67,7 +67,7 @@@ public class IndexRefreshCommandImpl im
final EsProvider esProvider,
final IndexProducer producer,
final IndexFig indexFig,
-- final MetricsFactory metricsFactory,
++ l final MetricsFactory metricsFactory,
final IndexCache indexCache ) {
http://git-wip-us.apache.org/repos/asf/usergrid/blob/e17dab4c/stack/rest/src/main/java/org/apache/usergrid/rest/system/ApplicationsResource.java
----------------------------------------------------------------------
diff --cc stack/rest/src/main/java/org/apache/usergrid/rest/system/ApplicationsResource.java
index 0000000,82bc4c8..55887c7
mode 000000,100644..100644
--- a/stack/rest/src/main/java/org/apache/usergrid/rest/system/ApplicationsResource.java
+++ b/stack/rest/src/main/java/org/apache/usergrid/rest/system/ApplicationsResource.java
@@@ -1,0 -1,168 +1,178 @@@
+ /*
+ *
+ * * Licensed to the Apache Software Foundation (ASF) under one or more
+ * * contributor license agreements. The ASF licenses this file to You
+ * * under the Apache License, Version 2.0 (the "License"); you may not
+ * * use this file except in compliance with the License.
+ * * You may obtain a copy of the License at
+ * *
+ * * http://www.apache.org/licenses/LICENSE-2.0
+ * *
+ * * Unless required by applicable law or agreed to in writing, software
+ * * distributed under the License is distributed on an "AS IS" BASIS,
+ * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * * See the License for the specific language governing permissions and
+ * * limitations under the License. For additional information regarding
+ * * copyright in this work, please see the NOTICE file in the top level
+ * * directory of this distribution.
+ *
+ */
+ package org.apache.usergrid.rest.system;
+
+ import com.fasterxml.jackson.jaxrs.json.annotation.JSONP;
+ import org.apache.usergrid.corepersistence.service.StatusService;
+ import org.apache.usergrid.persistence.EntityManager;
+ import org.apache.usergrid.persistence.model.util.UUIDGenerator;
+ import org.apache.usergrid.rest.AbstractContextResource;
+ import org.apache.usergrid.rest.ApiResponse;
+ import org.apache.usergrid.rest.security.annotations.RequireSystemAccess;
+ import org.slf4j.Logger;
+ import org.slf4j.LoggerFactory;
+ import org.springframework.context.annotation.Scope;
+ import org.springframework.stereotype.Component;
+
+ import javax.ws.rs.*;
+ import javax.ws.rs.core.Context;
+ import javax.ws.rs.core.MediaType;
+ import javax.ws.rs.core.UriInfo;
+ import java.util.HashMap;
+ import java.util.LinkedHashMap;
+ import java.util.Map;
+ import java.util.UUID;
+ import java.util.concurrent.atomic.AtomicInteger;
+
+ /**
+ * Classy class class.
+ */
+ @Component
+ @Scope( "singleton" )
+ @Produces( {
+ MediaType.APPLICATION_JSON, "application/javascript", "application/x-javascript", "text/ecmascript",
+ "application/ecmascript", "text/jscript"
+ } )
+ public class ApplicationsResource extends AbstractContextResource {
+
+ private static final Logger logger = LoggerFactory.getLogger(ApplicationsResource.class);
+
+
+ public ApplicationsResource() {
+ logger.info( "ApplicationsResource initialized" );
+ }
+
+ @RequireSystemAccess
+ @DELETE
+ @JSONP
+ @Path( "{applicationId}" )
+ public ApiResponse clearApplication(
+ @Context UriInfo ui,
+ @PathParam("applicationId") UUID applicationId,
+ @QueryParam( "confirmApplicationName" ) String confirmApplicationName,
+ @QueryParam( "limit" ) int limit,
+ @QueryParam( "callback" ) @DefaultValue( "callback" ) String callback )
+
+ throws Exception {
+
+ if(confirmApplicationName == null){
+ throw new IllegalArgumentException("please make add a QueryString for confirmApplicationName");
+ }
+
+ final UUID jobId = UUIDGenerator.newTimeUUID();
+
+ final EntityManager em = emf.getEntityManager(applicationId);
+ final String name = em.getApplication().getApplicationName();
+ if(!name.toLowerCase().equals(confirmApplicationName.toLowerCase())){
+ throw new IllegalArgumentException(
+ "confirmApplicationName: " + confirmApplicationName + " does not equal " + name);
+ }
+ final StatusService statusService = injector.getInstance(StatusService.class);
+
+ final ApiResponse response = createApiResponse();
+
+ response.setAction( "clear application" );
+
+ logger.info("clearing up application");
+
+ final Thread delete = new Thread() {
+
+ @Override
+ public void run() {
+ final AtomicInteger itemsDeleted = new AtomicInteger(0);
+ try {
+ management.deleteAllEntities(applicationId, limit)
+ .map(id -> itemsDeleted.incrementAndGet())
+ .doOnNext(count -> {
+ if( count % 100 == 0 ){
+ Map<String,Object> map = new LinkedHashMap<>();
+ map.put("count",itemsDeleted.intValue());
+ final StatusService statusService = injector.getInstance(StatusService.class);
- statusService.setStatus(applicationId, jobId, StatusService.Status.INPROGRESS,map).subscribe();
++ statusService.setStatus(applicationId, jobId, StatusService.Status.INPROGRESS,map)
++ .subscribe();//do not want to throw this exception
+ }
+ })
- .doOnCompleted(() ->{
- Map<String,Object> map = new LinkedHashMap<>();
- map.put("count",itemsDeleted.intValue());
++ .doOnCompleted(() -> {
++ Map<String, Object> map = new LinkedHashMap<>();
++ map.put("count", itemsDeleted.intValue());
+ final StatusService statusService = injector.getInstance(StatusService.class);
- statusService.setStatus(applicationId,jobId, StatusService.Status.COMPLETE,map).subscribe();
++ statusService.setStatus(applicationId, jobId, StatusService.Status.COMPLETE, map)
++ .toBlocking().lastOrDefault(null);//want to rethrow this exception
+ })
- .subscribe();
++ .toBlocking().lastOrDefault(null);//expecting exception to be caught if job fails
+
+ } catch ( Exception e ) {
+ Map<String,Object> map = new LinkedHashMap<>();
+ map.put("exception",e);
- statusService.setStatus(applicationId,jobId, StatusService.Status.FAILED,map).subscribe();
++ try {
++ statusService.setStatus(applicationId, jobId, StatusService.Status.FAILED, map).toBlocking().lastOrDefault(null);//leave as subscribe if fails retry
++ }catch (Exception subE){
++ logger.error("failed to update status "+jobId,subE);
++ }
+ logger.error( "Failed to delete appid:"+applicationId + " jobid:"+jobId+" count:"+itemsDeleted, e );
+ }
+ }
+ };
+
+ delete.setName("Delete for app : " + applicationId + " job: " + jobId);
+ delete.setDaemon(true);
+ delete.start();
+
- statusService.setStatus(applicationId,jobId, StatusService.Status.STARTED,new LinkedHashMap<>()).subscribe();
-
++ try {
++ //should throw exception if can't start
++ statusService.setStatus(applicationId, jobId, StatusService.Status.STARTED, new LinkedHashMap<>()).toBlocking().lastOrDefault(null);
++ }catch (Exception e){
++ logger.error("failed to set status for " + jobId, e);
++ }
+ Map<String,Object> data = new HashMap<>();
+ data.put("jobId",jobId);
+ data.put("status",StatusService.Status.STARTED);
+ response.setData(data);
+ response.setSuccess();
+ return response;
+ }
+
+ @RequireSystemAccess
+ @GET
+ @Path( "{applicationId}/job/{jobId}" )
+ public ApiResponse getStatus(
+ @Context UriInfo ui,
+ @PathParam("applicationId") UUID applicationId,
+ @PathParam("jobId") UUID jobId,
+ @QueryParam( "callback" ) @DefaultValue( "callback" ) String callback ) throws Exception{
+
+ final StatusService statusService = injector.getInstance(StatusService.class);
+
+ final ApiResponse response = createApiResponse();
+
+ response.setAction( "clear application" );
+
+ StatusService.JobStatus jobStatus = statusService.getStatus(applicationId, jobId).toBlocking().lastOrDefault(null);
+
+ Map<String,Object> data = new HashMap<>();
+ data.put("jobId",jobId);
+ data.put( "status", jobStatus.getStatus().toString() );
+ data.put( "metadata", jobStatus.getData() );
+ response.setData(data);
+ response.setSuccess();
+ return response;
+ }
+
+ }
[11/23] usergrid git commit: change queue impl
Posted by sf...@apache.org.
change queue impl
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/4fa6749f
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/4fa6749f
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/4fa6749f
Branch: refs/heads/master
Commit: 4fa6749f027ece9388725dcdeb875f4bddbf7f5f
Parents: 0cc225f
Author: Shawn Feldman <sf...@apache.org>
Authored: Fri Oct 16 15:07:36 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Fri Oct 16 15:07:36 2015 -0600
----------------------------------------------------------------------
.../persistence/queue/DefaultQueueManager.java | 34 ++++++++++++++++----
1 file changed, 27 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/4fa6749f/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
index 0ef2849..ae0b0aa 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
@@ -30,17 +30,29 @@ import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.PriorityBlockingQueue;
+import java.util.concurrent.TimeUnit;
/**
* Default queue manager implementation, uses in memory linked queue
*/
public class DefaultQueueManager implements QueueManager {
- public LinkedBlockingQueue<QueueMessage> queue = new LinkedBlockingQueue<>();
+ public ArrayBlockingQueue<QueueMessage> queue = new ArrayBlockingQueue<>(1000);
@Override
- public Observable<QueueMessage> getMessages(int limit, int transactionTimeout, int waitTime, Class klass) {
+ public Observable<QueueMessage> getMessages(int limit, int transactionTimeout, int waitTime, Class klass) {
List<QueueMessage> returnQueue = new ArrayList<>();
- queue.drainTo(returnQueue,1000);
+ try {
+ QueueMessage message=null;
+ int count = 10;
+ do {
+ message = queue.poll(100, TimeUnit.MILLISECONDS);
+ if (message != null) {
+ returnQueue.add(message);
+ }
+ }while(message!=null && count-->0);
+ }catch (InterruptedException ie){
+ throw new RuntimeException(ie);
+ }
return Observable.from( returnQueue);
}
@@ -58,17 +70,25 @@ public class DefaultQueueManager implements QueueManager {
}
@Override
- public synchronized void sendMessages(List bodies) throws IOException {
+ public void sendMessages(List bodies) throws IOException {
for(Object body : bodies){
String uuid = UUID.randomUUID().toString();
- queue.add(new QueueMessage(uuid,"handle_"+uuid,body,"putappriate type here"));
+ try {
+ queue.put(new QueueMessage(uuid, "handle_" + uuid, body, "put type here"));
+ }catch (InterruptedException ie){
+ throw new RuntimeException(ie);
+ }
}
}
@Override
- public synchronized void sendMessage(Object body) throws IOException {
+ public void sendMessage(Object body) throws IOException {
String uuid = UUID.randomUUID().toString();
- queue.add(new QueueMessage(uuid,"handle_"+uuid,body,"put type here"));
+ try {
+ queue.put(new QueueMessage(uuid, "handle_" + uuid, body, "put type here"));
+ }catch (InterruptedException ie){
+ throw new RuntimeException(ie);
+ }
}
@Override
[08/23] usergrid git commit: change queue impl
Posted by sf...@apache.org.
change queue impl
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/d536b5b4
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/d536b5b4
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/d536b5b4
Branch: refs/heads/master
Commit: d536b5b4e3a377948e3427c95bedd5b178428850
Parents: 505b44a
Author: Shawn Feldman <sf...@apache.org>
Authored: Wed Oct 14 15:38:25 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Wed Oct 14 15:38:25 2015 -0600
----------------------------------------------------------------------
.../usergrid/persistence/queue/DefaultQueueManager.java | 7 +++++--
1 file changed, 5 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/d536b5b4/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
index edd3b6b..d9a141e 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
@@ -23,21 +23,24 @@ package org.apache.usergrid.persistence.queue;
import rx.Observable;
import java.io.IOException;
+import java.util.AbstractQueue;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.PriorityBlockingQueue;
/**
* Default queue manager implementation, uses in memory linked queue
*/
public class DefaultQueueManager implements QueueManager {
- public ArrayBlockingQueue<QueueMessage> queue = new ArrayBlockingQueue<>(10000000);
+ public LinkedBlockingQueue<QueueMessage> queue = new LinkedBlockingQueue<>();
@Override
public synchronized Observable<QueueMessage> getMessages(int limit, int transactionTimeout, int waitTime, Class klass) {
List<QueueMessage> returnQueue = new ArrayList<>();
- queue.drainTo(returnQueue,1);
+ queue.drainTo(returnQueue,10);
return Observable.from( returnQueue);
}
[20/23] usergrid git commit: merge
Posted by sf...@apache.org.
merge
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/68befd2f
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/68befd2f
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/68befd2f
Branch: refs/heads/master
Commit: 68befd2f04d03dbd3e9370af89c257a8a28112a1
Parents: 00175cc
Author: Shawn Feldman <sf...@apache.org>
Authored: Mon Oct 19 15:13:34 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Mon Oct 19 15:13:34 2015 -0600
----------------------------------------------------------------------
.../index/IndexProcessorFig.java | 36 ++++++--------------
1 file changed, 11 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/68befd2f/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
index 113b539..8a8c8a5 100644
--- 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
@@ -49,15 +49,15 @@ public interface IndexProcessorFig extends GuicyFig {
* Set the amount of time to wait when Elasticsearch rejects a requests before
* retrying. This provides simple back pressure. (in milliseconds)
*/
- @Default( "1000" )
- @Key( FAILURE_REJECTED_RETRY_WAIT_TIME )
+ @Default("1000")
+ @Key(FAILURE_REJECTED_RETRY_WAIT_TIME)
long getFailureRetryTime();
/**
* Set the read timeout for processing messages in the queue. (in milliseconds)
*/
- @Default( "10000" )
- @Key( INDEX_QUEUE_READ_TIMEOUT )
+ @Default("10000")
+ @Key(INDEX_QUEUE_READ_TIMEOUT)
int getIndexQueueTimeout();
/**
@@ -65,15 +65,15 @@ public interface IndexProcessorFig extends GuicyFig {
* Received messages will remain 'in flight' until they are ack'd(deleted) or this timeout occurs.
* If the timeout occurs, the messages will become visible again for re-processing.
*/
- @Default( "5000" ) // 5 seconds
- @Key( INDEX_QUEUE_VISIBILITY_TIMEOUT )
+ @Default("5000") // 5 seconds
+ @Key(INDEX_QUEUE_VISIBILITY_TIMEOUT)
int getIndexQueueVisibilityTimeout();
/**
* The number of worker threads used to read index write requests from the queue.
*/
- @Default( "16" )
- @Key( ELASTICSEARCH_WORKER_COUNT )
+ @Default("16")
+ @Key(ELASTICSEARCH_WORKER_COUNT)
int getWorkerCount();
/**
@@ -81,26 +81,12 @@ public interface IndexProcessorFig extends GuicyFig {
* Valid values: TEST, LOCAL, SQS, SNS
* NOTE: SQS and SNS equate to the same implementation of Amazon queue services.
*/
- @Default( "LOCAL" )
- @Key( ELASTICSEARCH_QUEUE_IMPL )
+ @Default("LOCAL")
+ @Key(ELASTICSEARCH_QUEUE_IMPL)
String getQueueImplementation();
@Default("1000")
- @Key("elasticsearch.reindex.flush.interval")
- int getUpdateInterval();
-
- @Default("100")
- @Key("elasticsearch.buffer.time_ms")
- int getBufferTime();
-
- @Default("1000")
- @Key( REINDEX_BUFFER_SIZE )
+ @Key(REINDEX_BUFFER_SIZE)
int getReindexBufferSize();
- /**
- * Flag to resolve the LOCAL queue implementation service synchronously.
- */
- @Default("false")
- @Key("elasticsearch.queue_impl.resolution")
- boolean resolveSynchronously();
}
[02/23] usergrid git commit: removed search from refresh need to use
queue
Posted by sf...@apache.org.
removed search from refresh need to use queue
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/f8614a68
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/f8614a68
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/f8614a68
Branch: refs/heads/master
Commit: f8614a68f112469713874f032e40ab9fcc9b24d8
Parents: 2d23aa6
Author: Shawn Feldman <sf...@apache.org>
Authored: Fri Oct 9 15:28:33 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Fri Oct 9 15:28:33 2015 -0600
----------------------------------------------------------------------
.../corepersistence/CpEntityManager.java | 43 ++++--
.../asyncevents/EventBuilderImpl.java | 10 +-
.../org/apache/usergrid/CoreApplication.java | 14 +-
.../corepersistence/StaleIndexCleanupTest.java | 8 +-
.../persistence/ApplicationServiceIT.java | 9 +-
.../index/impl/IndexRefreshCommandImpl.java | 134 +++----------------
.../persistence/queue/DefaultQueueManager.java | 2 +-
7 files changed, 76 insertions(+), 144 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/f8614a68/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 d46c112..643a2b8 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
@@ -41,6 +41,11 @@ import org.apache.usergrid.corepersistence.service.ConnectionService;
import org.apache.usergrid.persistence.index.EntityIndex;
import org.apache.usergrid.persistence.index.IndexLocationStrategy;
import org.apache.usergrid.persistence.index.IndexRefreshCommand;
+import org.apache.usergrid.persistence.index.utils.*;
+import org.apache.usergrid.utils.*;
+import org.apache.usergrid.utils.ClassUtils;
+import org.apache.usergrid.utils.ConversionUtils;
+import org.apache.usergrid.utils.UUIDUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;
@@ -102,11 +107,6 @@ import org.apache.usergrid.persistence.model.entity.SimpleId;
import org.apache.usergrid.persistence.model.field.Field;
import org.apache.usergrid.persistence.model.field.StringField;
import org.apache.usergrid.persistence.model.util.UUIDGenerator;
-import org.apache.usergrid.utils.ClassUtils;
-import org.apache.usergrid.utils.CompositeUtils;
-import org.apache.usergrid.utils.Inflector;
-import org.apache.usergrid.utils.StringUtils;
-import org.apache.usergrid.utils.UUIDUtils;
import com.codahale.metrics.Meter;
import com.codahale.metrics.Timer;
@@ -2874,13 +2874,36 @@ public class CpEntityManager implements EntityManager {
* TODO, these 3 methods are super janky. During refactoring we should clean this model up
*/
public IndexRefreshCommand.IndexRefreshCommandInfo refreshIndex() {
+ try {
+ long start = System.currentTimeMillis();
+ // refresh special indexes without calling EntityManager refresh because stack overflow
+ Map<String, Object> map = new org.apache.usergrid.persistence.index.utils.MapUtils.HashMapBuilder<>();
+ map.put("some prop", "test");
+ boolean hasFinished = false;
+ Entity refreshEntity = create("refresh", map);
+ try {
+ for (int i = 0; i < 10; i++) {
+ if (searchCollection(
+ new SimpleEntityRef(org.apache.usergrid.persistence.entities.Application.ENTITY_TYPE, getApplicationId()),
+ InflectionUtils.pluralize("refresh"),
+ Query.fromQL("select * where uuid='" + refreshEntity.getUuid() + "'")
+ ).size() > 0
+ ) {
+ hasFinished = true;
+ break;
+ }
+ Thread.sleep(250);
+ return managerCache.getEntityIndex(applicationScope).refreshAsync().toBlocking().first();
+ }
+ }finally {
+ delete(refreshEntity);
+ }
+ return new IndexRefreshCommand.IndexRefreshCommandInfo(hasFinished,System.currentTimeMillis() - start);
+ } catch (Exception e) {
+ throw new RuntimeException("refresh failed",e);
+ }
- // refresh special indexes without calling EntityManager refresh because stack overflow
-
- return managerCache.getEntityIndex(applicationScope).refreshAsync().toBlocking().first();
}
-
-
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/f8614a68/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/EventBuilderImpl.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/EventBuilderImpl.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/EventBuilderImpl.java
index 4bf5695..bc72207 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/EventBuilderImpl.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/EventBuilderImpl.java
@@ -108,10 +108,12 @@ public class EventBuilderImpl implements EventBuilder {
log.debug( "Deleting in app scope {} with edge {} }", applicationScope, edge );
final Observable<IndexOperationMessage> edgeObservable =
- indexService.deleteIndexEdge( applicationScope, edge ).flatMap( batch -> {
- final GraphManager gm = graphManagerFactory.createEdgeManager( applicationScope );
- return gm.deleteEdge( edge ).map( deletedEdge -> batch );
- } );
+ indexService.deleteIndexEdge( applicationScope, edge )
+ .map( batch -> {
+ final GraphManager gm = graphManagerFactory.createEdgeManager(applicationScope);
+ gm.deleteEdge(edge).toBlocking().lastOrDefault(null);
+ return batch;
+ } );
return edgeObservable;
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/f8614a68/stack/core/src/test/java/org/apache/usergrid/CoreApplication.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/CoreApplication.java b/stack/core/src/test/java/org/apache/usergrid/CoreApplication.java
index 9c96fb8..d207894 100644
--- a/stack/core/src/test/java/org/apache/usergrid/CoreApplication.java
+++ b/stack/core/src/test/java/org/apache/usergrid/CoreApplication.java
@@ -17,10 +17,7 @@
package org.apache.usergrid;
-import java.util.HashMap;
-import java.util.LinkedHashMap;
-import java.util.Map;
-import java.util.UUID;
+import java.util.*;
import java.util.concurrent.ExecutionException;
import com.google.inject.Injector;
@@ -28,8 +25,10 @@ import org.apache.usergrid.corepersistence.index.IndexLocationStrategyFactory;
import org.apache.usergrid.corepersistence.service.ApplicationService;
import org.apache.usergrid.corepersistence.util.CpNamingUtils;
import org.apache.usergrid.persistence.index.*;
+import org.apache.usergrid.persistence.index.utils.MapUtils;
import org.apache.usergrid.persistence.model.entity.Id;
import org.apache.usergrid.persistence.model.entity.SimpleId;
+import org.apache.usergrid.utils.InflectionUtils;
import org.junit.rules.TestRule;
import org.junit.runner.Description;
import org.junit.runners.model.Statement;
@@ -235,14 +234,11 @@ public class CoreApplication implements Application, TestRule {
//Insert test entity and find it
setup.getEmf().refreshIndex(CpNamingUtils.getManagementApplicationId().getUuid());
- if(!em.getApplicationId().equals(CpNamingUtils.getManagementApplicationId().getUuid())) {
+ if (!em.getApplicationId().equals(CpNamingUtils.getManagementApplicationId().getUuid())) {
setup.getEmf().refreshIndex(em.getApplicationId());
}
- try {
- Thread.sleep(2000);
- }catch (Exception e){
- }
+ em.refreshIndex();
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/f8614a68/stack/core/src/test/java/org/apache/usergrid/corepersistence/StaleIndexCleanupTest.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/corepersistence/StaleIndexCleanupTest.java b/stack/core/src/test/java/org/apache/usergrid/corepersistence/StaleIndexCleanupTest.java
index 3e46e4f..df93e68 100644
--- a/stack/core/src/test/java/org/apache/usergrid/corepersistence/StaleIndexCleanupTest.java
+++ b/stack/core/src/test/java/org/apache/usergrid/corepersistence/StaleIndexCleanupTest.java
@@ -321,8 +321,8 @@ public class StaleIndexCleanupTest extends AbstractCoreIT {
final EntityManager em = app.getEntityManager();
- final int numEntities = 20;
- final int numUpdates = 40;
+ final int numEntities = 5;
+ final int numUpdates = 5;
// create lots of entities
final List<Entity> things = new ArrayList<Entity>(numEntities);
@@ -348,7 +348,7 @@ public class StaleIndexCleanupTest extends AbstractCoreIT {
for ( int j=0; j<numUpdates; j++) {
toUpdate = em.get( thing.getUuid() );
- toUpdate.setProperty( "property" + j, RandomStringUtils.randomAlphanumeric(10));
+ toUpdate.setProperty( "property" + j, UUID.randomUUID().toString());
em.update(toUpdate);
@@ -367,9 +367,7 @@ public class StaleIndexCleanupTest extends AbstractCoreIT {
if(numEntities * (numUpdates + 1) == crs.size()){
break;
}
- Thread.sleep(250);
crs = queryCollectionCp("things", "thing", "select *");
-
}
// Assert.assertEquals("Expect stale candidates", numEntities * (numUpdates + 1), crs.size());
http://git-wip-us.apache.org/repos/asf/usergrid/blob/f8614a68/stack/core/src/test/java/org/apache/usergrid/persistence/ApplicationServiceIT.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/persistence/ApplicationServiceIT.java b/stack/core/src/test/java/org/apache/usergrid/persistence/ApplicationServiceIT.java
index d870114..f8079e5 100644
--- a/stack/core/src/test/java/org/apache/usergrid/persistence/ApplicationServiceIT.java
+++ b/stack/core/src/test/java/org/apache/usergrid/persistence/ApplicationServiceIT.java
@@ -76,7 +76,6 @@ public class ApplicationServiceIT extends AbstractCoreIT {
count = ids.count().toBlocking().last();
Assert.assertEquals(count, 5);
this.app.refreshIndex();
- Thread.sleep(5000);
Injector injector = SpringResource.getInstance().getBean(Injector.class);
GraphManagerFactory factory = injector.getInstance(GraphManagerFactory.class);
GraphManager graphManager = factory.createEdgeManager(appScope);
@@ -88,7 +87,13 @@ public class ApplicationServiceIT extends AbstractCoreIT {
Iterator<Edge> results = graphManager.loadEdgesFromSource(simpleSearchByEdgeType).toBlocking().getIterator();
if(results.hasNext()){
- Assert.fail("should be empty");
+ int i = 0;
+
+ while(results.hasNext()){
+ results.next();
+ i++;
+ }
+ Assert.fail("should be empty but has "+i);
}else{
Results searchCollection = entityManager.searchCollection(entityManager.getApplication(), "tests", Query.all());
http://git-wip-us.apache.org/repos/asf/usergrid/blob/f8614a68/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexRefreshCommandImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexRefreshCommandImpl.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexRefreshCommandImpl.java
index 087eefe..6b8b024 100644
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexRefreshCommandImpl.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexRefreshCommandImpl.java
@@ -85,118 +85,26 @@ public class IndexRefreshCommandImpl implements IndexRefreshCommand {
final long start = System.currentTimeMillis();
-
- //id to hunt for
- final UUID uuid = UUIDUtils.newTimeUUID();
- final Entity entity = new Entity( new SimpleId( uuid, "ug_refresh_index_type" ) );
- EntityUtils.setVersion( entity, UUIDGenerator.newTimeUUID() );
- final Id appId = new SimpleId( "ug_refresh_index" );
- final ApplicationScope appScope = new ApplicationScopeImpl( appId );
- final IndexEdge edge = new IndexEdgeImpl( appId, "refresh", SearchEdge.NodeType.SOURCE, uuid.timestamp() );
- final String docId = IndexingUtils.createIndexDocId( appScope, entity, edge );
- final Map<String, Object> entityData = EntityToMapConverter.convert( appScope, edge, entity );
- final String entityId = entityData.get( IndexingUtils.ENTITY_ID_FIELDNAME ).toString();
- //add a tracer record
- IndexOperation indexRequest = new IndexOperation( alias.getWriteAlias(), docId, entityData );
- //save the item
- final IndexOperationMessage message = new IndexOperationMessage();
- message.addIndexRequest( indexRequest );
-
- //add the record to the index
- final Observable<IndexOperationMessage> addRecord = producer.put( message );
-
- //refresh the index
- // final Observable<Boolean> refresh = refresh( indexes );
-
- /**
- * We have to search. Get by ID returns immediately, even if search isn't ready, therefore we have to search
- */
- //set our filter for entityId fieldname
-
-
- /**
- * We want to search once we've added our record, then refreshed
- */
- final Observable<IndexRefreshCommandInfo> searchObservable =
- Observable.create(sub -> {
- try {
- IndexRefreshCommandInfo info = null;
- for(int i = 0; i<indexFig.maxRefreshSearches();i++) {
- final SearchRequestBuilder builder = esProvider.getClient().prepareSearch(alias.getReadAlias())
- .setTypes(IndexingUtils.ES_ENTITY_TYPE)
- .setPostFilter(FilterBuilders
- .termFilter(IndexingUtils.ENTITY_ID_FIELDNAME,
- entityId));
-
- info = new IndexRefreshCommandInfo(builder.execute().get().getHits().totalHits() > 0,
- System.currentTimeMillis() - start);
- if(info.hasFinished()){
- break;
- }else {
- Thread.sleep(50);
- }
- }
- sub.onNext(info);
- sub.onCompleted();
- } catch (Exception ee) {
- logger.error("Failed during refresh search for " + uuid, ee);
- throw new RuntimeException("Failed during refresh search for " + uuid, ee);
- }
- });
-
-
- //chain it all together
-
- //add the record, take it's last result. On the last add, we then execute the refresh command
-
- final Observable<IndexRefreshCommandInfo> refreshResults = addRecord
-
- //after our add, run a refresh
- .doOnNext( addResult -> {
-
-
- if ( indexes.length == 0 ) {
- logger.debug( "Not refreshing indexes. none found" );
- }
- //Added For Graphite Metrics
- RefreshResponse response =
- esProvider.getClient().admin().indices().prepareRefresh( indexes ).execute().actionGet();
- int failedShards = response.getFailedShards();
- int successfulShards = response.getSuccessfulShards();
- ShardOperationFailedException[] sfes = response.getShardFailures();
- if ( sfes != null ) {
- for ( ShardOperationFailedException sfe : sfes ) {
- logger.error( "Failed to refresh index:{} reason:{}", sfe.index(), sfe.reason() );
- }
- }
- logger.debug( "Refreshed indexes: {},success:{} failed:{} ", StringUtils.join( indexes, ", " ),
- successfulShards, failedShards);
- })
-
- //once the refresh is done execute the search
- .flatMap(refreshCommandResult -> searchObservable)
-
- //check when found
- .doOnNext(found -> {
- if (!found.hasFinished()) {
- logger.error("Couldn't find record during refresh uuid: {} took ms:{} ", uuid,
- found.getExecutionTime());
- } else {
- logger.info("found record during refresh uuid: {} took ms:{} ", uuid, found.getExecutionTime());
- }
- }).doOnCompleted(() -> {
- //clean up our data
- String[] aliases = indexCache.getIndexes(alias, EntityIndex.AliasType.Read);
- DeIndexOperation deIndexRequest =
- new DeIndexOperation(aliases, appScope, edge, entity.getId(), entity.getVersion());
-
- //delete the item
- IndexOperationMessage indexOperationMessage = new IndexOperationMessage();
- indexOperationMessage.addDeIndexRequest( deIndexRequest );
- producer.put( indexOperationMessage ).subscribe();
- } );
-
-
- return ObservableTimer.time( refreshResults, timer ) ;
+ if (indexes.length == 0) {
+ logger.debug("Not refreshing indexes. none found");
+ }
+ //Added For Graphite Metrics
+ RefreshResponse response =
+ esProvider.getClient().admin().indices().prepareRefresh(indexes).execute().actionGet();
+ int failedShards = response.getFailedShards();
+ int successfulShards = response.getSuccessfulShards();
+ ShardOperationFailedException[] sfes = response.getShardFailures();
+ if (sfes != null) {
+ for (ShardOperationFailedException sfe : sfes) {
+ logger.error("Failed to refresh index:{} reason:{}", sfe.index(), sfe.reason());
+ }
+ }
+ logger.debug("Refreshed indexes: {},success:{} failed:{} ", StringUtils.join(indexes, ", "),
+ successfulShards, failedShards);
+
+ IndexRefreshCommandInfo refreshResults = new IndexRefreshCommandInfo(failedShards == 0,
+ System.currentTimeMillis() - start);
+
+ return ObservableTimer.time(Observable.just(refreshResults), timer);
}
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/f8614a68/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
index f36d3c1..edd3b6b 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
@@ -37,7 +37,7 @@ public class DefaultQueueManager implements QueueManager {
@Override
public synchronized Observable<QueueMessage> getMessages(int limit, int transactionTimeout, int waitTime, Class klass) {
List<QueueMessage> returnQueue = new ArrayList<>();
- queue.drainTo(returnQueue);
+ queue.drainTo(returnQueue,1);
return Observable.from( returnQueue);
}
[13/23] usergrid git commit: Merge branch '2.1-release' of
https://git-wip-us.apache.org/repos/asf/usergrid into
remove-inmemory-event-service
Posted by sf...@apache.org.
Merge branch '2.1-release' of https://git-wip-us.apache.org/repos/asf/usergrid into remove-inmemory-event-service
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/9673169d
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/9673169d
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/9673169d
Branch: refs/heads/master
Commit: 9673169dc7c28167ccf391496afd7faeaaffd321
Parents: b6f8e71 2b22c61
Author: Shawn Feldman <sf...@apache.org>
Authored: Fri Oct 16 15:08:18 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Fri Oct 16 15:08:18 2015 -0600
----------------------------------------------------------------------
.../usergrid/corepersistence/index/IndexProcessorFig.java | 7 +++----
.../java/org/apache/usergrid/persistence/queue/QueueFig.java | 6 +++---
.../usergrid/persistence/queue/impl/SNSQueueManagerImpl.java | 2 +-
.../persistence/queue/util/AmazonNotificationUtils.java | 4 +++-
4 files changed, 10 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/9673169d/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexProcessorFig.java
----------------------------------------------------------------------
[18/23] usergrid git commit: merge
Posted by sf...@apache.org.
merge
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/5ecc57f3
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/5ecc57f3
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/5ecc57f3
Branch: refs/heads/master
Commit: 5ecc57f388794decd53e25a404c5706eeda2baa9
Parents: efffe47 a09485a
Author: Shawn Feldman <sf...@apache.org>
Authored: Mon Oct 19 13:20:51 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Mon Oct 19 13:20:51 2015 -0600
----------------------------------------------------------------------
.../asyncevents/AmazonAsyncEventService.java | 40 ++---
.../asyncevents/AsyncIndexProvider.java | 9 +-
.../asyncevents/model/AsyncEvent.java | 14 +-
.../asyncevents/model/EdgeDeleteEvent.java | 6 +-
.../asyncevents/model/EdgeIndexEvent.java | 9 +-
.../asyncevents/model/EntityDeleteEvent.java | 8 +-
.../asyncevents/model/EntityIndexEvent.java | 6 +-
.../model/InitializeApplicationIndexEvent.java | 4 +-
.../index/AmazonAsyncEventServiceTest.java | 6 +-
.../cache/CachedEntityCollectionManager.java | 147 -------------------
.../EntityCollectionManagerFactoryImpl.java | 6 -
.../usergrid/persistence/queue/QueueFig.java | 2 +-
.../queue/impl/SNSQueueManagerImpl.java | 6 +-
.../queue/impl/SQSQueueManagerImpl.java | 2 +-
14 files changed, 64 insertions(+), 201 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/5ecc57f3/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/5ecc57f3/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncIndexProvider.java
----------------------------------------------------------------------
diff --cc stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncIndexProvider.java
index f5197d2,8b44714..fb63276
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncIndexProvider.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncIndexProvider.java
@@@ -27,7 -27,7 +27,8 @@@ import org.apache.usergrid.persistence.
import org.apache.usergrid.persistence.core.metrics.MetricsFactory;
import org.apache.usergrid.persistence.index.EntityIndexFactory;
import org.apache.usergrid.persistence.index.impl.IndexProducer;
+import org.apache.usergrid.persistence.queue.LocalQueueManager;
+ import org.apache.usergrid.persistence.queue.QueueFig;
import org.apache.usergrid.persistence.queue.QueueManagerFactory;
import com.google.inject.Inject;
@@@ -96,16 -98,13 +99,16 @@@ public class AsyncIndexProvider impleme
switch (impl) {
case LOCAL:
- return new InMemoryAsyncEventService(eventBuilder, rxTaskScheduler, indexProducer,indexProcessorFig.resolveSynchronously());
+ AmazonAsyncEventService eventService = new AmazonAsyncEventService(scope -> new LocalQueueManager(), indexProcessorFig, indexProducer, metricsFactory,
+ entityCollectionManagerFactory, indexLocationStrategyFactory, entityIndexFactory, eventBuilder, rxTaskScheduler);
+ eventService.MAX_TAKE = 1000;
+ return eventService;
case SQS:
return new AmazonAsyncEventService(queueManagerFactory, indexProcessorFig, indexProducer, metricsFactory,
- entityCollectionManagerFactory, indexLocationStrategyFactory,entityIndexFactory, eventBuilder, rxTaskScheduler );
+ entityCollectionManagerFactory, indexLocationStrategyFactory,entityIndexFactory, eventBuilder, rxTaskScheduler,queueFig );
case SNS:
return new AmazonAsyncEventService(queueManagerFactory, indexProcessorFig, indexProducer, metricsFactory,
- entityCollectionManagerFactory, indexLocationStrategyFactory,entityIndexFactory, eventBuilder, rxTaskScheduler );
+ entityCollectionManagerFactory, indexLocationStrategyFactory,entityIndexFactory, eventBuilder, rxTaskScheduler, queueFig);
default:
throw new IllegalArgumentException("Configuration value of " + getErrorValues() + " are allowed");
}
[17/23] usergrid git commit: remove excess interfaces
Posted by sf...@apache.org.
remove excess interfaces
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/efffe474
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/efffe474
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/efffe474
Branch: refs/heads/master
Commit: efffe474bdca1223222f955204301f604dea40b8
Parents: e5e9d02
Author: Shawn Feldman <sf...@apache.org>
Authored: Mon Oct 19 11:38:25 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Mon Oct 19 11:38:25 2015 -0600
----------------------------------------------------------------------
.../corepersistence/CpEntityManager.java | 7 +-
.../corepersistence/CpEntityManagerFactory.java | 3 +-
.../asyncevents/AsyncIndexProvider.java | 7 +-
.../usergrid/persistence/EntityManager.java | 5 +-
.../persistence/EntityManagerFactory.java | 4 +-
.../corepersistence/StaleIndexCleanupTest.java | 5 +-
.../persistence/query/IteratingQueryIT.java | 1 +
.../usergrid/persistence/index/EntityIndex.java | 20 +++-
.../persistence/index/IndexRefreshCommand.java | 50 ----------
.../persistence/index/guice/IndexModule.java | 1 -
.../index/impl/EsEntityIndexFactoryImpl.java | 6 +-
.../index/impl/EsEntityIndexImpl.java | 38 ++++++--
.../index/impl/IndexRefreshCommandImpl.java | 97 -------------------
.../persistence/index/impl/EntityIndexTest.java | 2 +-
.../persistence/queue/DefaultQueueManager.java | 98 --------------------
.../persistence/queue/LocalQueueManager.java | 98 ++++++++++++++++++++
.../queue/impl/QueueManagerFactoryImpl.java | 4 +-
.../apns/NotificationsServiceIT.java | 4 +-
.../gcm/NotificationsServiceIT.java | 2 -
19 files changed, 170 insertions(+), 282 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/efffe474/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 addd6ef..4b32389 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
@@ -40,7 +40,6 @@ import org.apache.usergrid.corepersistence.service.CollectionService;
import org.apache.usergrid.corepersistence.service.ConnectionService;
import org.apache.usergrid.persistence.index.EntityIndex;
import org.apache.usergrid.persistence.index.IndexLocationStrategy;
-import org.apache.usergrid.persistence.index.IndexRefreshCommand;
import org.apache.usergrid.persistence.index.utils.*;
import org.apache.usergrid.utils.*;
import org.apache.usergrid.utils.ClassUtils;
@@ -2877,7 +2876,7 @@ public class CpEntityManager implements EntityManager {
/**
* TODO, these 3 methods are super janky. During refactoring we should clean this model up
*/
- public IndexRefreshCommand.IndexRefreshCommandInfo refreshIndex() {
+ public EntityIndex.IndexRefreshCommandInfo refreshIndex() {
try {
long start = System.currentTimeMillis();
// refresh special indexes without calling EntityManager refresh because stack overflow
@@ -2885,7 +2884,7 @@ public class CpEntityManager implements EntityManager {
map.put("some prop", "test");
boolean hasFinished = false;
Entity refreshEntity = create("refresh", map);
- IndexRefreshCommand.IndexRefreshCommandInfo indexRefreshCommandInfo
+ EntityIndex.IndexRefreshCommandInfo indexRefreshCommandInfo
= managerCache.getEntityIndex(applicationScope).refreshAsync().toBlocking().first();
try {
for (int i = 0; i < 10; i++) {
@@ -2899,6 +2898,7 @@ public class CpEntityManager implements EntityManager {
break;
}
Thread.sleep(200);
+
indexRefreshCommandInfo
= managerCache.getEntityIndex(applicationScope).refreshAsync().toBlocking().first();
}
@@ -2908,6 +2908,7 @@ public class CpEntityManager implements EntityManager {
}finally {
delete(refreshEntity);
}
+ Thread.sleep(200);
return indexRefreshCommandInfo;
} catch (Exception e) {
http://git-wip-us.apache.org/repos/asf/usergrid/blob/efffe474/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 4bdade5..8d890e6 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
@@ -66,7 +66,6 @@ import org.apache.usergrid.persistence.graph.GraphManagerFactory;
import org.apache.usergrid.persistence.graph.SearchByEdgeType;
import org.apache.usergrid.persistence.graph.impl.SimpleSearchByEdgeType;
import org.apache.usergrid.persistence.index.EntityIndex;
-import org.apache.usergrid.persistence.index.IndexRefreshCommand;
import org.apache.usergrid.persistence.model.entity.Id;
import org.apache.usergrid.persistence.model.entity.SimpleId;
import org.apache.usergrid.persistence.model.util.UUIDGenerator;
@@ -689,7 +688,7 @@ public class CpEntityManagerFactory implements EntityManagerFactory, Application
/**
* TODO, these 3 methods are super janky. During refactoring we should clean this model up
*/
- public IndexRefreshCommand.IndexRefreshCommandInfo refreshIndex(UUID applicationId) {
+ public EntityIndex.IndexRefreshCommandInfo refreshIndex(UUID applicationId) {
return getEntityManager(applicationId).refreshIndex();
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/efffe474/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncIndexProvider.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncIndexProvider.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncIndexProvider.java
index 0677aaf..f5197d2 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncIndexProvider.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncIndexProvider.java
@@ -22,20 +22,17 @@ package org.apache.usergrid.corepersistence.asyncevents;
import org.apache.usergrid.corepersistence.index.IndexLocationStrategyFactory;
import org.apache.usergrid.corepersistence.index.IndexProcessorFig;
-import org.apache.usergrid.corepersistence.index.IndexService;
import org.apache.usergrid.persistence.collection.EntityCollectionManagerFactory;
import org.apache.usergrid.persistence.core.rx.RxTaskScheduler;
import org.apache.usergrid.persistence.core.metrics.MetricsFactory;
import org.apache.usergrid.persistence.index.EntityIndexFactory;
import org.apache.usergrid.persistence.index.impl.IndexProducer;
-import org.apache.usergrid.persistence.queue.DefaultQueueManager;
-import org.apache.usergrid.persistence.queue.QueueManager;
+import org.apache.usergrid.persistence.queue.LocalQueueManager;
import org.apache.usergrid.persistence.queue.QueueManagerFactory;
import com.google.inject.Inject;
import com.google.inject.Provider;
import com.google.inject.Singleton;
-import org.apache.usergrid.persistence.queue.QueueScope;
/**
@@ -99,7 +96,7 @@ public class AsyncIndexProvider implements Provider<AsyncEventService> {
switch (impl) {
case LOCAL:
- AmazonAsyncEventService eventService = new AmazonAsyncEventService(scope -> new DefaultQueueManager(), indexProcessorFig, indexProducer, metricsFactory,
+ AmazonAsyncEventService eventService = new AmazonAsyncEventService(scope -> new LocalQueueManager(), indexProcessorFig, indexProducer, metricsFactory,
entityCollectionManagerFactory, indexLocationStrategyFactory, entityIndexFactory, eventBuilder, rxTaskScheduler);
eventService.MAX_TAKE = 1000;
return eventService;
http://git-wip-us.apache.org/repos/asf/usergrid/blob/efffe474/stack/core/src/main/java/org/apache/usergrid/persistence/EntityManager.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/persistence/EntityManager.java b/stack/core/src/main/java/org/apache/usergrid/persistence/EntityManager.java
index 9c4a929..20e8d23 100644
--- a/stack/core/src/main/java/org/apache/usergrid/persistence/EntityManager.java
+++ b/stack/core/src/main/java/org/apache/usergrid/persistence/EntityManager.java
@@ -29,7 +29,7 @@ import org.apache.usergrid.persistence.Query.Level;
import org.apache.usergrid.persistence.cassandra.CassandraService;
import org.apache.usergrid.persistence.entities.Application;
import org.apache.usergrid.persistence.entities.Role;
-import org.apache.usergrid.persistence.index.IndexRefreshCommand;
+import org.apache.usergrid.persistence.index.EntityIndex;
import org.apache.usergrid.persistence.index.query.CounterResolution;
import org.apache.usergrid.persistence.index.query.Identifier;
import org.apache.usergrid.persistence.model.entity.Id;
@@ -734,6 +734,5 @@ public interface EntityManager {
* init the index
*/
void initializeIndex();
-
- IndexRefreshCommand.IndexRefreshCommandInfo refreshIndex();
+ EntityIndex.IndexRefreshCommandInfo refreshIndex();
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/efffe474/stack/core/src/main/java/org/apache/usergrid/persistence/EntityManagerFactory.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/persistence/EntityManagerFactory.java b/stack/core/src/main/java/org/apache/usergrid/persistence/EntityManagerFactory.java
index bde1bd2..cb44d4d 100644
--- a/stack/core/src/main/java/org/apache/usergrid/persistence/EntityManagerFactory.java
+++ b/stack/core/src/main/java/org/apache/usergrid/persistence/EntityManagerFactory.java
@@ -25,7 +25,7 @@ import com.google.common.base.Optional;
import org.apache.usergrid.corepersistence.index.ReIndexService;
import org.apache.usergrid.persistence.core.util.Health;
-import org.apache.usergrid.persistence.index.IndexRefreshCommand;
+import org.apache.usergrid.persistence.index.EntityIndex;
import rx.Observable;
@@ -166,7 +166,7 @@ public interface EntityManagerFactory {
public UUID getManagementAppId();
- public IndexRefreshCommand.IndexRefreshCommandInfo refreshIndex(UUID applicationId);
+ public EntityIndex.IndexRefreshCommandInfo refreshIndex(UUID applicationId);
/**
* Perform a realtime count of every entity in the system. This can be slow as it traverses the entire system graph
http://git-wip-us.apache.org/repos/asf/usergrid/blob/efffe474/stack/core/src/test/java/org/apache/usergrid/corepersistence/StaleIndexCleanupTest.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/corepersistence/StaleIndexCleanupTest.java b/stack/core/src/test/java/org/apache/usergrid/corepersistence/StaleIndexCleanupTest.java
index 16d9a5b..5e0e062 100644
--- a/stack/core/src/test/java/org/apache/usergrid/corepersistence/StaleIndexCleanupTest.java
+++ b/stack/core/src/test/java/org/apache/usergrid/corepersistence/StaleIndexCleanupTest.java
@@ -364,10 +364,13 @@ public class StaleIndexCleanupTest extends AbstractCoreIT {
// query Core Persistence directly for total number of result candidates
for(int i = 0;i<10;i++){
+
+ crs = queryCollectionCp("things", "thing", "select *");
if(numEntities * (numUpdates + 1) == crs.size()){
break;
+ }else{
+ Thread.sleep(1100);
}
- crs = queryCollectionCp("things", "thing", "select *");
}
// Assert.assertEquals("Expect stale candidates", numEntities * (numUpdates + 1), crs.size());
http://git-wip-us.apache.org/repos/asf/usergrid/blob/efffe474/stack/core/src/test/java/org/apache/usergrid/persistence/query/IteratingQueryIT.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/persistence/query/IteratingQueryIT.java b/stack/core/src/test/java/org/apache/usergrid/persistence/query/IteratingQueryIT.java
index e589be4..3ff48b6 100644
--- a/stack/core/src/test/java/org/apache/usergrid/persistence/query/IteratingQueryIT.java
+++ b/stack/core/src/test/java/org/apache/usergrid/persistence/query/IteratingQueryIT.java
@@ -926,6 +926,7 @@ public class IteratingQueryIT {
}
app.refreshIndex();
+ Thread.sleep(500);
long stop = System.currentTimeMillis();
LOG.info( "Writes took {} ms", stop - start );
http://git-wip-us.apache.org/repos/asf/usergrid/blob/efffe474/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/EntityIndex.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/EntityIndex.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/EntityIndex.java
index 72300a5..81f900c 100644
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/EntityIndex.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/EntityIndex.java
@@ -55,7 +55,7 @@ public interface EntityIndex extends CPManager {
/**
* Refresh the index.
*/
- Observable<IndexRefreshCommand.IndexRefreshCommandInfo> refreshAsync();
+ Observable<IndexRefreshCommandInfo> refreshAsync();
/**
@@ -71,6 +71,7 @@ public interface EntityIndex extends CPManager {
/**
* get total entity size by an edge -> "term":{"edgeName":"zzzcollzzz|roles"}
+ *
* @param edge
* @return
*/
@@ -149,6 +150,23 @@ public interface EntityIndex extends CPManager {
enum AliasType {
Read, Write
}
+ class IndexRefreshCommandInfo{
+ private final boolean hasFinished;
+ private final long executionTime;
+
+ public IndexRefreshCommandInfo(boolean hasFinished, long executionTime){
+ this.hasFinished = hasFinished;
+ this.executionTime = executionTime;
+ }
+
+ public boolean hasFinished() {
+ return hasFinished;
+ }
+
+ public long getExecutionTime() {
+ return executionTime;
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/efffe474/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/IndexRefreshCommand.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/IndexRefreshCommand.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/IndexRefreshCommand.java
deleted file mode 100644
index 2226d89..0000000
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/IndexRefreshCommand.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. The ASF licenses this file to You
- * under the Apache License, Version 2.0 (the "License"); you may not
- * use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License. For additional information regarding
- * copyright in this work, please see the NOTICE file in the top level
- * directory of this distribution.
- *
- */
-package org.apache.usergrid.persistence.index;
-
-import rx.Observable;
-
-/**
- * Classy class class.
- */
-public interface IndexRefreshCommand {
-
- Observable<IndexRefreshCommandInfo> execute(IndexAlias alias, String[] indexes);
-
- class IndexRefreshCommandInfo{
- private final boolean hasFinished;
- private final long executionTime;
-
- public IndexRefreshCommandInfo(boolean hasFinished, long executionTime){
- this.hasFinished = hasFinished;
- this.executionTime = executionTime;
- }
-
- public boolean hasFinished() {
- return hasFinished;
- }
-
- public long getExecutionTime() {
- return executionTime;
- }
- }
-}
-
-
http://git-wip-us.apache.org/repos/asf/usergrid/blob/efffe474/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/guice/IndexModule.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/guice/IndexModule.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/guice/IndexModule.java
index 46559ad..b828934 100644
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/guice/IndexModule.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/guice/IndexModule.java
@@ -50,7 +50,6 @@ public abstract class IndexModule extends AbstractModule {
bind( EntityIndexFactory.class ).to( EsEntityIndexFactoryImpl.class );
bind(IndexCache.class).to(EsIndexCacheImpl.class);
- bind(IndexRefreshCommand.class).to(IndexRefreshCommandImpl.class);
bind(IndexProducer.class).to(EsIndexProducerImpl.class).asEagerSingleton();
http://git-wip-us.apache.org/repos/asf/usergrid/blob/efffe474/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexFactoryImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexFactoryImpl.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexFactoryImpl.java
index c91057b..f4e89ad 100644
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexFactoryImpl.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexFactoryImpl.java
@@ -38,7 +38,6 @@ public class EsEntityIndexFactoryImpl implements EntityIndexFactory{
private final IndexCache indexCache;
private final EsProvider provider;
private final MetricsFactory metricsFactory;
- private final IndexRefreshCommand refreshCommand;
private LoadingCache<IndexLocationStrategy, EntityIndex> eiCache =
CacheBuilder.newBuilder().maximumSize( 1000 ).build( new CacheLoader<IndexLocationStrategy, EntityIndex>() {
@@ -47,7 +46,6 @@ public class EsEntityIndexFactoryImpl implements EntityIndexFactory{
provider,
indexCache,
config,
- refreshCommand,
metricsFactory,
locationStrategy
);
@@ -60,15 +58,13 @@ public class EsEntityIndexFactoryImpl implements EntityIndexFactory{
public EsEntityIndexFactoryImpl( final IndexFig indexFig,
final IndexCache indexCache,
final EsProvider provider,
- final MetricsFactory metricsFactory,
- final IndexRefreshCommand refreshCommand
+ final MetricsFactory metricsFactory
){
this.config = indexFig;
this.indexCache = indexCache;
this.provider = provider;
this.metricsFactory = metricsFactory;
- this.refreshCommand = refreshCommand;
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/efffe474/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexImpl.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexImpl.java
index f6ebce2..5f2a547 100644
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexImpl.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexImpl.java
@@ -31,6 +31,7 @@ import org.apache.usergrid.persistence.core.metrics.ObservableTimer;
import org.apache.usergrid.persistence.core.migration.data.VersionedData;
import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.core.util.Health;
+import org.apache.usergrid.persistence.core.util.StringUtils;
import org.apache.usergrid.persistence.core.util.ValidationUtils;
import org.apache.usergrid.persistence.index.*;
import org.apache.usergrid.persistence.index.ElasticSearchQueryBuilder.SearchRequestBuilderStrategyV2;
@@ -50,6 +51,7 @@ import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequestBuilder;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingResponse;
+import org.elasticsearch.action.admin.indices.refresh.RefreshResponse;
import org.elasticsearch.action.deletebyquery.DeleteByQueryResponse;
import org.elasticsearch.action.deletebyquery.IndexDeleteByQueryResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
@@ -98,7 +100,6 @@ public class EsEntityIndexImpl implements EntityIndex,VersionedData {
* We purposefully make this per instance. Some indexes may work, while others may fail
*/
private final EsProvider esProvider;
- private final IndexRefreshCommand indexRefreshCommand;
//number of times to wait for the index to refresh properly.
private static final int MAX_WAITS = 10;
@@ -118,6 +119,7 @@ public class EsEntityIndexImpl implements EntityIndex,VersionedData {
private final long queryTimeout;
private final FailureMonitorImpl failureMonitor;
private final Timer aggregationTimer;
+ private final Timer refreshTimer;
private IndexCache aliasCache;
private Timer mappingTimer;
@@ -128,7 +130,6 @@ public class EsEntityIndexImpl implements EntityIndex,VersionedData {
public EsEntityIndexImpl( final EsProvider provider,
final IndexCache indexCache,
final IndexFig indexFig,
- final IndexRefreshCommand indexRefreshCommand,
final MetricsFactory metricsFactory,
final IndexLocationStrategy indexLocationStrategy
) {
@@ -137,7 +138,6 @@ public class EsEntityIndexImpl implements EntityIndex,VersionedData {
this.indexLocationStrategy = indexLocationStrategy;
this.failureMonitor = new FailureMonitorImpl( indexFig, provider );
this.esProvider = provider;
- this.indexRefreshCommand = indexRefreshCommand;
this.alias = indexLocationStrategy.getAlias();
this.aliasCache = indexCache;
this.applicationScope = indexLocationStrategy.getApplicationScope();
@@ -153,6 +153,7 @@ public class EsEntityIndexImpl implements EntityIndex,VersionedData {
this.refreshIndexMeter = metricsFactory.getMeter(EsEntityIndexImpl.class, "index.refresh_index");
this.searchTimer = metricsFactory.getTimer(EsEntityIndexImpl.class, "search");
this.aggregationTimer = metricsFactory.getTimer( EsEntityIndexImpl.class, "aggregations" );
+ this.refreshTimer = metricsFactory.getTimer( EsEntityIndexImpl.class, "index.refresh" );
}
@@ -350,11 +351,34 @@ public class EsEntityIndexImpl implements EntityIndex,VersionedData {
-
- public Observable<IndexRefreshCommand.IndexRefreshCommandInfo> refreshAsync() {
+ public Observable<IndexRefreshCommandInfo> refreshAsync() {
refreshIndexMeter.mark();
- return indexRefreshCommand.execute(alias, getIndexes());
+ final long start = System.currentTimeMillis();
+
+ String[] indexes = getIndexes();
+ if (indexes.length == 0) {
+ logger.debug("Not refreshing indexes. none found");
+ }
+ //Added For Graphite Metrics
+ RefreshResponse response =
+ esProvider.getClient().admin().indices().prepareRefresh(indexes).execute().actionGet();
+ int failedShards = response.getFailedShards();
+ int successfulShards = response.getSuccessfulShards();
+ ShardOperationFailedException[] sfes = response.getShardFailures();
+ if (sfes != null) {
+ for (ShardOperationFailedException sfe : sfes) {
+ logger.error("Failed to refresh index:{} reason:{}", sfe.index(), sfe.reason());
+ }
+ }
+ logger.debug("Refreshed indexes: {},success:{} failed:{} ", StringUtils.join(indexes, ", "),
+ successfulShards, failedShards);
+
+ IndexRefreshCommandInfo refreshResults = new IndexRefreshCommandInfo(failedShards == 0,
+ System.currentTimeMillis() - start);
+
+
+ return ObservableTimer.time(Observable.just(refreshResults), refreshTimer);
}
@@ -782,4 +806,6 @@ public class EsEntityIndexImpl implements EntityIndex,VersionedData {
}
+
+
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/efffe474/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexRefreshCommandImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexRefreshCommandImpl.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexRefreshCommandImpl.java
deleted file mode 100644
index bd67fb6..0000000
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexRefreshCommandImpl.java
+++ /dev/null
@@ -1,97 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. The ASF licenses this file to You
- * under the Apache License, Version 2.0 (the "License"); you may not
- * use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License. For additional information regarding
- * copyright in this work, please see the NOTICE file in the top level
- * directory of this distribution.
- *
- */
-package org.apache.usergrid.persistence.index.impl;
-
-
-import java.util.Map;
-import java.util.UUID;
-
-import org.apache.usergrid.persistence.index.*;
-import org.elasticsearch.action.ShardOperationFailedException;
-import org.elasticsearch.action.admin.indices.refresh.RefreshResponse;
-import org.elasticsearch.action.search.SearchRequestBuilder;
-import org.elasticsearch.index.query.FilterBuilders;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.usergrid.persistence.core.metrics.MetricsFactory;
-import org.apache.usergrid.persistence.core.metrics.ObservableTimer;
-import org.apache.usergrid.persistence.core.scope.ApplicationScope;
-import org.apache.usergrid.persistence.core.scope.ApplicationScopeImpl;
-import org.apache.usergrid.persistence.core.util.StringUtils;
-import org.apache.usergrid.persistence.index.utils.UUIDUtils;
-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.util.EntityUtils;
-import org.apache.usergrid.persistence.model.util.UUIDGenerator;
-
-import com.codahale.metrics.Timer;
-import com.google.inject.Inject;
-
-import rx.Observable;
-
-
-/**
- * Classy class class.
- */
-public class IndexRefreshCommandImpl implements IndexRefreshCommand {
- private static final Logger logger = LoggerFactory.getLogger( IndexRefreshCommandImpl.class );
-
- private final EsProvider esProvider;
- private final Timer timer;
-
-
- @Inject
- public IndexRefreshCommandImpl( final EsProvider esProvider,final MetricsFactory metricsFactory) {
-
- this.timer = metricsFactory.getTimer( IndexRefreshCommandImpl.class, "index.refresh" );
- this.esProvider = esProvider;
- }
-
-
- @Override
- public Observable<IndexRefreshCommandInfo> execute(IndexAlias alias, String[] indexes ) {
-
- final long start = System.currentTimeMillis();
-
- if (indexes.length == 0) {
- logger.debug("Not refreshing indexes. none found");
- }
- //Added For Graphite Metrics
- RefreshResponse response =
- esProvider.getClient().admin().indices().prepareRefresh(indexes).execute().actionGet();
- int failedShards = response.getFailedShards();
- int successfulShards = response.getSuccessfulShards();
- ShardOperationFailedException[] sfes = response.getShardFailures();
- if (sfes != null) {
- for (ShardOperationFailedException sfe : sfes) {
- logger.error("Failed to refresh index:{} reason:{}", sfe.index(), sfe.reason());
- }
- }
- logger.debug("Refreshed indexes: {},success:{} failed:{} ", StringUtils.join(indexes, ", "),
- successfulShards, failedShards);
-
- IndexRefreshCommandInfo refreshResults = new IndexRefreshCommandInfo(failedShards == 0,
- System.currentTimeMillis() - start);
-
- return ObservableTimer.time(Observable.just(refreshResults), timer);
- }
-}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/efffe474/stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/index/impl/EntityIndexTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/index/impl/EntityIndexTest.java b/stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/index/impl/EntityIndexTest.java
index 3a91f12..8b51748 100644
--- a/stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/index/impl/EntityIndexTest.java
+++ b/stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/index/impl/EntityIndexTest.java
@@ -310,7 +310,7 @@ public class EntityIndexTest extends BaseIT {
EntityIndexBatch batch = entityIndex.createBatch();
insertJsonBlob(sampleJson, batch, entityType, indexEdge, max, startIndex);
indexProducer.put(batch.build()).subscribe();;
- IndexRefreshCommandImpl.IndexRefreshCommandInfo info = entityIndex.refreshAsync().toBlocking().first();
+ EntityIndex.IndexRefreshCommandInfo info = entityIndex.refreshAsync().toBlocking().first();
long time = info.getExecutionTime();
log.info("refresh took ms:" + time);
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/efffe474/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
deleted file mode 100644
index 08f17a0..0000000
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
+++ /dev/null
@@ -1,98 +0,0 @@
-/*
- *
- * * Licensed to the Apache Software Foundation (ASF) under one or more
- * * contributor license agreements. The ASF licenses this file to You
- * * under the Apache License, Version 2.0 (the "License"); you may not
- * * use this file except in compliance with the License.
- * * You may obtain a copy of the License at
- * *
- * * http://www.apache.org/licenses/LICENSE-2.0
- * *
- * * Unless required by applicable law or agreed to in writing, software
- * * distributed under the License is distributed on an "AS IS" BASIS,
- * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * * See the License for the specific language governing permissions and
- * * limitations under the License. For additional information regarding
- * * copyright in this work, please see the NOTICE file in the top level
- * * directory of this distribution.
- *
- */
-
-package org.apache.usergrid.persistence.queue;
-
-import rx.Observable;
-
-import java.io.IOException;
-import java.util.AbstractQueue;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.UUID;
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.PriorityBlockingQueue;
-import java.util.concurrent.TimeUnit;
-
-/**
- * Default queue manager implementation, uses in memory linked queue
- */
-public class DefaultQueueManager implements QueueManager {
- public ArrayBlockingQueue<QueueMessage> queue = new ArrayBlockingQueue<>(10000);
-
- @Override
- public Observable<QueueMessage> getMessages(int limit, int transactionTimeout, int waitTime, Class klass) {
- List<QueueMessage> returnQueue = new ArrayList<>();
- try {
- QueueMessage message=null;
- int count = 5;
- do {
- message = queue.poll(100, TimeUnit.MILLISECONDS);
- if (message != null) {
- returnQueue.add(message);
- }
- }while(message!=null && count-->0);
- }catch (InterruptedException ie){
- throw new RuntimeException(ie);
- }
- return Observable.from( returnQueue);
- }
-
- @Override
- public long getQueueDepth() {
- return queue.size();
- }
-
- @Override
- public void commitMessage(QueueMessage queueMessage) {
- }
-
- @Override
- public void commitMessages(List<QueueMessage> queueMessages) {
- }
-
- @Override
- public void sendMessages(List bodies) throws IOException {
- for(Object body : bodies){
- String uuid = UUID.randomUUID().toString();
- try {
- queue.put(new QueueMessage(uuid, "handle_" + uuid, body, "put type here"));
- }catch (InterruptedException ie){
- throw new RuntimeException(ie);
- }
- }
- }
-
- @Override
- public void sendMessage(Object body) throws IOException {
- String uuid = UUID.randomUUID().toString();
- try {
- queue.put(new QueueMessage(uuid, "handle_" + uuid, body, "put type here"));
- }catch (InterruptedException ie){
- throw new RuntimeException(ie);
- }
- }
-
- @Override
- public void deleteQueue() {
-
- }
-}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/efffe474/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/LocalQueueManager.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/LocalQueueManager.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/LocalQueueManager.java
new file mode 100644
index 0000000..4b4218a
--- /dev/null
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/LocalQueueManager.java
@@ -0,0 +1,98 @@
+/*
+ *
+ * * Licensed to the Apache Software Foundation (ASF) under one or more
+ * * contributor license agreements. The ASF licenses this file to You
+ * * under the Apache License, Version 2.0 (the "License"); you may not
+ * * use this file except in compliance with the License.
+ * * You may obtain a copy of the License at
+ * *
+ * * http://www.apache.org/licenses/LICENSE-2.0
+ * *
+ * * Unless required by applicable law or agreed to in writing, software
+ * * distributed under the License is distributed on an "AS IS" BASIS,
+ * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * * See the License for the specific language governing permissions and
+ * * limitations under the License. For additional information regarding
+ * * copyright in this work, please see the NOTICE file in the top level
+ * * directory of this distribution.
+ *
+ */
+
+package org.apache.usergrid.persistence.queue;
+
+import rx.Observable;
+
+import java.io.IOException;
+import java.util.AbstractQueue;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.PriorityBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Default queue manager implementation, uses in memory linked queue
+ */
+public class LocalQueueManager implements QueueManager {
+ public ArrayBlockingQueue<QueueMessage> queue = new ArrayBlockingQueue<>(10000);
+
+ @Override
+ public Observable<QueueMessage> getMessages(int limit, int transactionTimeout, int waitTime, Class klass) {
+ List<QueueMessage> returnQueue = new ArrayList<>();
+ try {
+ QueueMessage message=null;
+ int count = 5;
+ do {
+ message = queue.poll(100, TimeUnit.MILLISECONDS);
+ if (message != null) {
+ returnQueue.add(message);
+ }
+ }while(message!=null && count-->0);
+ }catch (InterruptedException ie){
+ throw new RuntimeException(ie);
+ }
+ return Observable.from( returnQueue);
+ }
+
+ @Override
+ public long getQueueDepth() {
+ return queue.size();
+ }
+
+ @Override
+ public void commitMessage(QueueMessage queueMessage) {
+ }
+
+ @Override
+ public void commitMessages(List<QueueMessage> queueMessages) {
+ }
+
+ @Override
+ public void sendMessages(List bodies) throws IOException {
+ for(Object body : bodies){
+ String uuid = UUID.randomUUID().toString();
+ try {
+ queue.put(new QueueMessage(uuid, "handle_" + uuid, body, "put type here"));
+ }catch (InterruptedException ie){
+ throw new RuntimeException(ie);
+ }
+ }
+ }
+
+ @Override
+ public void sendMessage(Object body) throws IOException {
+ String uuid = UUID.randomUUID().toString();
+ try {
+ queue.put(new QueueMessage(uuid, "handle_" + uuid, body, "put type here"));
+ }catch (InterruptedException ie){
+ throw new RuntimeException(ie);
+ }
+ }
+
+ @Override
+ public void deleteQueue() {
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/efffe474/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/QueueManagerFactoryImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/QueueManagerFactoryImpl.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/QueueManagerFactoryImpl.java
index d0ed1ef..0f78678 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/QueueManagerFactoryImpl.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/QueueManagerFactoryImpl.java
@@ -19,9 +19,7 @@
*/
package org.apache.usergrid.persistence.queue.impl;
-import com.google.inject.Guice;
import com.google.inject.Inject;
-import com.google.inject.Provider;
import com.google.inject.Singleton;
import org.apache.usergrid.persistence.queue.*;
@@ -50,7 +48,7 @@ public class QueueManagerFactoryImpl implements QueueManagerFactory {
if(queueFig.overrideQueueForDefault()){
QueueManager manager = defaultManager.get(scope.getName());
if(manager==null){
- manager = new DefaultQueueManager();
+ manager = new LocalQueueManager();
defaultManager.put(scope.getName(),manager);
}
return manager;
http://git-wip-us.apache.org/repos/asf/usergrid/blob/efffe474/stack/services/src/test/java/org/apache/usergrid/services/notifications/apns/NotificationsServiceIT.java
----------------------------------------------------------------------
diff --git a/stack/services/src/test/java/org/apache/usergrid/services/notifications/apns/NotificationsServiceIT.java b/stack/services/src/test/java/org/apache/usergrid/services/notifications/apns/NotificationsServiceIT.java
index 9c1bcbe..8b94661 100644
--- a/stack/services/src/test/java/org/apache/usergrid/services/notifications/apns/NotificationsServiceIT.java
+++ b/stack/services/src/test/java/org/apache/usergrid/services/notifications/apns/NotificationsServiceIT.java
@@ -21,7 +21,7 @@ import org.apache.commons.io.IOUtils;
import org.apache.usergrid.persistence.*;
import org.apache.usergrid.persistence.entities.*;
import org.apache.usergrid.persistence.Query;
-import org.apache.usergrid.persistence.queue.DefaultQueueManager;
+import org.apache.usergrid.persistence.queue.LocalQueueManager;
import org.apache.usergrid.services.notifications.*;
import org.junit.*;
import org.slf4j.Logger;
@@ -119,7 +119,7 @@ public class NotificationsServiceIT extends AbstractServiceNotificationIT {
ns = getNotificationService();
- DefaultQueueManager qm = new DefaultQueueManager();
+ LocalQueueManager qm = new LocalQueueManager();
setup.getEntityIndex().refresh(app.getId());
http://git-wip-us.apache.org/repos/asf/usergrid/blob/efffe474/stack/services/src/test/java/org/apache/usergrid/services/notifications/gcm/NotificationsServiceIT.java
----------------------------------------------------------------------
diff --git a/stack/services/src/test/java/org/apache/usergrid/services/notifications/gcm/NotificationsServiceIT.java b/stack/services/src/test/java/org/apache/usergrid/services/notifications/gcm/NotificationsServiceIT.java
index 96653a5..3218137 100644
--- a/stack/services/src/test/java/org/apache/usergrid/services/notifications/gcm/NotificationsServiceIT.java
+++ b/stack/services/src/test/java/org/apache/usergrid/services/notifications/gcm/NotificationsServiceIT.java
@@ -18,10 +18,8 @@ package org.apache.usergrid.services.notifications.gcm;
import org.apache.usergrid.persistence.*;
import org.apache.usergrid.persistence.entities.*;
-import org.apache.usergrid.persistence.queue.DefaultQueueManager;
import org.apache.usergrid.services.notifications.*;
import org.junit.*;
-import org.junit.rules.TestName;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
[21/23] usergrid git commit: merge from 1048
Posted by sf...@apache.org.
merge from 1048
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/5eed63d4
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/5eed63d4
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/5eed63d4
Branch: refs/heads/master
Commit: 5eed63d43999c8f381e2fbcbe7d2c1fdb6b73729
Parents: 68befd2 4013f17
Author: Shawn Feldman <sf...@apache.org>
Authored: Mon Oct 19 15:19:33 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Mon Oct 19 15:19:33 2015 -0600
----------------------------------------------------------------------
.../asyncevents/AmazonAsyncEventService.java | 377 ++++++++----
.../asyncevents/AsyncEventService.java | 1 +
.../asyncevents/AsyncIndexProvider.java | 14 +-
.../asyncevents/model/AsyncEvent.java | 8 +-
.../model/ElasticsearchIndexEvent.java | 50 ++
.../index/IndexProcessorFig.java | 17 +
.../util/ObjectJsonSerializer.java | 92 +++
.../index/AmazonAsyncEventServiceTest.java | 6 +-
.../index/AsyncIndexServiceTest.java | 2 +-
.../usergrid/persistence/map/MapManager.java | 25 +-
.../persistence/map/impl/MapManagerImpl.java | 6 +
.../persistence/map/impl/MapSerialization.java | 27 +-
.../map/impl/MapSerializationImpl.java | 265 ++++----
.../index/impl/DeIndexOperation.java | 4 +
.../persistence/index/impl/IndexOperation.java | 4 +
.../index/impl/IndexOperationMessage.java | 5 +
.../persistence/queue/LocalQueueManager.java | 12 +-
.../persistence/queue/QueueManager.java | 8 +-
.../persistence/queue/guice/QueueModule.java | 1 -
.../queue/impl/SNSQueueManagerImpl.java | 615 +++++++++++--------
.../queue/impl/SQSQueueManagerImpl.java | 352 -----------
.../services/queues/ImportQueueManager.java | 9 +-
22 files changed, 1017 insertions(+), 883 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/5eed63d4/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/5eed63d4/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncIndexProvider.java
----------------------------------------------------------------------
diff --cc stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncIndexProvider.java
index 1ad9b86,2bace8d..aafb63a
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncIndexProvider.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncIndexProvider.java
@@@ -27,7 -27,7 +27,8 @@@ import org.apache.usergrid.persistence.
import org.apache.usergrid.persistence.core.metrics.MetricsFactory;
import org.apache.usergrid.persistence.index.EntityIndexFactory;
import org.apache.usergrid.persistence.index.impl.IndexProducer;
+import org.apache.usergrid.persistence.queue.LocalQueueManager;
+ import org.apache.usergrid.persistence.map.MapManagerFactory;
import org.apache.usergrid.persistence.queue.QueueFig;
import org.apache.usergrid.persistence.queue.QueueManagerFactory;
@@@ -99,16 -103,12 +104,15 @@@ public class AsyncIndexProvider impleme
switch (impl) {
case LOCAL:
- return new InMemoryAsyncEventService(eventBuilder, rxTaskScheduler, indexProducer,indexProcessorFig.resolveSynchronously());
+ AmazonAsyncEventService eventService = new AmazonAsyncEventService(scope -> new LocalQueueManager(), indexProcessorFig, indexProducer, metricsFactory,
- entityCollectionManagerFactory, indexLocationStrategyFactory, entityIndexFactory, eventBuilder, rxTaskScheduler,queueFig);
++ entityCollectionManagerFactory, indexLocationStrategyFactory, entityIndexFactory, eventBuilder,mapManagerFactory, queueFig,rxTaskScheduler);
+ eventService.MAX_TAKE = 1000;
+ return eventService;
case SQS:
- return new AmazonAsyncEventService(queueManagerFactory, indexProcessorFig, indexProducer, metricsFactory,
- entityCollectionManagerFactory, indexLocationStrategyFactory,entityIndexFactory, eventBuilder, rxTaskScheduler,queueFig );
+ throw new IllegalArgumentException("Configuration value of SQS is no longer allowed. Use SNS instead with only a single region");
case SNS:
return new AmazonAsyncEventService(queueManagerFactory, indexProcessorFig, indexProducer, metricsFactory,
- entityCollectionManagerFactory, indexLocationStrategyFactory,entityIndexFactory, eventBuilder, rxTaskScheduler, queueFig);
+ entityCollectionManagerFactory, indexLocationStrategyFactory,entityIndexFactory, eventBuilder, mapManagerFactory, queueFig, rxTaskScheduler );
default:
throw new IllegalArgumentException("Configuration value of " + getErrorValues() + " are allowed");
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/5eed63d4/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexProcessorFig.java
----------------------------------------------------------------------
diff --cc stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexProcessorFig.java
index 8a8c8a5,ec9b315..69d5e18
--- 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
@@@ -86,7 -86,32 +86,24 @@@ public interface IndexProcessorFig exte
String getQueueImplementation();
@Default("1000")
- @Key("elasticsearch.reindex.flush.interval")
- int getUpdateInterval();
-
- @Default("100")
- @Key("elasticsearch.buffer.time_ms")
- int getBufferTime();
-
- @Default("1000")
- @Key( REINDEX_BUFFER_SIZE )
+ @Key(REINDEX_BUFFER_SIZE)
int getReindexBufferSize();
+ /**
+ * Flag to resolve the LOCAL queue implementation service synchronously.
+ */
+ @Default("false")
+ @Key("elasticsearch.queue_impl.resolution")
+ boolean resolveSynchronously();
+
+ /**
+ * Get the message TTL in milliseconds. Defaults to 24 hours
+ *
+ * 24 * 60 * 60 * 1000
+ *
+ * @return
+ */
+ @Default("86400000")
+ @Key( "elasticsearch.message.ttl" )
+ int getIndexMessageTtl();
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/5eed63d4/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/LocalQueueManager.java
----------------------------------------------------------------------
diff --cc stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/LocalQueueManager.java
index 4b4218a,0000000..8be6099
mode 100644,000000..100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/LocalQueueManager.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/LocalQueueManager.java
@@@ -1,98 -1,0 +1,108 @@@
+/*
+ *
+ * * Licensed to the Apache Software Foundation (ASF) under one or more
+ * * contributor license agreements. The ASF licenses this file to You
+ * * under the Apache License, Version 2.0 (the "License"); you may not
+ * * use this file except in compliance with the License.
+ * * You may obtain a copy of the License at
+ * *
+ * * http://www.apache.org/licenses/LICENSE-2.0
+ * *
+ * * Unless required by applicable law or agreed to in writing, software
+ * * distributed under the License is distributed on an "AS IS" BASIS,
+ * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * * See the License for the specific language governing permissions and
+ * * limitations under the License. For additional information regarding
+ * * copyright in this work, please see the NOTICE file in the top level
+ * * directory of this distribution.
+ *
+ */
+
+package org.apache.usergrid.persistence.queue;
+
+import rx.Observable;
+
+import java.io.IOException;
+import java.util.AbstractQueue;
++import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.PriorityBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Default queue manager implementation, uses in memory linked queue
+ */
+public class LocalQueueManager implements QueueManager {
+ public ArrayBlockingQueue<QueueMessage> queue = new ArrayBlockingQueue<>(10000);
+
+ @Override
+ public Observable<QueueMessage> getMessages(int limit, int transactionTimeout, int waitTime, Class klass) {
+ List<QueueMessage> returnQueue = new ArrayList<>();
+ try {
+ QueueMessage message=null;
+ int count = 5;
+ do {
+ message = queue.poll(100, TimeUnit.MILLISECONDS);
+ if (message != null) {
+ returnQueue.add(message);
+ }
+ }while(message!=null && count-->0);
+ }catch (InterruptedException ie){
+ throw new RuntimeException(ie);
+ }
+ return Observable.from( returnQueue);
+ }
+
+ @Override
+ public long getQueueDepth() {
+ return queue.size();
+ }
+
+ @Override
+ public void commitMessage(QueueMessage queueMessage) {
+ }
+
+ @Override
+ public void commitMessages(List<QueueMessage> queueMessages) {
+ }
+
+ @Override
+ public void sendMessages(List bodies) throws IOException {
+ for(Object body : bodies){
+ String uuid = UUID.randomUUID().toString();
+ try {
+ queue.put(new QueueMessage(uuid, "handle_" + uuid, body, "put type here"));
+ }catch (InterruptedException ie){
+ throw new RuntimeException(ie);
+ }
+ }
+ }
+
++
+ @Override
- public void sendMessage(Object body) throws IOException {
++ public <T extends Serializable> void sendMessage( final T body ) throws IOException {
+ String uuid = UUID.randomUUID().toString();
+ try {
+ queue.put(new QueueMessage(uuid, "handle_" + uuid, body, "put type here"));
+ }catch (InterruptedException ie){
+ throw new RuntimeException(ie);
+ }
+ }
+
++
++
++ @Override
++ public <T extends Serializable> void sendMessageToTopic( final T body ) throws IOException {
++ sendMessage( body );
++ }
++
++
+ @Override
+ public void deleteQueue() {
+
+ }
+}
[04/23] usergrid git commit: Merge branch '2.1-release' of
https://git-wip-us.apache.org/repos/asf/usergrid into
remove-inmemory-event-service
Posted by sf...@apache.org.
Merge branch '2.1-release' of https://git-wip-us.apache.org/repos/asf/usergrid into remove-inmemory-event-service
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/f6409ce3
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/f6409ce3
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/f6409ce3
Branch: refs/heads/master
Commit: f6409ce353e9b9be2adad8c47e3a6800f94402ae
Parents: e17dab4 9cbe283
Author: Shawn Feldman <sf...@apache.org>
Authored: Tue Oct 13 10:14:11 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Tue Oct 13 10:14:11 2015 -0600
----------------------------------------------------------------------
.../asyncevents/AmazonAsyncEventService.java | 13 +++++++----
.../asyncevents/EventBuilderImpl.java | 3 ++-
.../core/rx/ExceptionBehaviorTest.java | 24 ++++++++++++++++++++
.../index/impl/EsIndexProducerImpl.java | 11 ++++++---
.../persistence/queue/QueueMessage.java | 10 ++++++++
.../queue/impl/SNSQueueManagerImpl.java | 2 ++
.../queue/impl/SQSQueueManagerImpl.java | 1 +
7 files changed, 56 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/f6409ce3/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/f6409ce3/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/EventBuilderImpl.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/f6409ce3/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsIndexProducerImpl.java
----------------------------------------------------------------------
[07/23] usergrid git commit: fix refresh
Posted by sf...@apache.org.
fix refresh
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/505b44a8
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/505b44a8
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/505b44a8
Branch: refs/heads/master
Commit: 505b44a8cdf494a53de1c171d9a8becff79d0063
Parents: 298f0d8
Author: Shawn Feldman <sf...@apache.org>
Authored: Tue Oct 13 16:00:21 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Tue Oct 13 16:00:21 2015 -0600
----------------------------------------------------------------------
.../org/apache/usergrid/corepersistence/CpEntityManager.java | 8 ++++++--
1 file changed, 6 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/505b44a8/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 4a13cd8..82bb6bb 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
@@ -2885,6 +2885,8 @@ public class CpEntityManager implements EntityManager {
map.put("some prop", "test");
boolean hasFinished = false;
Entity refreshEntity = create("refresh", map);
+ IndexRefreshCommand.IndexRefreshCommandInfo indexRefreshCommandInfo
+ = managerCache.getEntityIndex(applicationScope).refreshAsync().toBlocking().first();
try {
for (int i = 0; i < 10; i++) {
if (searchCollection(
@@ -2897,12 +2899,14 @@ public class CpEntityManager implements EntityManager {
break;
}
Thread.sleep(250);
- return managerCache.getEntityIndex(applicationScope).refreshAsync().toBlocking().first();
+ }
+ if(!hasFinished){
+ logger.warn("Did not find entity {} during refresh.",refreshEntity.getUuid());
}
}finally {
delete(refreshEntity);
}
- return new IndexRefreshCommand.IndexRefreshCommandInfo(hasFinished,System.currentTimeMillis() - start);
+ return indexRefreshCommandInfo;
} catch (Exception e) {
throw new RuntimeException("refresh failed",e);
}
[19/23] usergrid git commit: merge
Posted by sf...@apache.org.
merge
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/00175cc6
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/00175cc6
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/00175cc6
Branch: refs/heads/master
Commit: 00175cc61111c7bc67971c11b1049bb7ddbc7c2b
Parents: 5ecc57f
Author: Shawn Feldman <sf...@apache.org>
Authored: Mon Oct 19 13:25:15 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Mon Oct 19 13:25:15 2015 -0600
----------------------------------------------------------------------
.../usergrid/corepersistence/asyncevents/AsyncIndexProvider.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/00175cc6/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncIndexProvider.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncIndexProvider.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncIndexProvider.java
index fb63276..1ad9b86 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncIndexProvider.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncIndexProvider.java
@@ -100,7 +100,7 @@ public class AsyncIndexProvider implements Provider<AsyncEventService> {
switch (impl) {
case LOCAL:
AmazonAsyncEventService eventService = new AmazonAsyncEventService(scope -> new LocalQueueManager(), indexProcessorFig, indexProducer, metricsFactory,
- entityCollectionManagerFactory, indexLocationStrategyFactory, entityIndexFactory, eventBuilder, rxTaskScheduler);
+ entityCollectionManagerFactory, indexLocationStrategyFactory, entityIndexFactory, eventBuilder, rxTaskScheduler,queueFig);
eventService.MAX_TAKE = 1000;
return eventService;
case SQS:
[16/23] usergrid git commit: change queue impl
Posted by sf...@apache.org.
change queue impl
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/e5e9d021
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/e5e9d021
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/e5e9d021
Branch: refs/heads/master
Commit: e5e9d021ba1b27688fe6cf88482473980c0c3141
Parents: 6a3bae1
Author: Shawn Feldman <sf...@apache.org>
Authored: Fri Oct 16 15:50:48 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Fri Oct 16 15:50:48 2015 -0600
----------------------------------------------------------------------
.../org/apache/usergrid/persistence/queue/DefaultQueueManager.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/e5e9d021/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
index 3b6fe07..08f17a0 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/DefaultQueueManager.java
@@ -36,7 +36,7 @@ import java.util.concurrent.TimeUnit;
* Default queue manager implementation, uses in memory linked queue
*/
public class DefaultQueueManager implements QueueManager {
- public ArrayBlockingQueue<QueueMessage> queue = new ArrayBlockingQueue<>(1000);
+ public ArrayBlockingQueue<QueueMessage> queue = new ArrayBlockingQueue<>(10000);
@Override
public Observable<QueueMessage> getMessages(int limit, int transactionTimeout, int waitTime, Class klass) {
[05/23] usergrid git commit: remove observable
Posted by sf...@apache.org.
remove observable
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/32d35e7d
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/32d35e7d
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/32d35e7d
Branch: refs/heads/master
Commit: 32d35e7d5e605005b3f530e3a1e390d394c29c0a
Parents: f6409ce
Author: Shawn Feldman <sf...@apache.org>
Authored: Tue Oct 13 10:42:58 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Tue Oct 13 10:42:58 2015 -0600
----------------------------------------------------------------------
.../asyncevents/AmazonAsyncEventService.java | 109 ++++++++++---------
1 file changed, 55 insertions(+), 54 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/32d35e7d/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
index 6f41563..0fef974 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
@@ -259,65 +259,66 @@ public class AmazonAsyncEventService implements AsyncEventService {
logger.debug("callEventHandlers with {} message", messages.size());
}
- Observable<IndexEventResult> masterObservable = Observable.from(messages).map(message -> {
- AsyncEvent event = null;
- try {
- event = (AsyncEvent) message.getBody();
- } catch (ClassCastException cce) {
- logger.error("Failed to deserialize message body", cce);
- }
-
- if (event == null) {
- logger.error("AsyncEvent type or event is null!");
- return new IndexEventResult(Optional.fromNullable(message), Optional.<IndexOperationMessage>absent(), System.currentTimeMillis());
- }
-
- final AsyncEvent thisEvent = event;
- if (logger.isDebugEnabled()) {
- logger.debug("Processing {} event", event);
- }
-
- try {
- Observable<IndexOperationMessage> indexoperationObservable;
- //merge each operation to a master observable;
- if (event instanceof EdgeDeleteEvent) {
- indexoperationObservable = handleEdgeDelete(message);
- } else if (event instanceof EdgeIndexEvent) {
- indexoperationObservable = handleEdgeIndex(message);
- } else if (event instanceof EntityDeleteEvent) {
- indexoperationObservable = handleEntityDelete(message);
- } else if (event instanceof EntityIndexEvent) {
- indexoperationObservable = handleEntityIndexUpdate(message);
- } else if (event instanceof InitializeApplicationIndexEvent) {
- //does not return observable
- handleInitializeApplicationIndex(event, message);
- indexoperationObservable = Observable.just(new IndexOperationMessage());
- } else {
- throw new Exception("Unknown EventType");//TODO: print json instead
+ List<IndexEventResult> indexEventResults = messages.stream()
+ .map(message -> {
+ AsyncEvent event = null;
+ try {
+ event = (AsyncEvent) message.getBody();
+ } catch (ClassCastException cce) {
+ logger.error("Failed to deserialize message body", cce);
}
- //collect all of the
- IndexOperationMessage indexOperationMessage =
- indexoperationObservable
- .collect(() -> new IndexOperationMessage(), (collector, single) -> collector.ingest(single))
- .toBlocking().lastOrDefault(null);
+ if (event == null) {
+ logger.error("AsyncEvent type or event is null!");
+ return new IndexEventResult(Optional.fromNullable(message), Optional.<IndexOperationMessage>absent(), System.currentTimeMillis());
+ }
- if (indexOperationMessage == null || indexOperationMessage.isEmpty()) {
- logger.info("Received empty index sequence message:({}), body:({}) ",
- message.getMessageId(),message.getStringBody());
+ final AsyncEvent thisEvent = event;
+ if (logger.isDebugEnabled()) {
+ logger.debug("Processing {} event", event);
}
- //return type that can be indexed and ack'd later
- return new IndexEventResult(Optional.fromNullable(message), Optional.fromNullable(indexOperationMessage), thisEvent.getCreationTime());
- } catch (Exception e) {
- logger.error("Failed to index message: " + message.getMessageId(), message.getStringBody() ,e);
- return new IndexEventResult(Optional.absent(), Optional.<IndexOperationMessage>absent(), event.getCreationTime());
- }
- });
- //resolve the list and return it.
- final List<IndexEventResult> indexEventResults = masterObservable
- .collect(() -> new ArrayList<IndexEventResult>(), (list,indexEventResult) -> list.add(indexEventResult) )
- .toBlocking().lastOrDefault(null);
+ try {
+ //check for empty sets
+ boolean validateEmptySets = true;
+ Observable<IndexOperationMessage> indexoperationObservable;
+ //merge each operation to a master observable;
+ if (event instanceof EdgeDeleteEvent) {
+ indexoperationObservable = handleEdgeDelete(message);
+ } else if (event instanceof EdgeIndexEvent) {
+ indexoperationObservable = handleEdgeIndex(message);
+ } else if (event instanceof EntityDeleteEvent) {
+ indexoperationObservable = handleEntityDelete(message);
+ } else if (event instanceof EntityIndexEvent) {
+ indexoperationObservable = handleEntityIndexUpdate(message);
+ } else if (event instanceof InitializeApplicationIndexEvent) {
+ //does not return observable
+ handleInitializeApplicationIndex(event, message);
+ indexoperationObservable = Observable.just(new IndexOperationMessage());
+ validateEmptySets = false;
+ } else {
+ throw new Exception("Unknown EventType");//TODO: print json instead
+ }
+
+ //collect all of the
+ IndexOperationMessage indexOperationMessage =
+ indexoperationObservable
+ .collect(() -> new IndexOperationMessage(), (collector, single) -> collector.ingest(single))
+ .toBlocking().lastOrDefault(null);
+
+ if (validateEmptySets && (indexOperationMessage == null || indexOperationMessage.isEmpty())) {
+ logger.error("Received empty index sequence message:({}), body:({}) ",
+ message.getMessageId(), message.getStringBody());
+ }
+
+ //return type that can be indexed and ack'd later
+ return new IndexEventResult(Optional.fromNullable(message), Optional.fromNullable(indexOperationMessage), thisEvent.getCreationTime());
+ } catch (Exception e) {
+ logger.error("Failed to index message: " + message.getMessageId(), message.getStringBody(), e);
+ return new IndexEventResult(Optional.absent(), Optional.<IndexOperationMessage>absent(), event.getCreationTime());
+ }
+ })
+ .collect(Collectors.toList());
return indexEventResults;
[14/23] usergrid git commit: Merge branch 'master' of
https://git-wip-us.apache.org/repos/asf/usergrid into
remove-inmemory-event-service
Posted by sf...@apache.org.
Merge branch 'master' of https://git-wip-us.apache.org/repos/asf/usergrid into remove-inmemory-event-service
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/4992f5ca
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/4992f5ca
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/4992f5ca
Branch: refs/heads/master
Commit: 4992f5cab75b0ce0bb8c591ea00182a6dc186e64
Parents: 9673169 e7c805f
Author: Shawn Feldman <sf...@apache.org>
Authored: Fri Oct 16 15:08:32 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Fri Oct 16 15:08:32 2015 -0600
----------------------------------------------------------------------
website/content/community/index.html | 2 +-
website/content/img/mike_d.JPG | Bin 36443 -> 0 bytes
website/content/img/miked.jpg | Bin 0 -> 36443 bytes
3 files changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------