modules/core/src/test/java/org/apache/ignite/spi/GridTcpSpiForwardingSelfTest.java

/*
 * 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.spi;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Callable;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.AddressResolver;
import org.apache.ignite.configuration.BasicAddressResolver;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.util.nio.GridCommunicationClient;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.lang.IgniteCallable;
import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.internal.TcpDiscoveryNode;
import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.junit.Test;

/**
 * Test for {@link TcpDiscoverySpi} and {@link TcpCommunicationSpi}.
 */
public class GridTcpSpiForwardingSelfTest extends GridCommonAbstractTest {
    /** */
    private static final int locPort1 = 47500;

    /** */
    private static final int locPort2 = 48500;

    /** */
    private static final int extPort1 = 10000;

    /** */
    private static final int extPort2 = 20000;

    /** */
    private static final int commLocPort1 = 47100;

    /** */
    private static final int commLocPort2 = 48100;

    /** */
    private static final int commExtPort1 = 10100;

    /** */
    private static final int commExtPort2 = 20100;

    /** */
    private AddressResolver rslvr;

    /** */
    private boolean ipFinderUseLocPorts;

    /** {@inheritDoc} */
    @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
        TcpDiscoveryVmIpFinder ipFinder = new TcpDiscoveryVmIpFinder();

        if (ipFinderUseLocPorts)
            ipFinder.setAddresses(Arrays.asList("127.0.0.1:" + locPort1, "127.0.0.1:" + locPort2));
        else
            ipFinder.setAddresses(Arrays.asList("127.0.0.1:" + extPort1, "127.0.0.1:" + extPort2));

        TcpDiscoverySpi spi = new TcpDiscoverySpi();

        final int locPort;
        final int commLocPort;

        if (getTestIgniteInstanceName(0).equals(igniteInstanceName)) {
            locPort = locPort1;
            commLocPort = commLocPort1;
        }
        else if (getTestIgniteInstanceName(1).equals(igniteInstanceName)) {
            locPort = locPort2;
            commLocPort = commLocPort2;
        }
        else
            throw new IllegalArgumentException("Unknown Ignite instance name");

        spi.setIpFinder(ipFinder);

        IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);

        spi.setLocalPort(locPort);
        spi.setLocalPortRange(1);
        cfg.setDiscoverySpi(spi);
        cfg.setLocalHost("127.0.0.1");
        cfg.setConnectorConfiguration(null);

        TcpCommunicationSpi commSpi = new TcpCommunicationSpi() {
            @Override protected GridCommunicationClient createTcpClient(ClusterNode node, int connIdx)
                throws IgniteCheckedException {
                Map<String, Object> attrs = new HashMap<>(node.attributes());

                attrs.remove(createSpiAttributeName(ATTR_PORT));

                ((TcpDiscoveryNode)node).setAttributes(attrs);

                return super.createTcpClient(node, connIdx);
            }
        };

        commSpi.setLocalAddress("127.0.0.1");
        commSpi.setLocalPort(commLocPort);
        commSpi.setLocalPortRange(1);
        commSpi.setSharedMemoryPort(-1);
        commSpi.setConnectionsPerNode(1);

        cfg.setCommunicationSpi(commSpi);

        assert rslvr != null;

        cfg.setAddressResolver(rslvr);

        return cfg;
    }

    /**
     * @throws Exception If any error occurs.
     */
    @Test
    public void testCustomResolver() throws Exception {
        final Map<InetSocketAddress, Collection<InetSocketAddress>> map = new HashMap<>();

        map.put(new InetSocketAddress("127.0.0.1", locPort1), F.asList(new InetSocketAddress("127.0.0.1", extPort1)));
        map.put(new InetSocketAddress("127.0.0.1", commLocPort1), F.asList(new InetSocketAddress("127.0.0.1", commExtPort1)));
        map.put(new InetSocketAddress("127.0.0.1", locPort2), F.asList(new InetSocketAddress("127.0.0.1", extPort2)));
        map.put(new InetSocketAddress("127.0.0.1", commLocPort2), F.asList(new InetSocketAddress("127.0.0.1", commExtPort2)));

        rslvr = new AddressResolver() {
            @Override public Collection<InetSocketAddress> getExternalAddresses(InetSocketAddress addr) {
                return map.get(addr);
            }
        };

        doTestForward();
    }

    /**
     * @throws Exception If failed.
     */
    @Test
    public void testBasicResolverMapPorts() throws Exception {
        Map<String, String> map = new HashMap<>();

        map.put("127.0.0.1:" + locPort1, "127.0.0.1:" + extPort1);
        map.put("127.0.0.1:" + commLocPort1, "127.0.0.1:" + commExtPort1);
        map.put("127.0.0.1:" + locPort2, "127.0.0.1:" + extPort2);
        map.put("127.0.0.1:" + commLocPort2, "127.0.0.1:" + commExtPort2);

        rslvr = new BasicAddressResolver(map);

        doTestForward();
    }

    /**
     * @throws Exception If failed.
     */
    @Test
    public void testBasicResolverMapAddress() throws Exception {
        Map<String, String> map = new HashMap<>();

        map.put("127.0.0.1", "127.0.0.1");

        rslvr = new BasicAddressResolver(map);

        ipFinderUseLocPorts = true;

        doTestForward();
    }

    /**
     * @throws Exception If failed.
     */
    @Test
    public void testBasicResolverErrors() throws Exception {
        GridTestUtils.assertThrows(
            log,
            new Callable<Object>() {
                @Override public Object call() throws Exception {
                    return new BasicAddressResolver(null);
                }
            },
            IllegalArgumentException.class,
            "At least one address mapping is required."
        );

        GridTestUtils.assertThrows(
            log,
            new Callable<Object>() {
                @Override public Object call() throws Exception {
                    return new BasicAddressResolver(new HashMap<String, String>());
                }
            },
            IllegalArgumentException.class,
            "At least one address mapping is required."
        );

        GridTestUtils.assertThrows(
            log,
            new Callable<Object>() {
                @Override public Object call() throws Exception {
                    Map<String, String> map = new HashMap<>();

                    map.put("from", null);

                    return new BasicAddressResolver(map);
                }
            },
            IllegalArgumentException.class,
            "Invalid address mapping: from=null"
        );

        GridTestUtils.assertThrows(
            log,
            new Callable<Object>() {
                @Override public Object call() throws Exception {
                    Map<String, String> map = new HashMap<>();

                    map.put("from", "");

                    return new BasicAddressResolver(map);
                }
            },
            IllegalArgumentException.class,
            "Invalid address mapping: from="
        );

        GridTestUtils.assertThrows(
            log,
            new Callable<Object>() {
                @Override public Object call() throws Exception {
                    Map<String, String> map = new HashMap<>();

                    map.put(null, "to");

                    return new BasicAddressResolver(map);
                }
            },
            IllegalArgumentException.class,
            "Invalid address mapping: null=to"
        );

        GridTestUtils.assertThrows(
            log,
            new Callable<Object>() {
                @Override public Object call() throws Exception {
                    Map<String, String> map = new HashMap<>();

                    map.put("", "to");

                    return new BasicAddressResolver(map);
                }
            },
            IllegalArgumentException.class,
            "Invalid address mapping: =to"
        );

        GridTestUtils.assertThrows(
            log,
            new Callable<Object>() {
                @Override public Object call() throws Exception {
                    Map<String, String> map = new HashMap<>();

                    map.put("from", "to:1111");

                    return new BasicAddressResolver(map);
                }
            },
            IllegalArgumentException.class,
            "Invalid address mapping: from=to:1111"
        );

        GridTestUtils.assertThrows(
            log,
            new Callable<Object>() {
                @Override public Object call() throws Exception {
                    Map<String, String> map = new HashMap<>();

                    map.put("from:1111", "to");

                    return new BasicAddressResolver(map);
                }
            },
            IllegalArgumentException.class,
            "Invalid address mapping: from:1111=to"
        );

        GridTestUtils.assertThrows(
            log,
            new Callable<Object>() {
                @Override public Object call() throws Exception {
                    Map<String, String> map = new HashMap<>();

                    map.put("from:1111:2222", "to:1111");

                    return new BasicAddressResolver(map);
                }
            },
            IllegalArgumentException.class,
            "Invalid address mapping: from:1111:2222=to:1111"
        );

        GridTestUtils.assertThrows(
            log,
            new Callable<Object>() {
                @Override public Object call() throws Exception {
                    Map<String, String> map = new HashMap<>();

                    map.put("from:1111", "to:1111:2222");

                    return new BasicAddressResolver(map);
                }
            },
            IllegalArgumentException.class,
            "Invalid address mapping: from:1111=to:1111:2222"
        );
    }

    /**
     * @throws Exception If failed.
     */
    private void doTestForward() throws Exception {
        InetAddress locHost = InetAddress.getByName("127.0.0.1");

        try (
            GridTcpForwarder tcpForward1 = new GridTcpForwarder(locHost, extPort1, locHost, locPort1, log);
            GridTcpForwarder tcpForward2 = new GridTcpForwarder(locHost, extPort2, locHost, locPort2, log);
            GridTcpForwarder tcpForward3 = new GridTcpForwarder(locHost, commExtPort1, locHost, commLocPort1, log);
            GridTcpForwarder tcpForward4 = new GridTcpForwarder(locHost, commExtPort2, locHost, commLocPort2, log);

            Ignite g1 = startGrid(0);
            Ignite g2 = startGrid(1)
        ) {
            assertEquals(2, grid(0).cluster().nodes().size());
            assertEquals(2, grid(1).cluster().nodes().size());

            Collection<Integer> t = g1.compute().broadcast(new IgniteCallable<Integer>() {
                @Override public Integer call() throws Exception {
                    return 13;
                }
            });

            assertEquals(F.asList(13, 13), t);
        }
    }
}