HCOLL-134 better handling of toString() and close() of the TCP connections

This commit is contained in:
Rob Austin
2014-11-20 18:48:58 +00:00
parent 272eaecaf8
commit a8479ee02f
6 changed files with 70 additions and 37 deletions
@@ -24,11 +24,17 @@ import java.io.Serializable;
/**
* Common configurations of {@link SingleChronicleHashReplication} and {@link ReplicationHub}.
*/
public abstract class AbstractReplication implements Serializable{
public abstract class AbstractReplication implements Serializable {
private final byte localIdentifier;
private final @Nullable TcpTransportAndNetworkConfig tcpConfig;
private final @Nullable UdpTransportConfig udpConfig;
private final @Nullable RemoteNodeValidator remoteNodeValidator;
private final
@Nullable
TcpTransportAndNetworkConfig tcpConfig;
private final
@Nullable
UdpTransportConfig udpConfig;
private final
@Nullable
RemoteNodeValidator remoteNodeValidator;
// package-private to forbid subclassing from outside of the package
AbstractReplication(byte localIdentifier, Builder builder) {
@@ -38,6 +44,16 @@ public abstract class AbstractReplication implements Serializable{
remoteNodeValidator = builder.remoteNodeValidator;
}
@Override
public String toString() {
return ", localIdentifier=" + localIdentifier +
", tcpConfig=" + tcpConfig +
", udpConfig=" + udpConfig +
", remoteNodeValidator=" + remoteNodeValidator;
}
public byte identifier() {
return localIdentifier;
}
@@ -58,13 +74,13 @@ 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;
@@ -72,7 +88,8 @@ public abstract class AbstractReplication implements Serializable{
private RemoteNodeValidator remoteNodeValidator = null;
// package-private to forbid subclassing from outside of the package
Builder() {}
Builder() {
}
@NotNull
public B tcpTransportAndNetwork(@Nullable TcpTransportAndNetworkConfig tcpConfig) {
@@ -81,8 +98,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
@@ -106,10 +123,10 @@ public abstract class AbstractReplication implements Serializable{
* @param identifier the node (server) identifier of the returned replication
* @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.
* @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.
*/
@NotNull
public abstract R createWithId(byte identifier);
@@ -120,12 +137,11 @@ public abstract class AbstractReplication implements Serializable{
" given");
}
@Override
public String toString() {
return
"tcpConfig=" + tcpConfig +
", udpConfig=" + udpConfig +
", remoteNodeValidator=" + remoteNodeValidator ;
return ", udpConfig=" + udpConfig +
", remoteNodeValidator=" + remoteNodeValidator;
}
}
@@ -72,11 +72,10 @@ public final class ReplicationHub extends AbstractReplication {
@Override
public String toString() {
return "ReplicationHub{" +
return "ReplicationHub{" + super.toString() +
"maxEntrySize=" + maxEntrySize +
", channels=" + Arrays.toString(channels) +
super.toString() +
'}';
", channels=" + Arrays.toString(channels) + '}';
}
@@ -23,9 +23,13 @@ import java.io.IOException;
/**
* @author Rob Austin.
*/
class IORuntimeException extends RuntimeException{
class IORuntimeException extends RuntimeException {
public IORuntimeException(IOException e) {
super(e);
}
public IORuntimeException(String message) {
super(message);
}
}
@@ -996,7 +996,11 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
private Bytes receive(int requiredNumberOfBytes, long timeoutTime) throws IOException {
while (bytes.buffer().position() < requiredNumberOfBytes) {
clientChannel.read(bytes.buffer());
int len = clientChannel.read(bytes.buffer());
if (len == -1)
throw new IORuntimeException("Disconnected to remote server");
checkTimeout(timeoutTime);
}
@@ -589,11 +589,22 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
final SocketChannel socketChannel = (SocketChannel) key.channel();
final Attached attached = (Attached) key.attachment();
if (attached.entryWriter.isWorkIncomplete())
return;
try {
if (attached.entryReader.readSocketToBuffer(socketChannel) <= 0)
int len = attached.entryReader.readSocketToBuffer(socketChannel);
if (len == -1) {
socketChannel.shutdownInput();
socketChannel.shutdownOutput();
//todo we should call close, but if we do it breaks the bootstrap test case, further
// investigation required
return;
}
if (len == 0)
return;
if (attached.entryWriter.isWorkIncomplete())
return;
} catch (IOException e) {
@@ -1108,9 +1119,8 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
// its set to MIN_VALUE when it should be read again
if (state == NOT_SET) {
if (out.remaining() < SIZE_OF_SIZE + 1) {
if (out.remaining() < SIZE_OF_SIZE + 1)
return;
}
// state is used for both heartbeat and stateless
state = out.readByte();