Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -58,11 +58,6 @@ default boolean readFrom(ByteBuffer buf, MessageReader reader) {
throw new UnsupportedOperationException();
}

/**
* Gets message type.
*
* @return Message type.
*/
/**
* Gets message type.
*
Expand All @@ -72,9 +67,8 @@ default short directType() {
var clazz = getClass();
Short type = REGISTRATIONS.get(clazz);

if (type == null) {
if (type == null)
throw new IgniteException("No registration for class " + clazz.getSimpleName());
}

return type;
}
Expand All @@ -91,8 +85,7 @@ default void registerAsDirectType(short directType) {
var clazz = getClass();
var type = REGISTRATIONS.putIfAbsent(clazz, directType);

if ((type != null) && (type != directType)) {
if ((type != null) && (type != directType))
throw new IgniteException(clazz.getSimpleName() + " is already registered for direct type " + type);
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -31,5 +31,4 @@ public class HandshakeWaitMessage implements Message {
@Override public String toString() {
return S.toString(HandshakeWaitMessage.class, this);
}

}
Original file line number Diff line number Diff line change
Expand Up @@ -23,18 +23,17 @@
import org.apache.ignite.cache.query.ContinuousQuery;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
import org.apache.ignite.internal.processors.metric.impl.MaxValueMetric;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.spi.MessagesPluginProvider;
import org.apache.ignite.spi.discovery.tcp.BlockTcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.DummyCustomDiscoveryMessage;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.ListeningTestLogger;
import org.apache.ignite.testframework.LogListener;
import org.apache.ignite.testframework.junits.WithSystemProperty;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.jetbrains.annotations.Nullable;
import org.junit.Test;

import static org.apache.ignite.internal.managers.discovery.GridDiscoveryManager.DISCO_METRICS;
Expand All @@ -58,6 +57,8 @@ public class OutboundIoMessageQueueSizeTest extends GridCommonAbstractTest {
cfg.setDiscoverySpi(new BlockTcpDiscoverySpi().setIpFinder(((TcpDiscoverySpi)cfg.getDiscoverySpi()).getIpFinder()));
cfg.setGridLogger(log);

cfg.setPluginProviders(new MessagesPluginProvider(DummyCustomDiscoveryMessage.class));

return cfg;
}

Expand Down Expand Up @@ -124,7 +125,7 @@ public void testDiscoveryMsgQueue() throws Exception {

metric.reset(); // Reset value accumulated before discovery SPI startup.

srv0.context().discovery().sendCustomEvent(new DummyCustomDiscoveryMessage(IgniteUuid.randomUuid()));
srv0.context().discovery().sendCustomEvent(new DummyCustomDiscoveryMessage());

// Assume our message can be added to queue concurrently with other messages
// (for example, with metrics update message).
Expand All @@ -142,35 +143,12 @@ public void testDiscoveryMsgQueue() throws Exception {

try {
for (int i = 0; i <= MSG_LIMIT; i++)
srv0.context().discovery().sendCustomEvent(new DummyCustomDiscoveryMessage(IgniteUuid.randomUuid()));
srv0.context().discovery().sendCustomEvent(new DummyCustomDiscoveryMessage());

assertTrue(metric.value() >= MSG_LIMIT);
}
finally {
latch.countDown();
}
}

/** */
private static class DummyCustomDiscoveryMessage implements DiscoveryCustomMessage {
/** */
private final IgniteUuid id;

/**
* @param id Message id.
*/
DummyCustomDiscoveryMessage(IgniteUuid id) {
this.id = id;
}

/** {@inheritDoc} */
@Override public IgniteUuid id() {
return id;
}

/** {@inheritDoc} */
@Nullable @Override public DiscoveryCustomMessage ackMessage() {
return null;
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,7 @@
import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.spi.discovery.DiscoverySpiCustomMessage;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.TestTcpDiscoverySpi;
import org.apache.ignite.spi.encryption.keystore.KeystoreEncryptionSpi;
import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
Expand Down Expand Up @@ -911,7 +912,7 @@ protected static BlockingCustomMessageDiscoverySpi discoSpi(IgniteEx ignite) {
}

/** */
protected static class BlockingCustomMessageDiscoverySpi extends TcpDiscoverySpi {
protected static class BlockingCustomMessageDiscoverySpi extends TestTcpDiscoverySpi {
/** List of messages which have been blocked. */
private final List<DiscoveryCustomMessage> blocked = new CopyOnWriteArrayList<>();

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -36,8 +36,7 @@
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.processors.cache.persistence.filename.SnapshotFileTree;
import org.apache.ignite.internal.util.distributed.MessagesPluginProvider;
import org.apache.ignite.internal.util.distributed.MessagesPluginProvider.MessagesInjectedTcpDiscoverySpi;
import org.apache.ignite.internal.util.distributed.TestIntegerMessage;
import org.apache.ignite.internal.util.distributed.TestUuidMessage;
import org.apache.ignite.internal.util.typedef.G;
import org.apache.ignite.internal.util.typedef.internal.U;
Expand All @@ -48,6 +47,7 @@
import org.apache.ignite.plugin.PluginContext;
import org.apache.ignite.plugin.PluginProvider;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.spi.MessagesPluginProvider;
import org.apache.ignite.testframework.GridTestUtils;
import org.jetbrains.annotations.Nullable;
import org.junit.Test;
Expand Down Expand Up @@ -76,8 +76,10 @@ public class IgniteClusterSnapshotHandlerTest extends IgniteClusterSnapshotResto
/** {@inheritDoc} */
@Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
return super.getConfiguration(igniteInstanceName)
.setPluginProviders(pluginProvider, new MessagesPluginProvider())
.setDiscoverySpi(new MessagesInjectedTcpDiscoverySpi());
.setPluginProviders(
pluginProvider,
new MessagesPluginProvider(TestIntegerMessage.class, TestUuidMessage.class)
);
}

/** {@inheritDoc} */
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -34,10 +34,9 @@
import org.apache.ignite.failure.StopNodeOrHaltFailureHandler;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.events.DiscoveryCustomEvent;
import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
import org.apache.ignite.internal.managers.discovery.SecurityAwareCustomMessageWrapper;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.spi.MessagesPluginProvider;
import org.apache.ignite.spi.discovery.DiscoverySpi;
import org.apache.ignite.spi.discovery.DiscoverySpiCustomMessage;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
Expand Down Expand Up @@ -87,6 +86,8 @@ public class NodeSecurityContextPropagationTest extends GridCommonAbstractTest {
.setIpFinder(new TcpDiscoveryVmIpFinder()
.setAddresses(Collections.singleton("127.0.0.1:47500")));

cfg.setPluginProviders(new MessagesPluginProvider(TestDiscoveryMessage.class, TestDiscoveryAcknowledgeMessage.class));

return cfg;
}

Expand Down Expand Up @@ -218,30 +219,6 @@ private Object discoveryRingMessageWorker(IgniteEx ignite) {
return U.field(impl, "msgWorker");
}

/** */
public static class TestDiscoveryMessage extends AbstractTestDiscoveryMessage {
/** {@inheritDoc} */
@Override public @Nullable DiscoveryCustomMessage ackMessage() {
return new TestDiscoveryAcknowledgeMessage();
}
}

/** */
public static class TestDiscoveryAcknowledgeMessage extends AbstractTestDiscoveryMessage { }

/** */
public abstract static class AbstractTestDiscoveryMessage implements DiscoveryCustomMessage {
/** {@inheritDoc} */
@Override public IgniteUuid id() {
return IgniteUuid.randomUuid();
}

/** {@inheritDoc} */
@Override public @Nullable DiscoveryCustomMessage ackMessage() {
return null;
}
}

/** */
public static class BlockingDequeWrapper<T> implements BlockingDeque<T> {
/** */
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,47 @@
/*
* 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.processors.security;

import org.apache.ignite.internal.Order;
import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.plugin.extensions.communication.MessageFactory;
import org.jetbrains.annotations.Nullable;

/** */
public class TestDiscoveryAcknowledgeMessage implements DiscoveryCustomMessage, Message {
/** */
@Order(0)
IgniteUuid id = IgniteUuid.randomUuid();

/** Constructor for {@link MessageFactory}. */
public TestDiscoveryAcknowledgeMessage() {
// No-op.
}

/** {@inheritDoc} */
@Override public IgniteUuid id() {
return id;
}

/** {@inheritDoc} */
@Override public @Nullable DiscoveryCustomMessage ackMessage() {
return null;
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,47 @@
/*
* 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.processors.security;

import org.apache.ignite.internal.Order;
import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.plugin.extensions.communication.MessageFactory;
import org.jetbrains.annotations.Nullable;

/** */
public class TestDiscoveryMessage implements DiscoveryCustomMessage, Message {
/** */
@Order(0)
IgniteUuid id = IgniteUuid.randomUuid();

/** Constructor for {@link MessageFactory}. */
public TestDiscoveryMessage() {
// No-op.
}

/** {@inheritDoc} */
@Override public IgniteUuid id() {
return id;
}

/** {@inheritDoc} */
@Override public @Nullable DiscoveryCustomMessage ackMessage() {
return new TestDiscoveryAcknowledgeMessage();
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -28,10 +28,10 @@
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.TestRecordingCommunicationSpi;
import org.apache.ignite.internal.util.distributed.MessagesPluginProvider.MessagesInjectedTcpDiscoverySpi;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.spi.MessagesPluginProvider;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.junit.Test;

Expand Down Expand Up @@ -70,9 +70,7 @@ public class DistributedProcessClientAwaitTest extends GridCommonAbstractTest {
IgniteConfiguration cfg = super.getConfiguration(instanceName);

cfg.setCommunicationSpi(new TestRecordingCommunicationSpi());

cfg.setPluginProviders(new MessagesPluginProvider());
cfg.setDiscoverySpi(new MessagesInjectedTcpDiscoverySpi());
cfg.setPluginProviders(new MessagesPluginProvider(TestIntegerMessage.class, TestUuidMessage.class));

return cfg;
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -28,8 +28,8 @@
import org.apache.ignite.failure.FailureHandler;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.util.distributed.MessagesPluginProvider.MessagesInjectedTcpDiscoverySpi;
import org.apache.ignite.internal.util.typedef.G;
import org.apache.ignite.spi.MessagesPluginProvider;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.junit.Test;

Expand Down Expand Up @@ -86,8 +86,7 @@ public class DistributedProcessCoordinatorLeftTest extends GridCommonAbstractTes
}
});

cfg.setPluginProviders(new MessagesPluginProvider());
cfg.setDiscoverySpi(new MessagesInjectedTcpDiscoverySpi());
cfg.setPluginProviders(new MessagesPluginProvider(TestIntegerMessage.class, TestUuidMessage.class));

return cfg;
}
Expand Down
Loading
Loading