Repository: usergrid Updated Branches: refs/heads/remove-inmemory-event-service 5eed63d43 -> fad8ecdb0 Updates the defaults to be more sensible in a multi-region environment Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/3ec0f588 Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/3ec0f588 Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/3ec0f588 Branch: refs/heads/remove-inmemory-event-service Commit: 3ec0f5886b82737d4a7ed64fae01afbdb6707763 Parents: 4013f17 Author: Todd Nine Authored: Mon Oct 19 17:44:39 2015 -0600 Committer: Todd Nine Committed: Mon Oct 19 17:44:39 2015 -0600 ---------------------------------------------------------------------- .../corepersistence/index/PublishRxTest.java | 95 ---------------- .../usergrid/corepersistence/index/RxTest.java | 108 +++++++++++++++++++ .../persistence/core/astyanax/CassandraFig.java | 6 +- 3 files changed, 111 insertions(+), 98 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/usergrid/blob/3ec0f588/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 deleted file mode 100644 index 973a42d..0000000 --- a/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/PublishRxTest.java +++ /dev/null @@ -1,95 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.usergrid.corepersistence.index; - - -import java.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 { - - @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 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 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/usergrid/blob/3ec0f588/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/RxTest.java ---------------------------------------------------------------------- diff --git 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 new file mode 100644 index 0000000..1d940d0 --- /dev/null +++ b/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/RxTest.java @@ -0,0 +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 ); + + 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 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 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); + } + + +} http://git-wip-us.apache.org/repos/asf/usergrid/blob/3ec0f588/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/astyanax/CassandraFig.java ---------------------------------------------------------------------- diff --git a/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/astyanax/CassandraFig.java b/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/astyanax/CassandraFig.java index 79c198f..e98e0fd 100644 --- a/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/astyanax/CassandraFig.java +++ b/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/astyanax/CassandraFig.java @@ -79,15 +79,15 @@ public interface CassandraFig extends GuicyFig { String getDiscoveryType(); - @Default("CL_LOCAL_ONE") + @Default("CL_LOCAL_QUORUM") @Key(READ_CL) String getReadCL(); - @Default("CL_LOCAL_QUORUM") + @Default("CL_QUORUM") @Key(READ_CONSISTENT_CL) String getConsistentReadCL(); - @Default("CL_QUORUM") + @Default("CL_LOCAL_QUORUM") @Key(WRITE_CL) String getWriteCL();