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

CE-20 refactors

Browse Source
This commit is contained in:
Rob Austin
2015-04-07 11:35:28 +01:00
parent 417fcdf30e
commit f550e07fdc
3 changed files with 75 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java
+67 -104
View File
@@ -47,6 +47,16 @@ class ClientWiredStatelessChronicleMap<K, V>
private static final Logger LOG =
LoggerFactory.getLogger(ClientWiredStatelessChronicleMap.class);
private static final WriteMarshallable EMPTY = wire -> {
};
public static final Consumer<WireOut> VOID_METHOD = new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.writeValue().marshallable(EMPTY);
}
};
private final ClientWiredStatelessTcpConnectionHub hub;
protected Class<K> kClass;
@@ -198,7 +208,7 @@ class ClientWiredStatelessChronicleMap<K, V>
@Override
public int hashCode() {
return proxyReturnInt(hashCode.toString());
return proxyReturnInt(hashCode);
}
@NotNull
@@ -218,7 +228,7 @@ class ClientWiredStatelessChronicleMap<K, V>
}
public boolean isEmpty() {
return proxyReturnBoolean(isEmpty.toString());
return proxyReturnBoolean(isEmpty);
}
public boolean containsKey(Object key) {
@@ -235,7 +245,7 @@ class ClientWiredStatelessChronicleMap<K, V>
}
public long longSize() {
return proxyReturnLong(longSize.toString());
return proxyReturnLong(longSize);
}
@Override
@@ -272,15 +282,8 @@ class ClientWiredStatelessChronicleMap<K, V>
@Override
public void createChannel(short channelID) {
proxyReturnVoid(
() -> {
proxyReturnVoid(createChannel, wireOut -> wireOut.writeValue().int16(channelID));
hub.outWire().writeDocument(false, wireOut -> {
hub.outWire().writeEventName(createChannel);
hub.outWire().writeValue().int16(channelID);
});
});
}
public void putAll(@NotNull Map<? extends K, ? extends V> map) {
@@ -306,7 +309,7 @@ class ClientWiredStatelessChronicleMap<K, V>
hub.startTime(startTime);
wire.write(Fields.csp).text("MAP");
wire.write(Fields.tid).int64(tid);
wire.write(Fields.timeStamp).int64(startTime);
wire.write(Fields.channelId).int16(channelID);
hub.outWire().writeEventName(putAll);
@@ -361,8 +364,7 @@ class ClientWiredStatelessChronicleMap<K, V>
}
public void clear() {
proxyReturnVoid(() -> hub.outWire().writeDocument(false, wireOut -> wireOut.writeEventName(clear)
));
proxyReturnVoid(clear);
}
@@ -705,41 +707,18 @@ class ClientWiredStatelessChronicleMap<K, V>
hub.inBytesLock().unlock();
}
}
/*
@SuppressWarnings("SameParameterValue")
private boolean proxyReturnBoolean(@NotNull final EventId methodName, Object... parameters) {
final long startTime = System.currentTimeMillis();
long tid;
hub.outBytesLock().lock();
try {
tid = hub.writeHeader(startTime, channelID, hub.outWire());
sendEvent(methodName, parameters);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
}
return readBoolean(tid, startTime);
}*/
@SuppressWarnings("SameParameterValue")
private boolean proxyReturnBoolean(
@NotNull final EventId methodName, Object... args) {
@NotNull final EventId eventId, Object... args) {
final long startTime = System.currentTimeMillis();
long tid;
hub.outBytesLock().lock();
try {
tid = hub.writeHeader(startTime, channelID, hub.outWire());
sendEvent(methodName, args);
sendEvent(eventId, args);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
@@ -791,92 +770,70 @@ class ClientWiredStatelessChronicleMap<K, V>
@SuppressWarnings("SameParameterValue")
private boolean proxyReturnBooleanK(@NotNull final EventId methodName, K key) {
final long startTime = System.currentTimeMillis();
return readBoolean(proxySend(methodName, startTime, key), startTime);
}
@SuppressWarnings("SameParameterValue")
private long proxyReturnLong(@NotNull final String methodName) {
private long proxyReturnLong(@NotNull final EventId eventId) {
final long startTime = System.currentTimeMillis();
long tid;
hub.outBytesLock().lock();
try {
tid = hub.writeHeader(startTime, channelID, hub.outWire());
hub.outWire().writeDocument(false, wireOut -> wireOut.writeEventName(() -> methodName)
);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
}
long tid = sendEvent(startTime, eventId, null);
return readLong(tid, startTime);
}
@SuppressWarnings("SameParameterValue")
private boolean proxyReturnBoolean(@NotNull final String methodName) {
final long startTime = System.currentTimeMillis();
long tid;
hub.outBytesLock().lock();
try {
tid = hub.writeHeader(startTime, channelID, hub.outWire());
hub.outWire().writeDocument(false, wireOut -> wireOut.writeEventName(() -> methodName)
);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
}
return readBoolean(tid, startTime);
}
@SuppressWarnings("SameParameterValue")
private void proxyReturnVoid(Runnable r) {
private void proxyReturnVoid(@NotNull final EventId eventId, Consumer<WireOut> consumer) {
final long startTime = System.currentTimeMillis();
long tid;
hub.outBytesLock().lock();
try {
tid = hub.writeHeader(startTime, channelID, hub.outWire());
r.run();
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
}
long tid = sendEvent(startTime, eventId, consumer);
readVoid(tid, startTime);
}
@SuppressWarnings("SameParameterValue")
private int proxyReturnInt(@NotNull final String methodName) {
final long startTime = System.currentTimeMillis();
private void proxyReturnVoid(@NotNull final EventId eventId) {
proxyReturnVoid(eventId, null);
}
private long sendEvent(final long startTime,
@NotNull final EventId eventId,
@Nullable final Consumer<WireOut> consumer) {
long tid;
hub.outBytesLock().lock();
try {
tid = hub.writeHeader(startTime, channelID, hub.outWire());
hub.outWire().writeDocument(false, wireOut -> wireOut.writeEventName(() -> methodName)
);
tid = hub.writeHeader(startTime, channelID, hub.outWire());
hub.outWire().writeDocument(false, wireOut -> {
wireOut.writeEventName(eventId);
if (consumer == null)
wireOut.writeValue().marshallable(EMPTY);
else
consumer.accept(wireOut);
});
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
}
return readInt(tid, startTime);
return tid;
}
@SuppressWarnings("SameParameterValue")
private boolean proxyReturnBoolean(@NotNull final EventId eventId) {
final long startTime = System.currentTimeMillis();
return readBoolean(sendEvent(startTime, eventId, VOID_METHOD), startTime);
}
@SuppressWarnings("SameParameterValue")
private int proxyReturnInt(@NotNull final EventId eventId) {
final long startTime = System.currentTimeMillis();
return readInt(sendEvent(startTime, eventId, VOID_METHOD), startTime);
}
@Nullable
private <R> R proxyReturnTypedObject(
@@ -912,14 +869,23 @@ class ClientWiredStatelessChronicleMap<K, V>
@Nullable
private <R> R proxyReturnObject(
Class<R> rClass, @NotNull final EventId methodName, Object key) {
Class<R> rClass, @NotNull final EventId eventId, Object key) {
final long startTime = System.currentTimeMillis();
/* long tid = sendEvent(startTime, eventId, wireOut -> {
long tid;
wireOut.writeValue().marshallable(m -> {
final ValueOut vo = m.write(eventId.params()[0]);
writeField(vo, key);
tid = proxySend(methodName, startTime, key);
});
});*/
if (eventReturnsNull(methodName))
long tid = proxySend(eventId, startTime, key);
if (eventReturnsNull(eventId))
return null;
if (rClass == vClass)
@@ -933,11 +899,8 @@ class ClientWiredStatelessChronicleMap<K, V>
long tid;
hub.outBytesLock().lock();
try {
tid = hub.writeHeader(startTime, channelID, hub.outWire());
sendEvent(methodName, args);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java
+8 -6
View File
@@ -27,7 +27,6 @@ import net.openhft.chronicle.map.MapWireHandler.EventId;
import net.openhft.chronicle.network2.event.EventGroup;
import net.openhft.chronicle.wire.TextWire;
import net.openhft.chronicle.wire.Wire;
import net.openhft.chronicle.wire.WireOut;
import net.openhft.chronicle.wire.Wires;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@@ -40,7 +39,6 @@ import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;
import static net.openhft.chronicle.map.MapWireHandler.EventId.getApplicationVersion;
import static net.openhft.chronicle.map.MapWireHandler.SIZE_OF_SIZE;
@@ -626,7 +624,11 @@ public class ClientWiredStatelessTcpConnectionHub {
outBytesLock().lock();
try {
long tid = writeHeader(startTime, channelID, wire);
wire.writeDocument(false, wireOut -> wireOut.writeEventName(methodName));
wire.writeDocument(false, wireOut -> {
wireOut.writeEventName(methodName);
wireOut.writeValue().marshallable(w -> {
});
});
writeSocket(wire);
return tid;
@@ -643,13 +645,13 @@ public class ClientWiredStatelessTcpConnectionHub {
@SuppressWarnings("SameParameterValue")
@Nullable
String proxyReturnString(@NotNull final EventId messageId, short channelID, Wire outWire) {
String proxyReturnString(@NotNull final EventId eventId, short channelID, Wire outWire) {
final long startTime = System.currentTimeMillis();
long tid;
outBytesLock().lock();
try {
tid = proxySend(messageId, startTime, channelID, outWire);
tid = proxySend(eventId, startTime, channelID, outWire);
} finally {
outBytesLock().unlock();
}
@@ -690,7 +692,7 @@ public class ClientWiredStatelessTcpConnectionHub {
wire.writeDocument(true, wireOut -> {
wireOut.write(csp).text("MAP");
wireOut.write(MapWireHandlerBuilder.Fields.tid).int64(tid);
wireOut.write(timeStamp).int64(startTime);
// wireOut.write(timeStamp).int64(startTime);
wireOut.write(channelId).int16(channelID);
});
src/main/java/net/openhft/chronicle/map/MapWireHandler.java
-4
View File
@@ -204,8 +204,6 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
final long tid = inWire.read(Fields.tid).int64();
// timestamp
inWire.read(timeStamp).int64();
channelId = inWire.read(Fields.channelId).int16();
final Bytes<?> bytes1 = inWire.bytes();
@@ -435,8 +433,6 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
} finally {
// if (len > 4)
if (EventGroup.IS_DEBUG) {
long len = bytes.position() - SIZE_OF_SIZE;
if (len == 0) {