#ignite-51: Change test for messaging.

apache · Mar 2, 2015 · 2f43d32 · 2f43d32
1 parent 29508fa
commit 2f43d32
Showing 1 changed file with 79 additions and 9 deletions.
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/GridCacheMessageSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/GridCacheMessageSelfTest.java
@@ -26,20 +26,27 @@ public class GridCacheMessageSelfTest extends GridCommonAbstractTest {
     /** */
     private static final byte DIRECT_TYPE = (byte)202;
 
+    /** */
+    private static final byte DIRECT_TYPE1 = (byte)203;
+
     static {
         GridIoMessageFactory.registerCustom(DIRECT_TYPE, new CO<Message>() {
-            @Override
+            @Override public Message apply() {
-            public Message apply() {
                 return new TestMessage();
             }
         });
 
         GridIoMessageFactory.registerCustom(GridTestMessage.DIRECT_TYPE, new CO<Message>() {
-            @Override
+            @Override public Message apply() {
-            public Message apply() {
                 return new GridTestMessage();
             }
         });
+
+        GridIoMessageFactory.registerCustom(DIRECT_TYPE1, new CO<Message>() {
+            @Override public Message apply() {
+                return new TestMessage1();
+            }
+        });
     }
 
     /** {@inheritDoc} */
@@ -50,7 +57,6 @@ public Message apply() {
 
         CacheConfiguration ccfg = new CacheConfiguration();
 
-        ccfg.setName(null);
         ccfg.setCacheMode(CacheMode.PARTITIONED);
         ccfg.setAtomicityMode(CacheAtomicityMode.ATOMIC);
         ccfg.setWriteSynchronizationMode(FULL_SYNC);
@@ -100,7 +106,11 @@ private void doSend() throws Exception {
         for (int i = 1; i <= SAMPLE_CNT; i++) {
             mgr0.send(grid(1).localNode(), topic, message, GridIoPolicy.PUBLIC_POOL);
 
-            message.add(new GridTestMessage(grid(1).localNode().id(), i, 0));
+            TestMessage1 mes1 = new TestMessage1();
+
+            mes1.init(new GridTestMessage(grid(1).localNode().id(), i, 0));
+
+            message.add(mes1);
         }
 
         assert latch.await(3, SECONDS);
@@ -109,13 +119,13 @@ private void doSend() throws Exception {
     /** */
     private static class TestMessage extends GridCacheMessage {
         /** */
-        @GridDirectCollection(GridTestMessage.class)
+        @GridDirectCollection(TestMessage1.class)
-        private Collection<GridTestMessage> entries = new ArrayList<>();
+        private Collection<TestMessage1> entries = new ArrayList<>();
 
         /**
          * @param entry Entry.
          */
-        public void add(GridTestMessage entry) {
+        public void add(TestMessage1 entry) {
             entries.add(entry);
         }
 
@@ -169,4 +179,64 @@ public void add(GridTestMessage entry) {
             return true;
         }
     }
+
+    private static class TestMessage1 extends GridCacheMessage {
+        GridTestMessage mes;
+
+        public void init(GridTestMessage mes) {
+            this.mes = mes;
+        }
+
+        /** {@inheritDoc} */
+        @Override public byte directType() {
+            return 0;
+        }
+
+        /** {@inheritDoc} */
+        @Override public byte fieldsCount() {
+            return 1;
+        }
+
+        /** {@inheritDoc} */
+        @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
+            writer.setBuffer(buf);
+
+            if (!writer.isHeaderWritten()) {
+                if (!writer.writeHeader(directType(), fieldsCount()))
+                    return false;
+
+                writer.onHeaderWritten();
+            }
+
+            switch (writer.state()) {
+                case 0:
+                    if (!mes.writeTo(buf, writer))
+                        return false;
+
+                    writer.incrementState();
+            }
+
+            return true;
+        }
+
+        /** {@inheritDoc} */
+        @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) {
+            reader.setBuffer(buf);
+
+            if (!reader.beforeMessageRead())
+                return false;
+
+            switch (reader.state()) {
+                case 0:
+                    mes.readFrom(buf, reader);
+
+                    if (!reader.isLastRead())
+                        return false;
+
+                    reader.incrementState();
+            }
+
+            return true;
+        }
+    }
 }