ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject ignite git commit: ignite-1.5
Date Thu, 24 Dec 2015 09:02:15 GMT
Repository: ignite
Updated Branches:
  refs/heads/ignite-1537 679cda5d2 -> 31f0ddf5e


ignite-1.5


Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/31f0ddf5
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/31f0ddf5
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/31f0ddf5

Branch: refs/heads/ignite-1537
Commit: 31f0ddf5e3c8aee63b66cd8dd571221a2bcb6b64
Parents: 679cda5
Author: sboikov <sboikov@gridgain.com>
Authored: Thu Dec 24 12:01:57 2015 +0300
Committer: sboikov <sboikov@gridgain.com>
Committed: Thu Dec 24 12:01:57 2015 +0300

----------------------------------------------------------------------
 .../stream/mqtt/IgniteMqttStreamerTest.java     | 49 ++++++++++++++------
 1 file changed, 34 insertions(+), 15 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/31f0ddf5/modules/mqtt/src/test/java/org/apache/ignite/stream/mqtt/IgniteMqttStreamerTest.java
----------------------------------------------------------------------
diff --git a/modules/mqtt/src/test/java/org/apache/ignite/stream/mqtt/IgniteMqttStreamerTest.java
b/modules/mqtt/src/test/java/org/apache/ignite/stream/mqtt/IgniteMqttStreamerTest.java
index 891866d..6c7f67a 100644
--- a/modules/mqtt/src/test/java/org/apache/ignite/stream/mqtt/IgniteMqttStreamerTest.java
+++ b/modules/mqtt/src/test/java/org/apache/ignite/stream/mqtt/IgniteMqttStreamerTest.java
@@ -37,12 +37,14 @@ import org.apache.ignite.IgniteCache;
 import org.apache.ignite.IgniteDataStreamer;
 import org.apache.ignite.cache.CachePeekMode;
 import org.apache.ignite.events.CacheEvent;
+import org.apache.ignite.internal.util.lang.GridAbsPredicate;
 import org.apache.ignite.internal.util.lang.GridMapEntry;
 import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.lang.IgniteBiPredicate;
 import org.apache.ignite.lang.IgniteInClosure;
 import org.apache.ignite.stream.StreamMultipleTupleExtractor;
 import org.apache.ignite.stream.StreamSingleTupleExtractor;
+import org.apache.ignite.testframework.GridTestUtils;
 import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
 
 import com.github.rholder.retry.StopStrategies;
@@ -87,7 +89,7 @@ public class IgniteMqttStreamerTest extends GridCommonAbstractTest {
     private MqttStreamer<Integer, String> streamer;
 
     /** The UUID of the currently active remote listener. */
-    private UUID remoteListener;
+    private UUID remoteLsnr;
 
     /** The Ignite data streamer. */
     private IgniteDataStreamer<Integer, String> dataStreamer;
@@ -105,7 +107,8 @@ public class IgniteMqttStreamerTest extends GridCommonAbstractTest {
     /**
      * @throws Exception If failed.
      */
-    @Before @SuppressWarnings("unchecked")
+    @Before
+    @SuppressWarnings("unchecked")
     public void beforeTest() throws Exception {
         grid().<Integer, String>getOrCreateCache(defaultCacheConfiguration());
 
@@ -121,13 +124,13 @@ public class IgniteMqttStreamerTest extends GridCommonAbstractTest {
         broker.setPersistenceAdapter(null);
         broker.setPersistenceFactory(null);
 
-        PolicyMap policyMap = new PolicyMap();
-        PolicyEntry policy = new PolicyEntry();
+        PolicyMap plcMap = new PolicyMap();
+        PolicyEntry plc = new PolicyEntry();
 
-        policy.setQueuePrefetch(1);
+        plc.setQueuePrefetch(1);
 
-        broker.setDestinationPolicy(policyMap);
-        broker.getDestinationPolicy().setDefaultEntry(policy);
+        broker.setDestinationPolicy(plcMap);
+        broker.getDestinationPolicy().setDefaultEntry(plc);
         broker.setSchedulerSupport(false);
 
         // add the MQTT transport connector to the broker
@@ -204,16 +207,32 @@ public class IgniteMqttStreamerTest extends GridCommonAbstractTest {
 
         // action time: repeat 5 times; make sure the connection state is kept correctly
every time
         for (int i = 0; i < 5; i++) {
+            log.info("Iteration: " + i);
+
+            GridTestUtils.waitForCondition(new GridAbsPredicate() {
+                @Override public boolean apply() {
+                    return streamer.isConnected();
+                }
+            }, 2000);
+
             assertTrue(streamer.isConnected());
 
             broker.stop();
 
+            GridTestUtils.waitForCondition(new GridAbsPredicate() {
+                @Override public boolean apply() {
+                    return !streamer.isConnected();
+                }
+            }, 2000);
+
             assertFalse(streamer.isConnected());
 
-            broker.start(true);
-            broker.waitUntilStarted();
+            if (i < 4) {
+                broker.start(true);
+                broker.waitUntilStarted();
 
-            Thread.sleep(500);
+                Thread.sleep(500);
+            }
         }
     }
 
@@ -355,7 +374,7 @@ public class IgniteMqttStreamerTest extends GridCommonAbstractTest {
     }
 
     /**
-     * @throws Exception
+     * @throws Exception If failed.
      */
     public void testSingleTopic_NoQoS_Reconnect() throws Exception {
         // configure streamer
@@ -557,7 +576,7 @@ public class IgniteMqttStreamerTest extends GridCommonAbstractTest {
         // Listen to cache PUT events and expect as many as messages as test data items
         final CountDownLatch latch = new CountDownLatch(expect);
 
-        IgniteBiPredicate<UUID, CacheEvent> callback = new IgniteBiPredicate<UUID,
CacheEvent>() {
+        IgniteBiPredicate<UUID, CacheEvent> cb = new IgniteBiPredicate<UUID, CacheEvent>()
{
             @Override public boolean apply(UUID uuid, CacheEvent evt) {
                 latch.countDown();
 
@@ -565,8 +584,8 @@ public class IgniteMqttStreamerTest extends GridCommonAbstractTest {
             }
         };
 
-        remoteListener = ignite.events(ignite.cluster().forCacheNodes(null))
-            .remoteListen(callback, null, EVT_CACHE_OBJECT_PUT);
+        remoteLsnr = ignite.events(ignite.cluster().forCacheNodes(null))
+            .remoteListen(cb, null, EVT_CACHE_OBJECT_PUT);
 
         return latch;
     }
@@ -586,7 +605,7 @@ public class IgniteMqttStreamerTest extends GridCommonAbstractTest {
         assertEquals(cnt, cache.size(CachePeekMode.ALL));
 
         // remove the event listener
-        grid().events(grid().cluster().forCacheNodes(null)).stopRemoteListen(remoteListener);
+        grid().events(grid().cluster().forCacheNodes(null)).stopRemoteListen(remoteLsnr);
     }
 
     /**


Mime
View raw message