Merge branch 'two-dot-o-dev' of https://git-wip-us.apache.org/repos/asf/incubator-usergrid into USERGRID-578
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/c77519ea Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/c77519ea Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/c77519ea Branch: refs/heads/two-dot-o-dev Commit: c77519ea78016b30ac9cb4957e4fc74a0093b983 Parents: 587e439 a58ccfa Author: Shawn Feldman <sfeld...@apache.org> Authored: Tue Apr 21 14:33:10 2015 -0600 Committer: Shawn Feldman <sfeld...@apache.org> Committed: Tue Apr 21 14:33:10 2015 -0600 ---------------------------------------------------------------------- .../index/AsyncIndexProvider.java | 5 +- .../index/InMemoryAsyncReIndexService.java | 43 +++++-- .../corepersistence/index/IndexService.java | 12 +- .../corepersistence/index/IndexServiceImpl.java | 55 ++++++--- .../corepersistence/TestIndexModule.java | 14 ++- .../corepersistence/index/IndexServiceTest.java | 112 +++++++++++++++++++ .../corepersistence/index/PublishRxTest.java | 95 ++++++++++++++++ .../corepersistence/index/PublishRxtest.java | 61 ---------- .../usergrid/persistence/CollectionIT.java | 1 - .../core/metrics/ObservableTimer.java | 72 ++++++++++++ .../core/rx/RxTaskSchedulerImpl.java | 5 +- .../persistence/model/entity/Entity.java | 2 +- .../persistence/index/impl/IndexRequest.java | 1 - .../usergrid/cassandra/SpringResource.java | 13 ++- 14 files changed, 379 insertions(+), 112 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/c77519ea/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/PublishRxTest.java ---------------------------------------------------------------------- diff --cc stack/core/src/test/java/org/apache/usergrid/corepersistence/index/PublishRxTest.java index 0000000,973a42d..4e65110 mode 000000,100644..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 @@@ -1,0 -1,95 +1,95 @@@ + /* + * 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 PublishRxTest { ++public class PublishRxtest { + + @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 ); + + 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 ); + } + } http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/c77519ea/stack/core/src/test/java/org/apache/usergrid/persistence/CollectionIT.java ----------------------------------------------------------------------
