Merge branch 'master' of github.com:OpenHFT/Chronicle-Map

Conflicts:
	src/test/java/net/openhft/chronicle/map/example/WordCount.java
This commit is contained in:
leventov
2014-10-03 07:15:14 +04:00
4 changed files with 543 additions and 332 deletions
@@ -1,38 +1,180 @@
package net.openhft.chronicle.map;
import junit.framework.TestCase;
import net.openhft.lang.io.ByteBufferBytes;
import org.junit.Ignore;
import org.junit.Test;
import org.testng.Assert;
import java.io.IOException;
import java.net.Inet4Address;
import static java.nio.ByteBuffer.allocate;
public class NodeDiscoveryBroadcasterTest extends TestCase {
public void test() {
// test to stop it erroring with no test
}
private ChronicleMap<Integer, CharSequence> map1;
private ChronicleMap<Integer, CharSequence> map2;
private volatile byte proposedIdentifier = 1;
private volatile boolean useAnotherIdentifer = false;
public static final int SERVER2_IDENTIFER = 10;
public static final int PROPOSED_IDENTIFIER = 5;
@Test
@Ignore
public void test2() throws IOException, InterruptedException {
public void test3() throws IOException, InterruptedException {
DiscoveryCluster discoveryCluster = new DiscoveryCluster();
// write broadcast our address on the bootstrap method
byte[] server1Address = Inet4Address.getLocalHost().getAddress();
AddressAndPort server1AddressAndPort = new AddressAndPort(server1Address, (short) 1234);
// write broadcast our address on the bootstrap method
byte[] server2Address = Inet4Address.getByName("129.168.0.2").getAddress();
AddressAndPort server2AddressAndPort = new AddressAndPort(server1Address, (short) 1234);
KnownNodes server1KnownNodes = new KnownNodes();
BytesExternalizableImpl server1BytesExternalizable = new BytesExternalizableImpl(server1KnownNodes, null);
// for the unit test we are using a ByteBufferBytes, but iltimately this data would have been send
// and received via UDP
ByteBufferBytes udpData = new ByteBufferBytes(allocate(1024));
// SERER1 - is the new node join the cluster
// we will first send the boostrap along with our host and port
{
BytesExternalizableImpl.ProposedNodes proposedNodes = new BytesExternalizableImpl.ProposedNodes(server1AddressAndPort, (byte) -1);
server1BytesExternalizable.sendBootStrap(proposedNodes);
server1BytesExternalizable.writeMarshallable(udpData);
}
KnownNodes server2KnownNodes = new KnownNodes();
server2KnownNodes.add(server2AddressAndPort, (byte) SERVER2_IDENTIFER);
BytesExternalizableImpl server2BytesExternalizable = new BytesExternalizableImpl(server2KnownNodes, null);
// add our identifier and host:port to the list of known identifiers
// SERVER 2 - is the node already in the cluster
// read the bootstrap along with the host and port it came from
{
udpData.flip();
server2BytesExternalizable.readMarshallable(udpData);
Assert.assertTrue(server2BytesExternalizable.proposedIdentifiersWithHost.values().contains
(new BytesExternalizableImpl.ProposedNodes(server1AddressAndPort, (byte) -1)));
}
// SERVER 2
// write/broadcast the the response to receiving a bootstrap message
{
//ByteBufferBytes bitSetBytes = new ByteBufferBytes(allocate(128 / 8));
// final DirectBitSet knownIdentifiers = new ATSDirectBitSet(bitSetBytes);
// RemoteNodes remoteNodes = new RemoteNodes(bitSetBytes);
udpData.clear();
// broadcasts in response to the bootstrap all the host:ports and identifiers that it knows about
// in this case given that there is only this node in the grid and one node joining,
// it will only be the node that has the bootstrap and this node, who's data will be sent
server2BytesExternalizable.writeMarshallable(udpData);
}
// SERER1
// receive
// read the bootstrap along with the host and port it came from
{
udpData.flip();
server1BytesExternalizable.readMarshallable(udpData);
Assert.assertTrue(server1BytesExternalizable.getRemoteNodes().addressAndPorts().contains
(server2AddressAndPort));
Assert.assertTrue(server1BytesExternalizable.getRemoteNodes().activeIdentifierBitSet().get
(SERVER2_IDENTIFER));
}
// SERVER 1
// send/broadcast - propose an identifier to use, by choosing one that is not used already
// for the purpose of this test we wont choose one at random ( as the code does ) but just pick
// identifier=5
{
udpData.clear();
KnownNodes knownNodes = new KnownNodes();
BytesExternalizableImpl bytesExternalizable = new BytesExternalizableImpl(knownNodes, null);
BytesExternalizableImpl.ProposedNodes proposedNodes = new BytesExternalizableImpl.ProposedNodes(server1AddressAndPort, (byte) PROPOSED_IDENTIFIER);
// send the bootstrap along with this newly proposed identifier
bytesExternalizable.sendBootStrap(proposedNodes);
bytesExternalizable.writeMarshallable(udpData);
}
// SERVER 2 - is the node already in the cluster
// read the bootstrap along with the proposed identifier and host:port it came from
{
udpData.flip();
server2BytesExternalizable.readMarshallable(udpData);
Assert.assertTrue(server2BytesExternalizable.proposedIdentifiersWithHost.values().contains
(new BytesExternalizableImpl.ProposedNodes(server1AddressAndPort, (byte) PROPOSED_IDENTIFIER)));
}
// SERVER 2
// write/broadcast the the response to receiving a bootstrap message
{
//ByteBufferBytes bitSetBytes = new ByteBufferBytes(allocate(128 / 8));
// final DirectBitSet knownIdentifiers = new ATSDirectBitSet(bitSetBytes);
// RemoteNodes remoteNodes = new RemoteNodes(bitSetBytes);
udpData.clear();
// broadcasts in response to the bootstrap all the host:ports and identifiers that it knows about
// in this case given that there is only this node in the grid and one node joining,
// it will only be the node that has the bootstrap and this node, who's data will be sent
server2BytesExternalizable.writeMarshallable(udpData);
}
// SERVER 1
// read the bootstrap along with all the proposed identifiers ( this will include our
// PROPOSED_IDENTIFIER and any that come from other server
{
udpData.flip();
server1BytesExternalizable.readMarshallable(udpData);
Assert.assertTrue(server1BytesExternalizable.getRemoteNodes().addressAndPorts().contains
(server2AddressAndPort));
Assert.assertTrue(server1BytesExternalizable.getRemoteNodes().activeIdentifierBitSet().get
(SERVER2_IDENTIFER));
Assert.assertTrue(server1BytesExternalizable.proposedIdentifiersWithHost.values().contains
(new BytesExternalizableImpl.ProposedNodes(server1AddressAndPort, (byte) PROPOSED_IDENTIFIER)));
}
ChronicleMap<Integer, CharSequence> map = discoveryCluster.discoverMap(1025, 8087);
Thread.sleep(10000);
}
}