Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

HCOLL-134 added a serialization test

Browse Source
This commit is contained in:
Rob Austin
2015-01-12 16:25:45 +00:00
parent aa853b8929
commit d76691fcaa
5 changed files with 121 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java
+15 -14
View File
@@ -34,7 +34,7 @@ public abstract class AbstractReplication implements Serializable {
UdpTransportConfig udpConfig;
private final
@Nullable
RemoteNodeValidator remoteNodeValidator;
transient RemoteNodeValidator remoteNodeValidator;
private final boolean bootstrapOnlyLocalEntries;
// package-private to forbid subclassing from outside of the package
@@ -79,13 +79,14 @@ public abstract class AbstractReplication implements Serializable {
}
/**
* Builder of {@link AbstractReplication} configurations. This class and it's subclasses are mutable,
* configuration methods mutate the builder and return it back for convenient chaining.
* Builder of {@link AbstractReplication} configurations. This class and it's subclasses are
* mutable, configuration methods mutate the builder and return it back for convenient
* chaining.
*
* @param <R> the concrete {@link AbstractReplication} subclass: {@link SingleChronicleHashReplication} or
* {@link ReplicationHub}
* @param <B> the concrete builder subclass: {@link SingleChronicleHashReplication.Builder} or {@link
* ReplicationHub.Builder}
* @param <R> the concrete {@link AbstractReplication} subclass: {@link
* SingleChronicleHashReplication} or {@link ReplicationHub}
* @param <B> the concrete builder subclass: {@link SingleChronicleHashReplication.Builder} or
* {@link ReplicationHub.Builder}
*/
public static abstract class Builder<R extends AbstractReplication, B extends Builder<R, B>> {
private TcpTransportAndNetworkConfig tcpConfig = null;
@@ -104,8 +105,8 @@ public abstract class AbstractReplication implements Serializable {
}
/**
* Configures UDP transport settings, used by Replications, created by this builder. {@code null}
* means that UDP transport shouldn't being used.
* Configures UDP transport settings, used by Replications, created by this builder. {@code
* null} means that UDP transport shouldn't being used.
*
* @param udpConfig the new UDP transport config for replications, created by this builder.
* @return this builder back, for chaining
@@ -135,8 +136,8 @@ public abstract class AbstractReplication implements Serializable {
* currently running.
*
* @param bootstrapOnlyLocalEntries if a node provided with this replication should
* replicate only local data to the new nodes joining
* the replication network
* replicate only local data to the new nodes joining the
* replication network
* @return this builder back
*/
@NotNull
@@ -153,9 +154,9 @@ public abstract class AbstractReplication implements Serializable {
* @return a new Replication instance with the specified node (server) identifier
* @throws IllegalArgumentException if the given identifier is non-positive
* @throws IllegalStateException if neither {@link #tcpTransportAndNetwork(TcpTransportAndNetworkConfig)}
* nor {@link #udpTransport(UdpTransportConfig)} are configured to
* non-{@code null}. At least one of the transport-level configs
* should be specified.
* nor {@link #udpTransport(UdpTransportConfig)} are
* configured to non-{@code null}. At least one of the
* transport-level configs should be specified.
*/
@NotNull
public abstract R createWithId(byte identifier);
src/main/java/net/openhft/chronicle/hash/replication/UdpTransportConfig.java
+4
View File
@@ -37,6 +37,7 @@ public final class UdpTransportConfig {
private final ThrottlingConfig throttlingConfig;
private final int udpBufferSize;
private String name;
private UdpTransportConfig(
InetAddress address,
@@ -156,4 +157,7 @@ public final class UdpTransportConfig {
return create(address(), port(), networkInterface(), throttlingConfig(), udpBufferSize);
}
public String name() {
return this.name;
}
}
src/main/java/net/openhft/chronicle/map/MapInstanceBuilder.java
+4 -4
View File
@@ -27,12 +27,12 @@ import java.io.Serializable;
import java.util.concurrent.atomic.AtomicBoolean;
final class MapInstanceBuilder<K, V>
implements ChronicleHashInstanceBuilder<ChronicleMap<K, V>>, Serializable{
implements ChronicleHashInstanceBuilder<ChronicleMap<K, V>>, Serializable {
final ChronicleMapBuilder<K, V> mapBuilder;
SingleChronicleHashReplication singleHashReplication;
ReplicationChannel channel;
File file;
transient SingleChronicleHashReplication singleHashReplication;
transient ReplicationChannel channel;
transient File file;
String name;
final AtomicBoolean used;
src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java
+77 -33
View File
@@ -25,10 +25,8 @@ import org.slf4j.LoggerFactory;
import java.io.Closeable;
import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.NetworkInterface;
import java.net.StandardSocketOptions;
import java.net.*;
import java.nio.BufferUnderflowException;
import java.nio.channels.*;
import java.util.Set;
@@ -61,6 +59,7 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica.
private SelectableChannel writeChannel;
private volatile boolean shouldEnableOpWrite;
private final String name;
/**
* @param replicationConfig
@@ -79,6 +78,7 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica.
address = replicationConfig.address();
port = replicationConfig.port();
networkInterface = replicationConfig.networkInterface();
name = replicationConfig.name();
serverConnector = new ServerConnector();
}
@@ -99,13 +99,15 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica.
/**
* binds to the server socket and process data This method will block until interrupted
*/
@Override
void processEvent() throws IOException {
connectClient().register(selector, OP_READ);
serverConnector.connectLater();
try {
connectClient().register(selector, OP_READ);
serverConnector.connectLater();
while (selector.isOpen()) {
registerPendingRegistrations();
@@ -122,46 +124,88 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica.
continue; // nothing to do
}
final Set<SelectionKey> selectionKeys = selector.selectedKeys();
for (final SelectionKey key : selectionKeys) {
if (useJavaNIOSelectionKeys) {
// use the standard java nio selector
final Set<SelectionKey> selectionKeys = selector.selectedKeys();
for (final SelectionKey key : selectionKeys) {
processKey(key);
}
selectionKeys.clear();
} else {
// use the netty like selector
final SelectionKey[] keys = selectedKeys.flip();
try {
for (int i = 0; i < keys.length && keys[i] != null; i++) {
final SelectionKey key = keys[i];
if (key.isReadable()) {
final DatagramChannel socketChannel = (DatagramChannel) key.channel();
reader.readAll(socketChannel);
}
if (key.isWritable()) {
final DatagramChannel socketChannel = (DatagramChannel) key.channel();
try {
int bytesJustWritten = writer.writeAll(socketChannel);
contemplateThrottleWrites(bytesJustWritten);
} catch (NotYetConnectedException e) {
if (LOG.isDebugEnabled())
LOG.debug("", e);
serverConnector.connectLater();
} catch (IOException e) {
if (LOG.isDebugEnabled())
LOG.debug("", e);
serverConnector.connectLater();
processKey(key);
} catch (BufferUnderflowException e) {
if (!isClosed)
LOG.error("", e);
}
}
} catch (Exception e) {
LOG.error("", e);
if (!isClosed)
closeEarlyAndQuietly(key.channel());
} finally {
for (int i = 0; i < keys.length && keys[i] != null; i++) {
keys[i] = null;
}
}
}
selectionKeys.clear();
}
} catch (CancelledKeyException | ConnectException | ClosedChannelException |
ClosedSelectorException e) {
if (LOG.isDebugEnabled())
LOG.debug("", e);
} catch (Exception e) {
LOG.error("", e);
} catch (Throwable e) {
LOG.error("", e);
throw e;
} finally {
if (LOG.isDebugEnabled())
LOG.debug("closing name=" + this.name);
if (!isClosed) {
closeResources();
}
}
}
private void processKey(SelectionKey key) {
try {
if (key.isReadable()) {
final DatagramChannel socketChannel = (DatagramChannel) key.channel();
reader.readAll(socketChannel);
}
if (key.isWritable()) {
final DatagramChannel socketChannel = (DatagramChannel) key.channel();
try {
int bytesJustWritten = writer.writeAll(socketChannel);
contemplateThrottleWrites(bytesJustWritten);
} catch (NotYetConnectedException e) {
if (LOG.isDebugEnabled())
LOG.debug("", e);
serverConnector.connectLater();
} catch (IOException e) {
if (LOG.isDebugEnabled())
LOG.debug("", e);
serverConnector.connectLater();
}
}
} catch (Exception e) {
LOG.error("", e);
if (!isClosed)
closeEarlyAndQuietly(key.channel());
}
}
private DatagramChannel connectClient() throws IOException {
final DatagramChannel client = address.isMulticastAddress() ?
DatagramChannel.open(address.getAddress().length == 4 ? INET : INET6) :
src/test/java/net/openhft/chronicle/map/MapByNameTest.java
+21
View File
@@ -87,6 +87,27 @@ public class MapByNameTest {
ChronicleMap<CharSequence, CharSequence> myMap2 = mapByName.from("myMap");
}
@Test
public void testReplicationHubSerialization() throws IOException, InterruptedException,
TimeoutException {
NodeDiscovery nodeDiscovery = new NodeDiscovery();
final ReplicationHubFindByName<CharSequence> mapByName = nodeDiscovery.mapByName();
Thread.sleep(2000);
MapInstanceBuilder<CharSequence, CharSequence> config =
(MapInstanceBuilder<CharSequence, CharSequence>)
ChronicleMapBuilder.of(CharSequence.class, CharSequence.class)
.minSegments(2)
.entries(2)
.removeReturnsNull(true)
.instance()
.name("myMap6");
mapByName.create(config);
}
public static void main(String... args) throws IOException, InterruptedException, TimeoutException {
NodeDiscovery nodeDiscovery = new NodeDiscovery();