diff --git a/src/main/java/net/openhft/chronicle/map/BytesChronicleMap.java b/src/main/java/net/openhft/chronicle/map/BytesChronicleMap.java index d65fbca30..10696ce66 100644 --- a/src/main/java/net/openhft/chronicle/map/BytesChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/BytesChronicleMap.java @@ -25,7 +25,7 @@ import java.util.Set; public class BytesChronicleMap implements AbstractChronicleMap { - final VanillaChronicleMap delegate; + final VanillaChronicleMap delegate; TcpReplicator.TcpSocketChannelEntryWriter output; public BytesChronicleMap(VanillaChronicleMap delegate) { diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredChronicleMapStatelessBuilder.java b/src/main/java/net/openhft/chronicle/map/ClientWiredChronicleMapStatelessBuilder.java index f8c3fa5c3..e1bd407b4 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredChronicleMapStatelessBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredChronicleMapStatelessBuilder.java @@ -22,14 +22,14 @@ import net.openhft.chronicle.hash.ChronicleHashStatelessClientBuilder; import net.openhft.lang.MemoryUnit; import java.io.IOException; -import java.lang.Class;import java.lang.IllegalStateException;import java.lang.Override;import java.lang.String;import java.net.InetSocketAddress; +import java.net.InetSocketAddress; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; /** * @author Rob Austin. */ - public final class ClientWiredChronicleMapStatelessBuilder implements +public final class ClientWiredChronicleMapStatelessBuilder implements ChronicleHashStatelessClientBuilder, ChronicleMap>, MapBuilder> { @@ -41,28 +41,23 @@ import java.util.concurrent.atomic.AtomicBoolean; private short channelID; private boolean doHandShaking; - public ClientWiredChronicleMapStatelessBuilder(InetSocketAddress remoteAddress, Class keyClass, Class valueClass, short channelID) { + public ClientWiredChronicleMapStatelessBuilder( + InetSocketAddress remoteAddress, Class keyClass, Class valueClass, short channelID) { this.keyClass = keyClass; this.valueClass = valueClass; this.remoteAddress = remoteAddress; this.channelID = channelID; } - public ClientWiredChronicleMapStatelessBuilder(ClientWiredStatelessTcpConnectionHub hub, Class keyClass, Class valueClass, short channelID) { + public ClientWiredChronicleMapStatelessBuilder( + ClientWiredStatelessTcpConnectionHub hub, Class keyClass, Class valueClass, + short channelID) { this.keyClass = keyClass; this.valueClass = valueClass; this.hub = hub; this.channelID = channelID; } - // This method is meaningful, despite you could call of().create(), because Java 7 doesn't - // infer type parameters if case of chained calls, like this; and then you couldn't omit class - // name. Then it is - // ChronicleMapStatelessClientBuilder..of(addr).create() - // vs - // createClientOf(addr) // static import - // -- the second is really shorter. - private InetSocketAddress remoteAddress; private boolean putReturnsNull = false; @@ -89,7 +84,8 @@ import java.util.concurrent.atomic.AtomicBoolean; } @Override - public ClientWiredChronicleMapStatelessBuilder removeReturnsNull(boolean removeReturnsNull) { + public ClientWiredChronicleMapStatelessBuilder removeReturnsNull( + boolean removeReturnsNull) { this.removeReturnsNull = removeReturnsNull; return this; } @@ -136,7 +132,8 @@ import java.util.concurrent.atomic.AtomicBoolean; hub = new ClientWiredStatelessTcpConnectionHub(this, localIdentifier, doHandShaking); if (!used.getAndSet(true)) { - return new ClientWiredStatelessChronicleMap(this, keyClass, valueClass, channelID); + return new ClientWiredStatelessChronicleMap( + this, keyClass, valueClass, channelID); } else { throw new IllegalStateException( @@ -158,7 +155,4 @@ import java.util.concurrent.atomic.AtomicBoolean; public ClientWiredStatelessTcpConnectionHub hub() { return hub; } - - - } diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java index 45876f3e8..4309156a6 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java @@ -40,9 +40,11 @@ import static net.openhft.chronicle.map.ClientWiredStatelessChronicleMap.EventId /** * @author Rob Austin. */ -class ClientWiredStatelessChronicleMap implements ChronicleMap, Cloneable, ChannelFactory { +class ClientWiredStatelessChronicleMap + implements ChronicleMap, Cloneable, ChannelFactory { - private static final Logger LOG = LoggerFactory.getLogger(ClientWiredStatelessChronicleMap.class); + private static final Logger LOG = + LoggerFactory.getLogger(ClientWiredStatelessChronicleMap.class); private final ClientWiredStatelessTcpConnectionHub hub; @@ -53,10 +55,9 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon private boolean removeReturnsNull; private short channelID; - public ClientWiredStatelessChronicleMap(@NotNull final ClientWiredChronicleMapStatelessBuilder config, - @NotNull final Class kClass, - @NotNull final Class vClass, - short channelID) { + public ClientWiredStatelessChronicleMap( + @NotNull final ClientWiredChronicleMapStatelessBuilder config, + @NotNull final Class kClass, @NotNull final Class vClass, short channelID) { this.channelID = channelID; hub = config.hub; this.putReturnsNull = config.putReturnsNull(); @@ -140,19 +141,19 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon @SuppressWarnings("NullableProblems") public boolean remove(Object key, Object value) { - if (key == null) throw new NullPointerException(); - return value != null && proxyReturnBoolean(REMOVE_WITH_VALUE.toString(), (K) key, (V) value); - + return value != null && + proxyReturnBoolean(REMOVE_WITH_VALUE.toString(), (K) key, (V) value); } @SuppressWarnings("NullableProblems") public boolean replace(K key, V oldValue, V newValue) { if (key == null || oldValue == null || newValue == null) throw new NullPointerException(); - return proxyReturnBoolean(REPLACE_WITH_OLD_AND_NEW_VALUE.toString(), key, oldValue, newValue); + return proxyReturnBoolean( + REPLACE_WITH_OLD_AND_NEW_VALUE.toString(), key, oldValue, newValue); } @SuppressWarnings("NullableProblems") @@ -257,7 +258,8 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon public V remove(Object key) { if (key == null) throw keyNotNullNPE(); - return proxyReturnObject(vClass, removeReturnsNull ? REMOVE_WITHOUT_ACC.toString() : REMOVE.toString(), (K) key); + return proxyReturnObject(vClass, + removeReturnsNull ? REMOVE_WITHOUT_ACC.toString() : REMOVE.toString(), (K) key); } @Override @@ -326,7 +328,8 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon public V put(K key, V value) { if (key == null || value == null) throw new NullPointerException(); - return proxyReturnObject(putReturnsNull ? PUT_WITHOUT_ACC.toString() : PUT.toString(), key, value, vClass); + return proxyReturnObject(putReturnsNull ? PUT_WITHOUT_ACC.toString() : PUT.toString(), + key, value, vClass); } @Nullable @@ -359,7 +362,8 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon return readChunked(vClass, "VALUES", usingCollection); } - private > Collection readChunked(Class vClass1, String values, A usingCollection) { + private > Collection readChunked( + Class vClass1, String values, A usingCollection) { final long startTime = System.currentTimeMillis(); // send @@ -484,7 +488,8 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon } else if (value instanceof Marshallable) { wire.write(fieldName).marshallable((Marshallable) value); } else { - throw new IllegalStateException("type=" + value.getClass() + " is unsupported, it must either be of type Marshallable or CharSequence"); + throw new IllegalStateException("type=" + value.getClass() + + " is unsupported, it must either be of type Marshallable or CharSequence"); } } @@ -628,7 +633,8 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon } @SuppressWarnings("SameParameterValue") - private boolean proxyReturnBoolean(@NotNull final String methodName, K key, V value1, V value2) { + private boolean proxyReturnBoolean( + @NotNull final String methodName, K key, V value1, V value2) { final long startTime = System.currentTimeMillis(); long transactionId; @@ -737,7 +743,8 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon @Nullable - private R proxyReturnObject(@NotNull final String methodName, K key, V value, Class resultType) { + private R proxyReturnObject( + @NotNull final String methodName, K key, V value, Class resultType) { final long startTime = System.currentTimeMillis(); long transactionId; @@ -764,8 +771,8 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon return (R) readValue(Fields.RESULT, transactionId, startTime, null); else - throw new UnsupportedOperationException("class of type class=" + resultType + " is not " + - "supported"); + throw new UnsupportedOperationException("class of type class=" + resultType + + " is not supported"); } @Nullable diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java index ca042457e..e8d849204 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java @@ -84,7 +84,9 @@ public class ClientWiredStatelessTcpConnectionHub { private boolean doHandShaking; - public ClientWiredStatelessTcpConnectionHub(ClientWiredChronicleMapStatelessBuilder config, byte localIdentifier, boolean doHandShaking) { + public ClientWiredStatelessTcpConnectionHub( + ClientWiredChronicleMapStatelessBuilder config, byte localIdentifier, + boolean doHandShaking) { this.localIdentifier = localIdentifier; this.doHandShaking = doHandShaking; this.remoteAddress = config.remoteAddress(); @@ -132,7 +134,8 @@ public class ClientWiredStatelessTcpConnectionHub { LOG.warn("DIFFERENT CHRONICLE-MAP VERSIONS: The Chronicle-Map-Server and " + "Stateless-Client are on different " + "versions, " + - " we suggest that you use the same version, server=" + serverApplicationVersion(channelID) + ", " + + " we suggest that you use the same version, server=" + + serverApplicationVersion(channelID) + ", " + "client=" + clientVersion); } } @@ -268,7 +271,8 @@ public class ClientWiredStatelessTcpConnectionHub { writeSocket(wire, timeoutTime); break; - } catch (@NotNull java.nio.channels.ClosedChannelException | ClosedConnectionException e) { + } catch (@NotNull java.nio.channels.ClosedChannelException | + ClosedConnectionException e) { checkTimeout(timeoutTime); lazyConnect(timeoutMs, remoteAddress); } @@ -347,8 +351,8 @@ public class ClientWiredStatelessTcpConnectionHub { // reads just the size readSocket(SIZE_OF_SIZE, timeoutTime); - final int messageSize = intWire.bytes().readUnsignedShort(intWire.bytes().position()); - + final int messageSize = + intWire.bytes().readUnsignedShort(intWire.bytes().position()); assert messageSize > 0 : "Invalid message size " + messageSize; assert messageSize < 16 << 20 : "Invalid message size " + messageSize; @@ -487,8 +491,8 @@ public class ClientWiredStatelessTcpConnectionHub { long outBytesPosition = outWire.bytes().position(); - // if we have other threads waiting to send and the buffer is not full, let the other threads - // write to the buffer + // if we have other threads waiting to send and the buffer is not full, + // let the other threads write to the buffer if (outBytesLock().hasQueuedThreads() && outBytesPosition + largestChunkSoFar <= tcpBufferSize) { return; diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java index 4fc423af2..09cbd15e2 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java @@ -65,7 +65,7 @@ class MapWireHandler implements WireHandler, Consumer { @NotNull private final StringBuilder methodName = new StringBuilder(); - private final Supplier>> chronicleHashInstanceBuilder; + private final Supplier>> mapFactory; private Wire inWire = null; private Wire outWire = null; private final Consumer writeElement = new Consumer>() { @@ -102,24 +102,24 @@ class MapWireHandler implements WireHandler, Consumer { private byte localIdentifier; private long timestamp; private short channelId; - private List channelList; + private List channels; private ReplicationHub hub; private byte remoteIdentifier; - public MapWireHandler(@NotNull final Supplier>> chronicleHashInstanceBuilder, - @NotNull final ReplicationHub hub, - byte localIdentifier, - @NotNull final List channelList) { - this(chronicleHashInstanceBuilder, hub); - this.channelList = channelList; + public MapWireHandler( + @NotNull Supplier>> mapFactory, + @NotNull ReplicationHub hub, byte localIdentifier, @NotNull List channels) { + this(mapFactory, hub); + this.channels = channels; this.localIdentifier = localIdentifier; } - public MapWireHandler(Supplier>> chronicleHashInstanceBuilder, ReplicationHub hub) { - this.chronicleHashInstanceBuilder = chronicleHashInstanceBuilder; + public MapWireHandler(Supplier>> mapFactory, + ReplicationHub hub) { + this.mapFactory = mapFactory; this.hub = hub; } @@ -136,8 +136,9 @@ class MapWireHandler implements WireHandler, Consumer { } - private void writeChunked(long transactionId, @NotNull final Function function, - @NotNull final Consumer c) throws StreamCorruptedException { + private void writeChunked( + long transactionId, @NotNull final Function function, + @NotNull final Consumer c) throws StreamCorruptedException { final BytesChronicleMap m = bytesMap(channelId); final Iterator iterator = function.apply(m); @@ -258,7 +259,7 @@ class MapWireHandler implements WireHandler, Consumer { if ("CREATE_CHANNEL".contentEquals(methodName)) { writeVoid(() -> { short channelId1 = inWire.read(ARG_1).int16(); - chronicleHashInstanceBuilder.get().replicatedViaChannel(hub.createChannel(channelId1)).create(); + mapFactory.get().replicatedViaChannel(hub.createChannel(channelId1)).create(); return null; }); return; @@ -281,12 +282,14 @@ class MapWireHandler implements WireHandler, Consumer { } if ("CONTAINS_KEY".contentEquals(methodName)) { - write(b -> outWire.write(RESULT).bool(b.delegate.containsKey(toByteArray(inWire, ARG_1)))); + write(b -> outWire.write(RESULT).bool( + b.delegate.containsKey(toByteArray(inWire, ARG_1)))); return; } if ("CONTAINS_VALUE".contentEquals(methodName)) { - write(b -> outWire.write(RESULT).bool(b.delegate.containsValue(toByteArray(inWire, ARG_1)))); + write(b -> outWire.write(RESULT).bool( + b.delegate.containsValue(toByteArray(inWire, ARG_1)))); return; } @@ -297,7 +300,8 @@ class MapWireHandler implements WireHandler, Consumer { if ("PUT".contentEquals(methodName)) { writeValue(b -> { - final net.openhft.lang.io.Bytes reader = MapWireHandler.this.toReader(inWire, ARG_1, ARG_2); + final net.openhft.lang.io.Bytes reader = + MapWireHandler.this.toReader(inWire, ARG_1, ARG_2); // todo call return b.put(reader, reader, timestamp, identifier()); return b.put(reader, reader); @@ -364,7 +368,8 @@ class MapWireHandler implements WireHandler, Consumer { if ("REMOVE_WITH_VALUE".contentEquals(methodName)) { write(bytesMap -> { final net.openhft.lang.io.Bytes reader = toReader(inWire, ARG_1, ARG_2); - // todo call outWire.write(RESULT).bool(bytesMap.remove(reader, reader, timestamp, identifier())); + // todo call outWire.write(RESULT) + // .bool(bytesMap.remove(reader, reader, timestamp, identifier())); outWire.write(RESULT).bool(bytesMap.remove(reader, reader)); }); return; @@ -394,7 +399,8 @@ class MapWireHandler implements WireHandler, Consumer { // if (len > 4) if (EventGroup.IS_DEBUG) { long len = outWire.bytes().position() - SIZE_OF_SIZE; - System.out.println("--------------------------------------------\nserver wrote:\n\n" + Bytes.toDebugString(outWire.bytes(), SIZE_OF_SIZE, len)); + System.out.println("--------------------------------------------\nserver wrote:\n\n" + + Bytes.toDebugString(outWire.bytes(), SIZE_OF_SIZE, len)); } } @@ -486,7 +492,8 @@ class MapWireHandler implements WireHandler, Consumer { } private byte identifier() { - // if the client provides a remote identifier then we will use that otherwise we will use the local identifier + // if the client provides a remote identifier then we will use that otherwise we will use + // the local identifier return remoteIdentifier == 0 ? localIdentifier : remoteIdentifier; } @@ -597,11 +604,11 @@ class MapWireHandler implements WireHandler, Consumer { private ReplicatedChronicleMap map(short channelId) { // todo this cast is a bit of a hack, improve later - final ReplicatedChronicleMap replicas = - (ReplicatedChronicleMap) channelList.get(channelId); + final ReplicatedChronicleMap map = + (ReplicatedChronicleMap) channels.get(channelId); - if (replicas != null) - return replicas; + if (map != null) + return map; throw new IllegalStateException(); } @@ -736,7 +743,8 @@ class MapWireHandler implements WireHandler, Consumer { */ @SuppressWarnings("UnusedDeclaration") private void showOutWire() { - System.out.println("pos=" + outWire.bytes().position() + ",bytes=" + Bytes.toDebugString(outWire.bytes(), 0, outWire.bytes().position())); + System.out.println("pos=" + outWire.bytes().position() + ",bytes=" + + Bytes.toDebugString(outWire.bytes(), 0, outWire.bytes().position())); } @SuppressWarnings("SameReturnValue") diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java b/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java index f67efcda8..f4de30927 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java @@ -34,11 +34,9 @@ import java.util.function.Supplier; public class MapWireHandlerBuilder { public static WireHandler of( - @NotNull final Supplier>> chronicleHashInstanceBuilder, - @NotNull final ReplicationHub hub, - byte localIdentifier, - @NotNull final List channelList) { - return new MapWireHandler(chronicleHashInstanceBuilder, hub, localIdentifier, channelList); + @NotNull Supplier>> mapFactory, + @NotNull ReplicationHub hub, byte localIdentifier, @NotNull List channels) { + return new MapWireHandler<>(mapFactory, hub, localIdentifier, channels); } public static enum Fields implements WireKey {