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

Formatting -- 100 columns max

Browse Source
This commit is contained in:
leventov
2015-03-14 20:28:44 +07:00
parent 1a8998d521
commit b0b0bb87c1
6 changed files with 83 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/BytesChronicleMap.java
+1 -1
View File
@@ -25,7 +25,7 @@ import java.util.Set;
public class BytesChronicleMap implements AbstractChronicleMap<Bytes, Bytes> {
final VanillaChronicleMap<?, ?, ?, ?, ?, ?> delegate;
final VanillaChronicleMap<?, ?, ?, ?, ?, ?> delegate;
TcpReplicator.TcpSocketChannelEntryWriter output;
public BytesChronicleMap(VanillaChronicleMap<?, ?, ?, ?, ?, ?> delegate) {
src/main/java/net/openhft/chronicle/map/ClientWiredChronicleMapStatelessBuilder.java
+11 -17
View File
@@ -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<K, V> implements
public final class ClientWiredChronicleMapStatelessBuilder<K, V> implements
ChronicleHashStatelessClientBuilder<ClientWiredChronicleMapStatelessBuilder<K, V>,
ChronicleMap<K, V>>,
MapBuilder<ClientWiredChronicleMapStatelessBuilder<K, V>> {
@@ -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.<Integer, Integer>.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<K, V> removeReturnsNull(boolean removeReturnsNull) {
public ClientWiredChronicleMapStatelessBuilder<K, V> 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<K, V>(this, keyClass, valueClass, channelID);
return new ClientWiredStatelessChronicleMap<K, V>(
this, keyClass, valueClass, channelID);
} else {
throw new IllegalStateException(
@@ -158,7 +155,4 @@ import java.util.concurrent.atomic.AtomicBoolean;
public ClientWiredStatelessTcpConnectionHub hub() {
return hub;
}
}
src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java
+25 -18
View File
@@ -40,9 +40,11 @@ import static net.openhft.chronicle.map.ClientWiredStatelessChronicleMap.EventId
/**
* @author Rob Austin.
*/
class ClientWiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Cloneable, ChannelFactory {
class ClientWiredStatelessChronicleMap<K, V>
implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, Clon
return readChunked(vClass, "VALUES", usingCollection);
}
private <E, A extends Collection<E>> Collection<E> readChunked(Class<E> vClass1, String values, A usingCollection) {
private <E, A extends Collection<E>> Collection<E> readChunked(
Class<E> vClass1, String values, A usingCollection) {
final long startTime = System.currentTimeMillis();
// send
@@ -484,7 +488,8 @@ class ClientWiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, Clon
@Nullable
private <R> R proxyReturnObject(@NotNull final String methodName, K key, V value, Class<V> resultType) {
private <R> R proxyReturnObject(
@NotNull final String methodName, K key, V value, Class<V> resultType) {
final long startTime = System.currentTimeMillis();
long transactionId;
@@ -764,8 +771,8 @@ class ClientWiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, 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
src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java
+11 -7
View File
@@ -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;
src/main/java/net/openhft/chronicle/map/MapWireHandler.java
+32 -24
View File
@@ -65,7 +65,7 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
@NotNull
private final StringBuilder methodName = new StringBuilder();
private final Supplier<ChronicleHashInstanceBuilder<ChronicleMap<K, V>>> chronicleHashInstanceBuilder;
private final Supplier<ChronicleHashInstanceBuilder<ChronicleMap<K, V>>> mapFactory;
private Wire inWire = null;
private Wire outWire = null;
private final Consumer writeElement = new Consumer<Iterator<byte[]>>() {
@@ -102,24 +102,24 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
private byte localIdentifier;
private long timestamp;
private short channelId;
private List<Replica> channelList;
private List<Replica> channels;
private ReplicationHub hub;
private byte remoteIdentifier;
public MapWireHandler(@NotNull final Supplier<ChronicleHashInstanceBuilder<ChronicleMap<K, V>>> chronicleHashInstanceBuilder,
@NotNull final ReplicationHub hub,
byte localIdentifier,
@NotNull final List<Replica> channelList) {
this(chronicleHashInstanceBuilder, hub);
this.channelList = channelList;
public MapWireHandler(
@NotNull Supplier<ChronicleHashInstanceBuilder<ChronicleMap<K, V>>> mapFactory,
@NotNull ReplicationHub hub, byte localIdentifier, @NotNull List<Replica> channels) {
this(mapFactory, hub);
this.channels = channels;
this.localIdentifier = localIdentifier;
}
public MapWireHandler(Supplier<ChronicleHashInstanceBuilder<ChronicleMap<K, V>>> chronicleHashInstanceBuilder, ReplicationHub hub) {
this.chronicleHashInstanceBuilder = chronicleHashInstanceBuilder;
public MapWireHandler(Supplier<ChronicleHashInstanceBuilder<ChronicleMap<K, V>>> mapFactory,
ReplicationHub hub) {
this.mapFactory = mapFactory;
this.hub = hub;
}
@@ -136,8 +136,9 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
}
private void writeChunked(long transactionId, @NotNull final Function<BytesChronicleMap, Iterator> function,
@NotNull final Consumer<Iterator> c) throws StreamCorruptedException {
private void writeChunked(
long transactionId, @NotNull final Function<BytesChronicleMap, Iterator> function,
@NotNull final Consumer<Iterator> c) throws StreamCorruptedException {
final BytesChronicleMap m = bytesMap(channelId);
final Iterator<byte[]> iterator = function.apply(m);
@@ -258,7 +259,7 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
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<K, V> implements WireHandler, Consumer<WireHandlers> {
}
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<K, V> implements WireHandler, Consumer<WireHandlers> {
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<K, V> implements WireHandler, Consumer<WireHandlers> {
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<K, V> implements WireHandler, Consumer<WireHandlers> {
// 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<K, V> implements WireHandler, Consumer<WireHandlers> {
}
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<K, V> implements WireHandler, Consumer<WireHandlers> {
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<K, V> implements WireHandler, Consumer<WireHandlers> {
*/
@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")
src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java
+3 -5
View File
@@ -34,11 +34,9 @@ import java.util.function.Supplier;
public class MapWireHandlerBuilder {
public static <K, V> WireHandler of(
@NotNull final Supplier<ChronicleHashInstanceBuilder<ChronicleMap<K, V>>> chronicleHashInstanceBuilder,
@NotNull final ReplicationHub hub,
byte localIdentifier,
@NotNull final List<Replica> channelList) {
return new MapWireHandler<K, V>(chronicleHashInstanceBuilder, hub, localIdentifier, channelList);
@NotNull Supplier<ChronicleHashInstanceBuilder<ChronicleMap<K, V>>> mapFactory,
@NotNull ReplicationHub hub, byte localIdentifier, @NotNull List<Replica> channels) {
return new MapWireHandler<>(mapFactory, hub, localIdentifier, channels);
}
public static enum Fields implements WireKey {