ignite-1.5 - MessageCollection + marshalling issue test
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/322a85a3 Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/322a85a3 Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/322a85a3 Branch: refs/heads/ignite-1.5.1 Commit: 322a85a359e0fc2c56f5c3aa38fc48a92e553289 Parents: c077522 Author: S.Vladykin <svlady...@gridgain.com> Authored: Tue Dec 8 13:17:36 2015 +0300 Committer: S.Vladykin <svlady...@gridgain.com> Committed: Tue Dec 8 13:17:36 2015 +0300 ---------------------------------------------------------------------- .../ignite/codegen/MessageCodeGenerator.java | 6 +- .../communication/GridIoMessageFactory.java | 8 +- .../internal/util/GridMessageCollection.java | 157 +++++++++++++++++++ .../internal/util/UUIDCollectionMessage.java | 42 ++++- .../testsuites/IgniteUtilSelfTestSuite.java | 2 + .../ignite/util/GridMessageCollectionTest.java | 105 +++++++++++++ 6 files changed, 314 insertions(+), 6 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/ignite/blob/322a85a3/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java ---------------------------------------------------------------------- diff --git a/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java b/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java index 8733bb3..587ad06 100644 --- a/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java +++ b/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java @@ -43,9 +43,6 @@ import org.apache.ignite.internal.GridDirectCollection; import org.apache.ignite.internal.GridDirectMap; import org.apache.ignite.internal.GridDirectTransient; import org.apache.ignite.internal.IgniteCodeGeneratingFail; -import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridNearAtomicUpdateRequest; -import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryEntry; -import org.apache.ignite.internal.util.UUIDCollectionMessage; import org.apache.ignite.internal.util.typedef.internal.SB; import org.apache.ignite.internal.util.typedef.internal.U; import org.apache.ignite.lang.IgniteUuid; @@ -170,8 +167,9 @@ public class MessageCodeGenerator { // gen.generateAll(true); - gen.generateAndWrite(GridNearAtomicUpdateRequest.class); +// gen.generateAndWrite(GridNearAtomicUpdateRequest.class); +// gen.generateAndWrite(GridMessageCollection.class); // gen.generateAndWrite(DataStreamerEntry.class); // gen.generateAndWrite(GridDistributedLockRequest.class); http://git-wip-us.apache.org/repos/asf/ignite/blob/322a85a3/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java index c9b8e27..b0ef3f4 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java @@ -125,6 +125,7 @@ import org.apache.ignite.internal.processors.rest.handlers.task.GridTaskResultRe import org.apache.ignite.internal.processors.rest.handlers.task.GridTaskResultResponse; import org.apache.ignite.internal.util.GridByteArrayList; import org.apache.ignite.internal.util.GridLongList; +import org.apache.ignite.internal.util.GridMessageCollection; import org.apache.ignite.internal.util.UUIDCollectionMessage; import org.apache.ignite.lang.IgniteOutClosure; import org.apache.ignite.plugin.extensions.communication.Message; @@ -720,7 +721,12 @@ public class GridIoMessageFactory implements MessageFactory { break; - // [-3..119] - this + case 124: + msg = new GridMessageCollection<>(); + + break; + + // [-3..119] [124] - this // [120..123] - DR // [-4..-22] - SQL default: http://git-wip-us.apache.org/repos/asf/ignite/blob/322a85a3/modules/core/src/main/java/org/apache/ignite/internal/util/GridMessageCollection.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/GridMessageCollection.java b/modules/core/src/main/java/org/apache/ignite/internal/util/GridMessageCollection.java new file mode 100644 index 0000000..09a5e96 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/util/GridMessageCollection.java @@ -0,0 +1,157 @@ +/* + * 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.util; + +import java.nio.ByteBuffer; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import org.apache.ignite.internal.GridDirectCollection; +import org.apache.ignite.internal.util.typedef.internal.S; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType; +import org.apache.ignite.plugin.extensions.communication.MessageReader; +import org.apache.ignite.plugin.extensions.communication.MessageWriter; + +/** + * Collection of messages. + */ +public final class GridMessageCollection<M extends Message> implements Message { + /** */ + @GridDirectCollection(Message.class) + private Collection<M> msgs; + + /** + * + */ + public GridMessageCollection() { + // No-op. + } + + /** + * @param msgs Collection of messages. + */ + public GridMessageCollection(Collection<M> msgs) { + this.msgs = msgs; + } + + /** + * @param msgs Messages. + * @return Message list. + */ + public static <X extends Message> GridMessageCollection<X> of(X... msgs) { + if (msgs == null || msgs.length == 0) + return null; + + List<X> list = msgs.length == 1 ? Collections.singletonList(msgs[0]) : Arrays.asList(msgs); + + return new GridMessageCollection<>(list); + } + + /** + * @return Messages. + */ + public Collection<M> messages() { + return msgs; + } + + /** + * @param msgs Messages. + */ + public void messages(Collection<M> msgs) { + this.msgs = msgs; + } + + /** {@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 (!writer.writeCollection("msgs", msgs, MessageCollectionItemType.MSG)) + 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: + msgs = reader.readCollection("msgs", MessageCollectionItemType.MSG); + + if (!reader.isLastRead()) + return false; + + reader.incrementState(); + + } + + return reader.afterMessageRead(GridMessageCollection.class); + } + + /** {@inheritDoc} */ + @Override public byte directType() { + return 124; + } + + /** {@inheritDoc} */ + @Override public byte fieldsCount() { + return 1; + } + + /** {@inheritDoc} */ + @Override public boolean equals(Object o) { + if (this == o) + return true; + + if (o == null || getClass() != o.getClass()) + return false; + + GridMessageCollection<?> that = (GridMessageCollection<?>)o; + + return msgs == that.msgs || (msgs != null && msgs.equals(that.msgs)); + } + + /** {@inheritDoc} */ + @Override public int hashCode() { + return msgs != null ? msgs.hashCode() : 0; + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(GridMessageCollection.class, this, "msgsSize", msgs == null ? null : msgs.size()); + } +} http://git-wip-us.apache.org/repos/asf/ignite/blob/322a85a3/modules/core/src/main/java/org/apache/ignite/internal/util/UUIDCollectionMessage.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/UUIDCollectionMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/util/UUIDCollectionMessage.java index 25e3376..33bce53 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/util/UUIDCollectionMessage.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/util/UUIDCollectionMessage.java @@ -18,16 +18,20 @@ package org.apache.ignite.internal.util; import java.nio.ByteBuffer; +import java.util.Arrays; import java.util.Collection; +import java.util.Collections; +import java.util.List; import java.util.UUID; import org.apache.ignite.internal.GridDirectCollection; +import org.apache.ignite.internal.util.typedef.internal.S; import org.apache.ignite.plugin.extensions.communication.Message; import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType; import org.apache.ignite.plugin.extensions.communication.MessageReader; import org.apache.ignite.plugin.extensions.communication.MessageWriter; /** - * + * Collection of UUIDs. */ public class UUIDCollectionMessage implements Message { /** */ @@ -52,6 +56,19 @@ public class UUIDCollectionMessage implements Message { } /** + * @param uuids UUIDs. + * @return Message. + */ + public static UUIDCollectionMessage of(UUID... uuids) { + if (uuids == null || uuids.length == 0) + return null; + + List<UUID> list = uuids.length == 1 ? Collections.singletonList(uuids[0]) : Arrays.asList(uuids); + + return new UUIDCollectionMessage(list); + } + + /** * @return The collection of UUIDs that was wrapped. */ public Collection<UUID> uuids() { @@ -111,4 +128,27 @@ public class UUIDCollectionMessage implements Message { @Override public byte fieldsCount() { return 1; } + + /** {@inheritDoc} */ + @Override public boolean equals(Object o) { + if (this == o) + return true; + + if (o == null || getClass() != o.getClass()) + return false; + + UUIDCollectionMessage that = (UUIDCollectionMessage)o; + + return uuids == that.uuids || (uuids != null && uuids.equals(that.uuids)); + } + + /** {@inheritDoc} */ + @Override public int hashCode() { + return uuids != null ? uuids.hashCode() : 0; + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(UUIDCollectionMessage.class, this, "uuidsSize", uuids == null ? null : uuids.size()); + } } http://git-wip-us.apache.org/repos/asf/ignite/blob/322a85a3/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteUtilSelfTestSuite.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteUtilSelfTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteUtilSelfTestSuite.java index ac3241d..bd7bb96 100644 --- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteUtilSelfTestSuite.java +++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteUtilSelfTestSuite.java @@ -37,6 +37,7 @@ import org.apache.ignite.spi.discovery.ClusterMetricsSnapshotSerializeSelfTest; import org.apache.ignite.testframework.GridTestUtils; import org.apache.ignite.thread.GridThreadPoolExecutorServiceSelfTest; import org.apache.ignite.util.GridLongListSelfTest; +import org.apache.ignite.util.GridMessageCollectionTest; import org.apache.ignite.util.GridQueueSelfTest; import org.apache.ignite.util.GridSpinReadWriteLockSelfTest; import org.apache.ignite.util.GridStringBuilderFactorySelfTest; @@ -73,6 +74,7 @@ public class IgniteUtilSelfTestSuite extends TestSuite { suite.addTestSuite(GridLongListSelfTest.class); suite.addTestSuite(GridCacheUtilsSelfTest.class); suite.addTestSuite(IgniteExceptionRegistrySelfTest.class); + suite.addTestSuite(GridMessageCollectionTest.class); // Metrics. suite.addTestSuite(ClusterMetricsSnapshotSerializeSelfTest.class); http://git-wip-us.apache.org/repos/asf/ignite/blob/322a85a3/modules/core/src/test/java/org/apache/ignite/util/GridMessageCollectionTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/util/GridMessageCollectionTest.java b/modules/core/src/test/java/org/apache/ignite/util/GridMessageCollectionTest.java new file mode 100644 index 0000000..e910a8a --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/util/GridMessageCollectionTest.java @@ -0,0 +1,105 @@ +/* + * 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.util; + +import java.nio.ByteBuffer; +import junit.framework.TestCase; +import org.apache.ignite.internal.direct.DirectMessageReader; +import org.apache.ignite.internal.direct.DirectMessageWriter; +import org.apache.ignite.internal.managers.communication.GridIoMessageFactory; +import org.apache.ignite.internal.util.UUIDCollectionMessage; +import org.apache.ignite.plugin.extensions.communication.Message; + +import static java.util.UUID.randomUUID; +import static org.apache.ignite.internal.util.GridMessageCollection.of; + +/** + * + */ +public class GridMessageCollectionTest extends TestCase { + /** */ + private byte proto; + + /** + * + */ + public void testMarshal() { + UUIDCollectionMessage um0 = UUIDCollectionMessage.of(); + UUIDCollectionMessage um1 = UUIDCollectionMessage.of(randomUUID()); + UUIDCollectionMessage um2 = UUIDCollectionMessage.of(randomUUID(), randomUUID()); + UUIDCollectionMessage um3 = UUIDCollectionMessage.of(randomUUID(), randomUUID(), randomUUID()); + + assertNull(um0); + assertEquals(3, um3.uuids().size()); + + proto = 2; + doTestMarshal(um0, um1, um2, um3); + + proto = 1; + doTestMarshal(um0, um1, um2, um3); + } + + /** + * @param um0 Null. + * @param um1 One uuid list. + * @param um2 Two uuid list. + * @param um3 Three uuid list. + */ + private void doTestMarshal( + UUIDCollectionMessage um0, + UUIDCollectionMessage um1, + UUIDCollectionMessage um2, + UUIDCollectionMessage um3 + ) { + doTestMarshal(um1); + doTestMarshal(um2); + doTestMarshal(um3); + + doTestMarshal(of(um0)); + doTestMarshal(of(um1)); + doTestMarshal(of(um2)); + doTestMarshal(of(um3)); + + doTestMarshal(of(um2, um3)); + doTestMarshal(of(um1, um0, um3)); + + doTestMarshal(of(of(um3), of(um2))); + doTestMarshal(of(of(of(of(of(um0))), um1, of(um3)))); + } + + /** + * @param m Message. + */ + private void doTestMarshal(Message m) { + ByteBuffer buf = ByteBuffer.allocate(8 * 1024); + + DirectMessageWriter w = new DirectMessageWriter(proto); + + m.writeTo(buf, w); + + buf.flip(); + + DirectMessageReader r = new DirectMessageReader(new GridIoMessageFactory(null), proto); + + r.setBuffer(buf); + + Message mx = r.readMessage(null); + + assertEquals(m, mx); + } +}