ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From a.@apache.org
Subject [38/50] [abbrv] incubator-ignite git commit: # ignite-901 client reconnect support
Date Fri, 17 Jul 2015 12:51:47 GMT
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/57ac2b3b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectApiExceptionTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectApiExceptionTest.java b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectApiExceptionTest.java
new file mode 100644
index 0000000..37773cd
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectApiExceptionTest.java
@@ -0,0 +1,846 @@
+/*
+ * 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.ignite.internal;
+
+import org.apache.ignite.*;
+import org.apache.ignite.cache.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.events.*;
+import org.apache.ignite.internal.util.typedef.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.ignite.lang.*;
+import org.apache.ignite.resources.*;
+import org.apache.ignite.testframework.*;
+
+import javax.cache.*;
+import javax.cache.processor.*;
+import java.util.*;
+import java.util.concurrent.*;
+
+import static java.util.concurrent.TimeUnit.*;
+import static org.apache.ignite.events.EventType.*;
+
+/**
+ *
+ */
+public class IgniteClientReconnectApiExceptionTest extends IgniteClientReconnectAbstractTest {
+    /** {@inheritDoc} */
+    @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+        IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+        cfg.setCacheConfiguration(new CacheConfiguration());
+
+        return cfg;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected int serverCount() {
+        return 1;
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testErrorOnDisconnect() throws Exception {
+        // Check cache operations.
+        cacheOperationsTest();
+
+        // Check cache operations.
+        beforeTestsStarted();
+        dataStructureOperationsTest();
+
+        // Check ignite operations.
+        beforeTestsStarted();
+        igniteOperationsTest();
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    @SuppressWarnings("unchecked")
+    public void dataStructureOperationsTest() throws Exception {
+        clientMode = true;
+
+        final Ignite client = startGrid(serverCount());
+
+        doTestIgniteOperationOnDisconnect(client, Arrays.asList(
+            // Check atomic long.
+            new T2<Callable, C1<Object, Boolean>>(
+                new Callable() {
+                    @Override public Object call() throws Exception {
+                        boolean failed = false;
+
+                        try {
+                            client.atomicLong("testAtomic", 41, true);
+                        }
+                        catch (IgniteClientDisconnectedException e) {
+                            failed = true;
+
+                            checkAndWait(e);
+                        }
+
+                        assertTrue(failed);
+
+                        return client.atomicLong("testAtomic", 41, true);
+                    }
+                },
+                new C1<Object, Boolean>() {
+                    @Override public Boolean apply(Object o) {
+                        assertNotNull(o);
+
+                        IgniteAtomicLong atomicLong = (IgniteAtomicLong)o;
+
+                        assertEquals(42, atomicLong.incrementAndGet());
+
+                        return true;
+                    }
+                }
+            ),
+            // Check set.
+            new T2<Callable, C1<Object, Boolean>>(
+                new Callable() {
+                    @Override public Object call() throws Exception {
+                        boolean failed = false;
+
+                        try {
+                            client.set("testSet", new CollectionConfiguration());
+                        }
+                        catch (IgniteClientDisconnectedException e) {
+                            failed = true;
+
+                            checkAndWait(e);
+                        }
+
+                        assertTrue(failed);
+
+                        return client.set("testSet", new CollectionConfiguration());
+                    }
+                },
+                new C1<Object, Boolean>() {
+                    @Override public Boolean apply(Object o) {
+                        assertNotNull(o);
+
+                        IgniteSet set = (IgniteSet)o;
+
+                        String val = "testVal";
+
+                        set.add(val);
+
+                        assertEquals(1, set.size());
+                        assertTrue(set.contains(val));
+
+                        return true;
+                    }
+                }
+            ),
+            // Check ignite queue.
+            new T2<Callable, C1<Object, Boolean>>(
+                new Callable() {
+                    @Override public Object call() throws Exception {
+                        boolean failed = false;
+
+                        try {
+                            client.queue("TestQueue", 10, new CollectionConfiguration());
+                        }
+                        catch (IgniteClientDisconnectedException e) {
+                            failed = true;
+
+                            checkAndWait(e);
+                        }
+
+                        assertTrue(failed);
+
+                        return client.queue("TestQueue", 10, new CollectionConfiguration());
+                    }
+                },
+                new C1<Object, Boolean>() {
+                    @Override public Boolean apply(Object o) {
+                        assertNotNull(o);
+
+                        IgniteQueue queue = (IgniteQueue)o;
+
+                        String val = "Test";
+
+                        queue.add(val);
+
+                        assertEquals(val, queue.poll());
+
+                        return true;
+                    }
+                }
+            )
+        ));
+
+        clientMode = false;
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    @SuppressWarnings("unchecked")
+    public void cacheOperationsTest() throws Exception {
+        clientMode = true;
+
+        final Ignite client = startGrid(serverCount());
+
+        final IgniteCache<Object, Object> dfltCache = client.cache(null);
+
+        assertNotNull(dfltCache);
+
+        doTestIgniteOperationOnDisconnect(client, Arrays.asList(
+            // Check put and get operation.
+            new T2<Callable, C1<Object, Boolean>>(
+                new Callable() {
+                    @Override public Object call() throws Exception {
+                        boolean failed = false;
+
+                        try {
+                            dfltCache.getAndPut(9999, 9999);
+                        }
+                        catch (CacheException e) {
+                            failed = true;
+
+                            checkAndWait(e);
+                        }
+
+                        assertTrue(failed);
+
+                        return dfltCache.getAndPut(9999, 9999);
+                    }
+                },
+                new C1<Object, Boolean>() {
+                    @Override public Boolean apply(Object o) {
+                        assertNull(o);
+
+                        assertEquals(9999, dfltCache.get(9999));
+
+                        return true;
+                    }
+                }
+            ),
+            // Check put operation.
+            new T2<Callable, C1<Object, Boolean>>(
+                new Callable() {
+                    @Override public Object call() throws Exception {
+                        boolean failed = false;
+
+                        try {
+                            dfltCache.put(10000, 10000);
+                        }
+                        catch (CacheException e) {
+                            failed = true;
+
+                            checkAndWait(e);
+                        }
+
+                        assertTrue(failed);
+
+                        dfltCache.put(10000, 10000);
+
+                        return true;
+                    }
+                },
+                new C1<Object, Boolean>() {
+                    @Override public Boolean apply(Object o) {
+                        assertTrue((Boolean)o);
+
+                        assertEquals(10000, dfltCache.get(10000));
+
+                        return true;
+                    }
+                }
+            ),
+            // Check get operation.
+            new T2<Callable, C1<Object, Boolean>>(
+                new Callable() {
+                    @Override public Object call() throws Exception {
+                        boolean failed = false;
+
+                        try {
+                            dfltCache.get(10001);
+                        }
+                        catch (CacheException e) {
+                            failed = true;
+
+                            checkAndWait(e);
+                        }
+
+                        assertTrue(failed);
+
+                        return dfltCache.get(10001);
+                    }
+                },
+                new C1<Object, Boolean>() {
+                    @Override public Boolean apply(Object o) {
+                        assertNull(o);
+
+                        return true;
+                    }
+                }
+            ),
+            // Check invoke operation.
+            new T2<Callable, C1<Object, Boolean>>(
+                new Callable() {
+                    @Override public Object call() throws Exception {
+                        boolean failed = false;
+
+                        try {
+                            dfltCache.invoke(10000, new CacheEntryProcessor<Object, Object, Object>() {
+                                @Override public Object process(MutableEntry<Object, Object> entry,
+                                    Object... arguments) throws EntryProcessorException {
+                                    assertTrue(entry.exists());
+
+                                    return (int)entry.getValue() * 2;
+                                }
+                            });
+                        }
+                        catch (CacheException e) {
+                            failed = true;
+
+                            checkAndWait(e);
+                        }
+
+                        assertTrue(failed);
+
+                        return dfltCache.invoke(10000, new CacheEntryProcessor<Object, Object, Object>() {
+                            @Override public Object process(MutableEntry<Object, Object> entry,
+                                Object... arguments) throws EntryProcessorException {
+                                assertTrue(entry.exists());
+
+                                return (int)entry.getValue() * 2;
+                            }
+                        });
+                    }
+                },
+                new C1<Object, Boolean>() {
+                    @Override public Boolean apply(Object o) {
+                        assertNotNull(o);
+
+                        assertEquals(20000, (int)o);
+
+                        return true;
+                    }
+                }
+            ),
+            // Check put async operation.
+            new T2<Callable, C1<Object, Boolean>>(
+                new Callable() {
+                    @Override public Object call() throws Exception {
+                        IgniteCache<Object, Object> async = dfltCache.withAsync();
+
+                        boolean failed = false;
+
+                        try {
+                            async.put(10002, 10002);
+
+                            async.future().get();
+                        }
+                        catch (CacheException e) {
+                            failed = true;
+
+                            checkAndWait(e);
+                        }
+
+                        assertTrue(failed);
+
+                        async.put(10002, 10002);
+
+                        return async.future().get();
+                    }
+                },
+                new C1<Object, Boolean>() {
+                    @Override public Boolean apply(Object o) {
+                        assertNull(o);
+
+                        assertEquals(10002, dfltCache.get(10002));
+
+                        return true;
+                    }
+                }
+            ),
+            // Check transaction.
+            new T2<Callable, C1<Object, Boolean>>(
+                new Callable() {
+                    @Override public Object call() throws Exception {
+                        boolean failed = false;
+
+                        try {
+                            client.transactions();
+                        }
+                        catch (IgniteClientDisconnectedException e) {
+                            failed = true;
+
+                            checkAndWait(e);
+                        }
+
+                        assertTrue(failed);
+
+                        return client.transactions();
+                    }
+                },
+                new C1<Object, Boolean>() {
+                    @Override public Boolean apply(Object o) {
+                        IgniteTransactions txs = (IgniteTransactions)o;
+
+                        assertNotNull(txs);
+
+                        return true;
+                    }
+                }
+            ),
+            // Check get cache.
+            new T2<Callable, C1<Object, Boolean>>(
+                new Callable() {
+                    @Override public Object call() throws Exception {
+                        boolean failed = false;
+
+                        try {
+                            client.cache(null);
+                        }
+                        catch (IgniteClientDisconnectedException e) {
+                            failed = true;
+
+                            checkAndWait(e);
+                        }
+
+                        assertTrue(failed);
+
+                        return client.cache(null);
+                    }
+                },
+                new C1<Object, Boolean>() {
+                    @Override public Boolean apply(Object o) {
+                        IgniteCache<Object, Object> cache0 = (IgniteCache<Object, Object>)o;
+
+                        assertNotNull(cache0);
+
+                        cache0.put(1, 1);
+
+                        assertEquals(1, cache0.get(1));
+
+                        return true;
+                    }
+                }
+            ),
+            // Check streamer.
+            new T2<Callable, C1<Object, Boolean>>(
+                new Callable() {
+                    @Override public Object call() throws Exception {
+                        boolean failed = false;
+
+                        try {
+                            client.dataStreamer(null);
+                        }
+                        catch (IgniteClientDisconnectedException e) {
+                            failed = true;
+
+                            checkAndWait(e);
+                        }
+
+                        assertTrue(failed);
+
+                        return client.dataStreamer(null);
+                    }
+                },
+                new C1<Object, Boolean>() {
+                    @Override public Boolean apply(Object o) {
+                        IgniteDataStreamer<Object, Object> streamer = (IgniteDataStreamer<Object, Object>)o;
+
+                        streamer.addData(2, 2);
+
+                        streamer.close();
+
+                        assertEquals(2, client.cache(null).get(2));
+
+                        return true;
+                    }
+                }
+            ),
+            // Check create cache.
+            new T2<Callable, C1<Object, Boolean>>(
+                new Callable() {
+                    @Override public Object call() throws Exception {
+                        boolean failed = false;
+
+                        try {
+                            client.createCache("test_cache");
+                        }
+                        catch (IgniteClientDisconnectedException e) {
+                            failed = true;
+
+                            checkAndWait(e);
+                        }
+
+                        assertTrue(failed);
+
+                        return client.createCache("test_cache");
+                    }
+                },
+                new C1<Object, Boolean>() {
+                    @Override public Boolean apply(Object o) {
+                        IgniteCache<Object, Object> cache = (IgniteCache<Object, Object>)o;
+
+                        assertNotNull(cache);
+
+                        cache.put(1, 1);
+
+                        assertEquals(1, cache.get(1));
+
+                        return true;
+                    }
+                }
+            )
+
+        ));
+
+        clientMode = false;
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    @SuppressWarnings("unchecked")
+    public void igniteOperationsTest() throws Exception {
+        clientMode = true;
+
+        final Ignite client = startGrid(serverCount());
+
+        final IgniteCache<Object, Object> dfltCache = client.cache(null);
+
+        final CountDownLatch recvLatch = new CountDownLatch(1);
+
+        assertNotNull(dfltCache);
+
+        doTestIgniteOperationOnDisconnect(client, Arrays.asList(
+            // Check compute.
+            new T2<Callable, C1<Object, Boolean>>(
+                new Callable() {
+                    @Override public Object call() throws Exception {
+                        boolean failed = false;
+
+                        try {
+                            client.compute();
+                        }
+                        catch (IgniteClientDisconnectedException e) {
+                            failed = true;
+
+                            checkAndWait(e);
+                        }
+
+                        assertTrue(failed);
+
+                        return client.compute();
+                    }
+                },
+                new C1<Object, Boolean>() {
+                    @Override public Boolean apply(Object o) {
+                        IgniteCompute comp = (IgniteCompute)o;
+
+                        Collection<UUID> uuids = comp.broadcast(new IgniteCallable<UUID>() {
+                            @IgniteInstanceResource
+                            private Ignite ignite;
+
+                            @Override public UUID call() throws Exception {
+                                return ignite.cluster().localNode().id();
+                            }
+                        });
+
+                        assertFalse(uuids.isEmpty());
+
+                        for (UUID uuid : uuids)
+                            assertNotNull(uuid);
+
+                        return true;
+                    }
+                }
+            ),
+
+            // Check ping node.
+            new T2<Callable, C1<Object, Boolean>>(
+                new Callable() {
+                    @Override public Object call() throws Exception {
+                        boolean failed = false;
+
+                        try {
+                            client.cluster().pingNode(new UUID(0, 0));
+                        }
+                        catch (IgniteClientDisconnectedException e) {
+                            failed = true;
+
+                            checkAndWait(e);
+                        }
+
+                        assertTrue(failed);
+
+                        return client.cluster().pingNode(new UUID(0, 0));
+                    }
+                },
+                new C1<Object, Boolean>() {
+                    @Override public Boolean apply(Object o) {
+                        Boolean pingNode = (Boolean)o;
+
+                        assertFalse(pingNode);
+
+                        return true;
+                    }
+                }
+            ),
+            // Check register remote listener.
+            new T2<Callable, C1<Object, Boolean>>(
+                new Callable() {
+                    @Override public Object call() throws Exception {
+                        boolean failed = false;
+
+                        try {
+                            client.events().remoteListen(null, new IgnitePredicate<Event>() {
+                                @Override public boolean apply(Event event) {
+                                    return true;
+                                }
+                            });
+                        }
+                        catch (IgniteClientDisconnectedException e) {
+                            failed = true;
+
+                            checkAndWait(e);
+                        }
+
+                        assertTrue(failed);
+
+                        return client.events().remoteListen(null, new IgnitePredicate<Event>() {
+                            @Override public boolean apply(Event event) {
+                                return true;
+                            }
+                        });
+                    }
+                },
+                new C1<Object, Boolean>() {
+                    @Override public Boolean apply(Object o) {
+                        UUID remoteId = (UUID)o;
+
+                        assertNotNull(remoteId);
+
+                        client.events().stopRemoteListen(remoteId);
+
+                        return true;
+                    }
+                }
+            ),
+            // Check message operation.
+            new T2<Callable, C1<Object, Boolean>>(
+                new Callable() {
+                    @Override public Object call() throws Exception {
+                        boolean failed = false;
+
+                        try {
+                            client.message().remoteListen(null, new IgniteBiPredicate<UUID, Object>() {
+                                @Override public boolean apply(UUID uuid, Object o) {
+                                    if (o.equals("Test message."))
+                                        recvLatch.countDown();
+
+                                    return true;
+                                }
+                            });
+                        }
+                        catch (IgniteClientDisconnectedException e) {
+                            failed = true;
+
+                            checkAndWait(e);
+                        }
+
+                        assertTrue(failed);
+
+                        return client.message().remoteListen(null, new IgniteBiPredicate<UUID, Object>() {
+                            @Override public boolean apply(UUID uuid, Object o) {
+                                if (o.equals("Test message."))
+                                    recvLatch.countDown();
+
+                                return true;
+                            }
+                        });
+                    }
+                },
+                new C1<Object, Boolean>() {
+                    @Override public Boolean apply(Object o) {
+                        assertNotNull(o);
+
+                        IgniteMessaging msg = client.message();
+
+                        msg.send(null, "Test message.");
+
+                        try {
+                            assertTrue(recvLatch.await(2, SECONDS));
+                        }
+                        catch (InterruptedException e) {
+                            fail("Message wasn't received.");
+                        }
+
+                        return true;
+                    }
+                }
+            ),
+            // Check executor.
+            new T2<Callable, C1<Object, Boolean>>(
+                new Callable() {
+                    @Override public Object call() throws Exception {
+                        boolean failed = false;
+
+                        try {
+                            client.executorService().submit(new Callable<Integer>() {
+                                @Override public Integer call() throws Exception {
+                                    return 42;
+                                }
+                            });
+                        }
+                        catch (IgniteClientDisconnectedException e) {
+                            failed = true;
+
+                            checkAndWait(e);
+                        }
+
+                        assertTrue(failed);
+
+                        return client.executorService().submit(new Callable<Integer>() {
+                            @Override public Integer call() throws Exception {
+                                return 42;
+                            }
+                        });
+                    }
+                },
+                new C1<Object, Boolean>() {
+                    @Override public Boolean apply(Object o) {
+                        assertNotNull(o);
+
+                        Future<Integer> fut = (Future<Integer>)o;
+
+                        try {
+                            assertEquals(42, (int)fut.get());
+                        }
+                        catch (Exception e) {
+                            fail("Failed submit task.");
+                        }
+
+                        return true;
+                    }
+                }
+            )
+        ));
+
+        clientMode = false;
+    }
+
+    /**
+     * @param client Client.
+     * @param ops Operations closures.
+     * @throws Exception If failed.
+     */
+    @SuppressWarnings("unchecked")
+    private void doTestIgniteOperationOnDisconnect(Ignite client, final List<T2<Callable, C1<Object, Boolean>>> ops)
+        throws Exception {
+        assertNotNull(client.cache(null));
+
+        final TestTcpDiscoverySpi clientSpi = spi(client);
+
+        Ignite srv = clientRouter(client);
+
+        TestTcpDiscoverySpi srvSpi = spi(srv);
+
+        final CountDownLatch disconnectLatch = new CountDownLatch(1);
+
+        final CountDownLatch reconnectLatch = new CountDownLatch(1);
+
+        log.info("Block reconnect.");
+
+        clientSpi.writeLatch = new CountDownLatch(1);
+
+        final List<IgniteInternalFuture> futs = new ArrayList<>();
+
+        client.events().localListen(new IgnitePredicate<Event>() {
+            @Override public boolean apply(Event evt) {
+                if (evt.type() == EVT_CLIENT_NODE_DISCONNECTED) {
+                    info("Disconnected: " + evt);
+
+                    assertEquals(1, reconnectLatch.getCount());
+
+                    for (T2<Callable, C1<Object, Boolean>> op : ops)
+                        futs.add(GridTestUtils.runAsync(op.get1()));
+
+                    disconnectLatch.countDown();
+                }
+                else if (evt.type() == EVT_CLIENT_NODE_RECONNECTED) {
+                    info("Reconnected: " + evt);
+
+                    reconnectLatch.countDown();
+                }
+
+                return true;
+            }
+        }, EVT_CLIENT_NODE_DISCONNECTED, EVT_CLIENT_NODE_RECONNECTED);
+
+        try {
+            log.info("Fail client.");
+
+            srvSpi.failNode(client.cluster().localNode().id(), null);
+
+            waitReconnectEvent(disconnectLatch);
+
+            assertEquals(ops.size(), futs.size());
+
+            for (IgniteInternalFuture<?> fut : futs)
+                assertNotDone(fut);
+
+            U.sleep(2000);
+
+            for (IgniteInternalFuture<?> fut : futs)
+                assertNotDone(fut);
+
+            log.info("Allow reconnect.");
+
+            clientSpi.writeLatch.countDown();
+
+            waitReconnectEvent(reconnectLatch);
+
+            // Check operation after reconnect working.
+            for (int i = 0; i < futs.size(); i++) {
+                final int i0 = i;
+
+                try {
+                    final Object futRes = futs.get(i0).get(2, SECONDS);
+
+                    assertTrue(GridTestUtils.runAsync(new Callable<Boolean>() {
+                        @Override public Boolean call() throws Exception {
+                            return ops.get(i0).get2().apply(futRes);
+                        }
+                    }).get(2, SECONDS));
+                }
+                catch (IgniteFutureTimeoutCheckedException e) {
+                    e.printStackTrace();
+
+                    fail("Operation timeout. Iteration: " + i + ".");
+                }
+            }
+        }
+        finally {
+            clientSpi.writeLatch.countDown();
+
+            for (IgniteInternalFuture fut : futs)
+                fut.cancel();
+
+            stopAllGrids();
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/57ac2b3b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectAtomicsTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectAtomicsTest.java b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectAtomicsTest.java
new file mode 100644
index 0000000..bb568ab
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectAtomicsTest.java
@@ -0,0 +1,672 @@
+/*
+ * 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.ignite.internal;
+
+import org.apache.ignite.*;
+import org.apache.ignite.internal.processors.cache.distributed.near.*;
+import org.apache.ignite.testframework.*;
+
+import java.util.concurrent.*;
+
+/**
+ *
+ */
+public class IgniteClientReconnectAtomicsTest extends IgniteClientReconnectAbstractTest {
+    /** {@inheritDoc} */
+    @Override protected int serverCount() {
+        return 1;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected int clientCount() {
+        return 1;
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testAtomicSeqReconnect() throws Exception {
+        Ignite client = grid(serverCount());
+
+        assertTrue(client.cluster().localNode().isClient());
+
+        Ignite srv = clientRouter(client);
+
+        IgniteAtomicSequence clientAtomicSeq = client.atomicSequence("atomicSeq", 0, true);
+
+        assertEquals(1L, clientAtomicSeq.incrementAndGet());
+
+        final IgniteAtomicSequence srvAtomicSeq = srv.atomicSequence("atomicSeq", 0, false);
+
+        assertEquals(1001L, srvAtomicSeq.incrementAndGet());
+
+        reconnectClientNode(client, srv, new Runnable() {
+            @Override public void run() {
+                assertEquals(1002L, srvAtomicSeq.incrementAndGet());
+            }
+        });
+
+        assertEquals(2L, clientAtomicSeq.incrementAndGet());
+
+        assertEquals(1003L, srvAtomicSeq.incrementAndGet());
+
+        assertEquals(3L, clientAtomicSeq.incrementAndGet());
+
+        clientAtomicSeq.close();
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testAtomicSeqReconnectRemoved() throws Exception {
+        Ignite client = grid(serverCount());
+
+        assertTrue(client.cluster().localNode().isClient());
+
+        Ignite srv = clientRouter(client);
+
+        final IgniteAtomicSequence clientAtomicSeq = client.atomicSequence("atomicSeqRmv", 0, true);
+
+        clientAtomicSeq.batchSize(1);
+
+        assertEquals(1L, clientAtomicSeq.incrementAndGet());
+
+        final IgniteAtomicSequence srvAtomicSeq = srv.atomicSequence("atomicSeqRmv", 0, false);
+
+        srvAtomicSeq.batchSize(1);
+
+        assertEquals(1001L, srvAtomicSeq.incrementAndGet());
+
+        reconnectClientNode(client, srv, new Runnable() {
+            @Override public void run() {
+                srvAtomicSeq.close();
+
+                assert srvAtomicSeq.removed();
+            }
+        });
+
+        GridTestUtils.assertThrows(log, new Callable<Object>() {
+            @Override public Object call() throws Exception {
+                for (int i = 0; i < 2000; i++)
+                    clientAtomicSeq.incrementAndGet();
+
+                return null;
+            }
+        }, IllegalStateException.class, null);
+
+        IgniteAtomicSequence newClientAtomicSeq = client.atomicSequence("atomicSeqRmv", 0, true);
+
+        assertEquals(0, newClientAtomicSeq.get());
+
+        assertEquals(1, newClientAtomicSeq.incrementAndGet());
+
+        newClientAtomicSeq.close();
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testAtomicSeqReconnectInProgress() throws Exception {
+        Ignite client = grid(serverCount());
+
+        assertTrue(client.cluster().localNode().isClient());
+
+        Ignite srv = clientRouter(client);
+
+        BlockTpcCommunicationSpi commSpi = commSpi(srv);
+
+        final IgniteAtomicSequence clientAtomicSeq = client.atomicSequence("atomicSeqInProg", 0, true);
+
+        clientAtomicSeq.batchSize(1);
+
+        final IgniteAtomicSequence srvAtomicSeq = srv.atomicSequence("atomicSeqInProg", 0, false);
+
+        srvAtomicSeq.batchSize(1);
+
+        commSpi.blockMessage(GridNearLockResponse.class);
+
+        final IgniteInternalFuture<Object> fut = GridTestUtils.runAsync(new Callable<Object>() {
+            @Override public Object call() throws Exception {
+                for (int i = 0; i < 3000; i++) {
+                    try {
+                        clientAtomicSeq.incrementAndGet();
+                    }
+                    catch (IgniteClientDisconnectedException e) {
+                        checkAndWait(e);
+
+                        return true;
+                    }
+                }
+
+                return false;
+            }
+        });
+
+        // Check that client waiting operation.
+        GridTestUtils.assertThrows(log, new Callable<Object>() {
+            @Override public Object call() throws Exception {
+                return fut.get(200);
+            }
+        }, IgniteFutureTimeoutCheckedException.class, null);
+
+        assertNotDone(fut);
+
+        commSpi.unblockMessage();
+
+        reconnectClientNode(client, srv, null);
+
+        assertTrue((Boolean)fut.get(2, TimeUnit.SECONDS));
+
+        // Check that after reconnect working.
+        assert clientAtomicSeq.incrementAndGet() >= 0;
+        assert srvAtomicSeq.incrementAndGet() >= 0;
+
+        clientAtomicSeq.close();
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testAtomicReferenceReconnect() throws Exception {
+        Ignite client = grid(serverCount());
+
+        assertTrue(client.cluster().localNode().isClient());
+
+        Ignite srv = clientRouter(client);
+
+        IgniteAtomicReference<String> clientAtomicRef = client.atomicReference("atomicRef", "1st value", true);
+
+        assertEquals("1st value", clientAtomicRef.get());
+        assertTrue(clientAtomicRef.compareAndSet("1st value", "2st value"));
+        assertEquals("2st value", clientAtomicRef.get());
+
+        final IgniteAtomicReference<String> srvAtomicRef = srv.atomicReference("atomicRef", "1st value", false);
+
+        assertEquals("2st value", srvAtomicRef.get());
+        assertTrue(srvAtomicRef.compareAndSet("2st value", "3st value"));
+        assertEquals("3st value", srvAtomicRef.get());
+
+        reconnectClientNode(client, srv, new Runnable() {
+            @Override public void run() {
+                assertEquals("3st value", srvAtomicRef.get());
+                assertTrue(srvAtomicRef.compareAndSet("3st value", "4st value"));
+                assertEquals("4st value", srvAtomicRef.get());
+            }
+        });
+
+        assertEquals("4st value", clientAtomicRef.get());
+        assertTrue(clientAtomicRef.compareAndSet("4st value", "5st value"));
+        assertEquals("5st value", clientAtomicRef.get());
+
+        assertEquals("5st value", srvAtomicRef.get());
+        assertTrue(srvAtomicRef.compareAndSet("5st value", "6st value"));
+        assertEquals("6st value", srvAtomicRef.get());
+
+        srvAtomicRef.close();
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testAtomicReferenceReconnectRemoved() throws Exception {
+        Ignite client = grid(serverCount());
+
+        assertTrue(client.cluster().localNode().isClient());
+
+        Ignite srv = clientRouter(client);
+
+        final IgniteAtomicReference<String> clientAtomicRef =
+            client.atomicReference("atomicRefRemoved", "1st value", true);
+
+        assertEquals("1st value", clientAtomicRef.get());
+        assertTrue(clientAtomicRef.compareAndSet("1st value", "2st value"));
+        assertEquals("2st value", clientAtomicRef.get());
+
+        final IgniteAtomicReference<String> srvAtomicRef = srv.atomicReference("atomicRefRemoved", "1st value", false);
+
+        assertEquals("2st value", srvAtomicRef.get());
+        assertTrue(srvAtomicRef.compareAndSet("2st value", "3st value"));
+        assertEquals("3st value", srvAtomicRef.get());
+
+        reconnectClientNode(client, srv, new Runnable() {
+            @Override public void run() {
+                srvAtomicRef.close();
+            }
+        });
+
+        GridTestUtils.assertThrows(log, new Callable<Object>() {
+            @Override public Object call() throws Exception {
+                clientAtomicRef.compareAndSet("3st value", "4st value");
+
+                return null;
+            }
+        }, IllegalStateException.class, null);
+
+        IgniteAtomicReference<String> newClientAtomicRef =
+            client.atomicReference("atomicRefRemoved", "1st value", true);
+
+        IgniteAtomicReference<String> newSrvAtomicRef = srv.atomicReference("atomicRefRemoved", "1st value", false);
+
+        assertEquals("1st value", newClientAtomicRef.get());
+        assertTrue(newClientAtomicRef.compareAndSet("1st value", "2st value"));
+        assertEquals("2st value", newClientAtomicRef.get());
+
+        assertEquals("2st value", newSrvAtomicRef.get());
+        assertTrue(newSrvAtomicRef.compareAndSet("2st value", "3st value"));
+        assertEquals("3st value", newSrvAtomicRef.get());
+
+        newClientAtomicRef.close();
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testAtomicReferenceReconnectInProgress() throws Exception {
+        Ignite client = grid(serverCount());
+
+        assertTrue(client.cluster().localNode().isClient());
+
+        Ignite srv = clientRouter(client);
+
+        final IgniteAtomicReference<String> clientAtomicRef =
+            client.atomicReference("atomicRefInProg", "1st value", true);
+
+        assertEquals("1st value", clientAtomicRef.get());
+        assertTrue(clientAtomicRef.compareAndSet("1st value", "2st value"));
+        assertEquals("2st value", clientAtomicRef.get());
+
+        IgniteAtomicReference<String> srvAtomicRef = srv.atomicReference("atomicRefInProg", "1st value", false);
+
+        assertEquals("2st value", srvAtomicRef.get());
+        assertTrue(srvAtomicRef.compareAndSet("2st value", "3st value"));
+        assertEquals("3st value", srvAtomicRef.get());
+
+        BlockTpcCommunicationSpi servCommSpi = commSpi(srv);
+
+        servCommSpi.blockMessage(GridNearLockResponse.class);
+
+        final IgniteInternalFuture<Object> fut = GridTestUtils.runAsync(new Callable<Object>() {
+            @Override public Object call() throws Exception {
+                try {
+                    clientAtomicRef.compareAndSet("3st value", "4st value");
+                }
+                catch (IgniteClientDisconnectedException e) {
+                    checkAndWait(e);
+
+                    return true;
+                }
+
+                return false;
+            }
+        });
+
+        // Check that client waiting operation.
+        GridTestUtils.assertThrows(log, new Callable<Object>() {
+            @Override public Object call() throws Exception {
+                return fut.get(200);
+            }
+        }, IgniteFutureTimeoutCheckedException.class, null);
+
+        assertNotDone(fut);
+
+        servCommSpi.unblockMessage();
+
+        reconnectClientNode(client, srv, null);
+
+        assertTrue((Boolean)fut.get(2, TimeUnit.SECONDS));
+
+        // Check that after reconnect working.
+        assertEquals("3st value", clientAtomicRef.get());
+        assertTrue(clientAtomicRef.compareAndSet("3st value", "4st value"));
+        assertEquals("4st value", clientAtomicRef.get());
+
+        assertEquals("4st value", srvAtomicRef.get());
+        assertTrue(srvAtomicRef.compareAndSet("4st value", "5st value"));
+        assertEquals("5st value", srvAtomicRef.get());
+
+        srvAtomicRef.close();
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testAtomicStampedReconnect() throws Exception {
+        Ignite client = grid(serverCount());
+
+        assertTrue(client.cluster().localNode().isClient());
+
+        Ignite srv = clientRouter(client);
+
+        IgniteAtomicStamped clientAtomicStamped = client.atomicStamped("atomicStamped", 0, 0, true);
+
+        assertEquals(true, clientAtomicStamped.compareAndSet(0, 1, 0, 1));
+        assertEquals(1, clientAtomicStamped.value());
+        assertEquals(1, clientAtomicStamped.stamp());
+
+        final IgniteAtomicStamped srvAtomicStamped = srv.atomicStamped("atomicStamped", 0, 0, false);
+
+        assertEquals(true, srvAtomicStamped.compareAndSet(1, 2, 1, 2));
+        assertEquals(2, srvAtomicStamped.value());
+        assertEquals(2, srvAtomicStamped.stamp());
+
+        reconnectClientNode(client, srv, new Runnable() {
+            @Override public void run() {
+                assertEquals(true, srvAtomicStamped.compareAndSet(2, 3, 2, 3));
+                assertEquals(3, srvAtomicStamped.value());
+                assertEquals(3, srvAtomicStamped.stamp());
+            }
+        });
+
+        assertEquals(true, clientAtomicStamped.compareAndSet(3, 4, 3, 4));
+        assertEquals(4, clientAtomicStamped.value());
+        assertEquals(4, clientAtomicStamped.stamp());
+
+        assertEquals(true, srvAtomicStamped.compareAndSet(4, 5, 4, 5));
+        assertEquals(5, srvAtomicStamped.value());
+        assertEquals(5, srvAtomicStamped.stamp());
+
+        srvAtomicStamped.close();
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testAtomicStampedReconnectRemoved() throws Exception {
+        Ignite client = grid(serverCount());
+
+        assertTrue(client.cluster().localNode().isClient());
+
+        Ignite srv = clientRouter(client);
+
+        final IgniteAtomicStamped clientAtomicStamped = client.atomicStamped("atomicStampedRemoved", 0, 0, true);
+
+        assertEquals(true, clientAtomicStamped.compareAndSet(0, 1, 0, 1));
+        assertEquals(1, clientAtomicStamped.value());
+        assertEquals(1, clientAtomicStamped.stamp());
+
+        final IgniteAtomicStamped srvAtomicStamped = srv.atomicStamped("atomicStampedRemoved", 0, 0, false);
+
+        assertEquals(true, srvAtomicStamped.compareAndSet(1, 2, 1, 2));
+        assertEquals(2, srvAtomicStamped.value());
+        assertEquals(2, srvAtomicStamped.stamp());
+
+        reconnectClientNode(client, srv, new Runnable() {
+            @Override public void run() {
+                srvAtomicStamped.close();
+            }
+        });
+
+        GridTestUtils.assertThrows(log, new Callable<Object>() {
+            @Override public Object call() throws Exception {
+                clientAtomicStamped.compareAndSet(2, 3, 2, 3);
+
+                return null;
+            }
+        }, IllegalStateException.class, null);
+
+        IgniteAtomicStamped newClientAtomicStamped = client.atomicStamped("atomicStampedRemoved", 0, 0, true);
+
+        assertEquals(true, newClientAtomicStamped.compareAndSet(0, 1, 0, 1));
+        assertEquals(1, newClientAtomicStamped.value());
+        assertEquals(1, newClientAtomicStamped.stamp());
+
+        IgniteAtomicStamped newSrvAtomicStamped = srv.atomicStamped("atomicStampedRemoved", 0, 0, false);
+
+        assertEquals(true, newSrvAtomicStamped.compareAndSet(1, 2, 1, 2));
+        assertEquals(2, newSrvAtomicStamped.value());
+        assertEquals(2, newSrvAtomicStamped.stamp());
+
+        newClientAtomicStamped.close();
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testAtomicStampedReconnectInProgress() throws Exception {
+        Ignite client = grid(serverCount());
+
+        assertTrue(client.cluster().localNode().isClient());
+
+        Ignite srv = clientRouter(client);
+
+        final IgniteAtomicStamped clientAtomicStamped = client.atomicStamped("atomicStampedInProgress", 0, 0, true);
+
+        assertEquals(true, clientAtomicStamped.compareAndSet(0, 1, 0, 1));
+        assertEquals(1, clientAtomicStamped.value());
+        assertEquals(1, clientAtomicStamped.stamp());
+
+        IgniteAtomicStamped srvAtomicStamped = srv.atomicStamped("atomicStampedInProgress", 0, 0, false);
+
+        assertEquals(true, srvAtomicStamped.compareAndSet(1, 2, 1, 2));
+        assertEquals(2, srvAtomicStamped.value());
+        assertEquals(2, srvAtomicStamped.stamp());
+
+        BlockTpcCommunicationSpi servCommSpi = commSpi(srv);
+
+        servCommSpi.blockMessage(GridNearLockResponse.class);
+
+        final IgniteInternalFuture<Object> fut = GridTestUtils.runAsync(new Callable<Object>() {
+            @Override public Object call() throws Exception {
+                try {
+                    clientAtomicStamped.compareAndSet(2, 3, 2, 3);
+                }
+                catch (IgniteClientDisconnectedException e) {
+                    checkAndWait(e);
+
+                    return true;
+                }
+
+                return false;
+            }
+        });
+
+        // Check that client waiting operation.
+        GridTestUtils.assertThrows(log, new Callable<Object>() {
+            @Override public Object call() throws Exception {
+                return fut.get(200);
+            }
+        }, IgniteFutureTimeoutCheckedException.class, null);
+
+        assertNotDone(fut);
+
+        servCommSpi.unblockMessage();
+
+        reconnectClientNode(client, srv, null);
+
+        assertTrue((Boolean)fut.get(2, TimeUnit.SECONDS));
+
+        // Check that after reconnect working.
+        assertEquals(true, clientAtomicStamped.compareAndSet(2, 3, 2, 3));
+        assertEquals(3, clientAtomicStamped.value());
+        assertEquals(3, clientAtomicStamped.stamp());
+
+        assertEquals(true, srvAtomicStamped.compareAndSet(3, 4, 3, 4));
+        assertEquals(4, srvAtomicStamped.value());
+        assertEquals(4, srvAtomicStamped.stamp());
+
+        srvAtomicStamped.close();
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testAtomicLongReconnect() throws Exception {
+        Ignite client = grid(serverCount());
+
+        assertTrue(client.cluster().localNode().isClient());
+
+        Ignite srv = clientRouter(client);
+
+        IgniteAtomicLong clientAtomicLong = client.atomicLong("atomicLong", 0, true);
+
+        assertEquals(0L, clientAtomicLong.getAndAdd(1));
+
+        final IgniteAtomicLong srvAtomicLong = srv.atomicLong("atomicLong", 0, false);
+
+        assertEquals(1L, srvAtomicLong.getAndAdd(1));
+
+        reconnectClientNode(client, srv, new Runnable() {
+            @Override public void run() {
+                assertEquals(2L, srvAtomicLong.getAndAdd(1));
+            }
+        });
+
+        assertEquals(3L, clientAtomicLong.getAndAdd(1));
+
+        assertEquals(4L, srvAtomicLong.getAndAdd(1));
+
+        assertEquals(5L, clientAtomicLong.getAndAdd(1));
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testAtomicLongReconnectRemoved() throws Exception {
+        Ignite client = grid(serverCount());
+
+        assertTrue(client.cluster().localNode().isClient());
+
+        Ignite srv = clientRouter(client);
+
+        final IgniteAtomicLong clientAtomicLong = client.atomicLong("atomicLongRmv", 0, true);
+
+        assertEquals(0L, clientAtomicLong.getAndAdd(1));
+
+        final IgniteAtomicLong srvAtomicLong = srv.atomicLong("atomicLongRmv", 0, false);
+
+        assertEquals(1L, srvAtomicLong.getAndAdd(1));
+
+        reconnectClientNode(client, srv, new Runnable() {
+            @Override public void run() {
+                srvAtomicLong.close();
+            }
+        });
+
+        GridTestUtils.assertThrows(log, new Callable<Object>() {
+            @Override public Object call() throws Exception {
+                clientAtomicLong.getAndAdd(1);
+
+                return null;
+            }
+        }, IllegalStateException.class, null);
+
+        IgniteAtomicLong newClientAtomicLong = client.atomicLong("atomicLongRmv", 0, true);
+
+        assertEquals(0L, newClientAtomicLong.getAndAdd(1));
+
+        IgniteAtomicLong newSrvAtomicLong = srv.atomicLong("atomicLongRmv", 0, false);
+
+        assertEquals(1L, newSrvAtomicLong.getAndAdd(1));
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testAtomicLongReconnectInProgress() throws Exception {
+        Ignite client = grid(serverCount());
+
+        assertTrue(client.cluster().localNode().isClient());
+
+        Ignite srv = clientRouter(client);
+
+        BlockTpcCommunicationSpi commSpi = commSpi(srv);
+
+        final IgniteAtomicLong clientAtomicLong = client.atomicLong("atomicLongInProggress", 0, true);
+
+        final IgniteAtomicLong srvAtomicLong = srv.atomicLong("atomicLongInProggress", 0, false);
+
+        commSpi.blockMessage(GridNearLockResponse.class);
+
+        final IgniteInternalFuture<Object> fut = GridTestUtils.runAsync(new Callable<Object>() {
+            @Override public Object call() throws Exception {
+                try {
+                    clientAtomicLong.getAndAdd(1);
+                }
+                catch (IgniteClientDisconnectedException e) {
+                    checkAndWait(e);
+
+                    return true;
+                }
+
+                return false;
+            }
+        });
+
+        // Check that client waiting operation.
+        GridTestUtils.assertThrows(log, new Callable<Object>() {
+            @Override public Object call() throws Exception {
+                return fut.get(200);
+            }
+        }, IgniteFutureTimeoutCheckedException.class, null);
+
+        assertNotDone(fut);
+
+        commSpi.unblockMessage();
+
+        reconnectClientNode(client, srv, null);
+
+        assertTrue((Boolean)fut.get(2, TimeUnit.SECONDS));
+
+        // Check that after reconnect working.
+        assertEquals(1, clientAtomicLong.addAndGet(1));
+        assertEquals(2, srvAtomicLong.addAndGet(1));
+
+        clientAtomicLong.close();
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testLatchReconnect() throws Exception {
+        Ignite client = grid(serverCount());
+
+        assertTrue(client.cluster().localNode().isClient());
+
+        Ignite srv = clientRouter(client);
+
+        IgniteCountDownLatch clientLatch = client.countDownLatch("latch1", 3, false, true);
+
+        assertEquals(3, clientLatch.count());
+
+        final IgniteCountDownLatch srvLatch = srv.countDownLatch("latch1", 3, false, false);
+
+        assertEquals(3, srvLatch.count());
+
+        reconnectClientNode(client, srv, new Runnable() {
+            @Override public void run() {
+                srvLatch.countDown();
+            }
+        });
+
+        assertEquals(2, srvLatch.count());
+        assertEquals(2, clientLatch.count());
+
+        srvLatch.countDown();
+
+        assertEquals(1, srvLatch.count());
+        assertEquals(1, clientLatch.count());
+
+        clientLatch.countDown();
+
+        assertEquals(0, srvLatch.count());
+        assertEquals(0, clientLatch.count());
+
+        assertTrue(srvLatch.await(1000));
+        assertTrue(clientLatch.await(1000));
+    }
+}


Mime
View raw message