diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java index 39516e456..29bafd771 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java @@ -47,6 +47,16 @@ class ClientWiredStatelessChronicleMap private static final Logger LOG = LoggerFactory.getLogger(ClientWiredStatelessChronicleMap.class); + private static final WriteMarshallable EMPTY = wire -> { + }; + public static final Consumer VOID_METHOD = new Consumer() { + @Override + public void accept(WireOut wireOut) { + + wireOut.writeValue().marshallable(EMPTY); + } + }; + private final ClientWiredStatelessTcpConnectionHub hub; protected Class kClass; @@ -198,7 +208,7 @@ class ClientWiredStatelessChronicleMap @Override public int hashCode() { - return proxyReturnInt(hashCode.toString()); + return proxyReturnInt(hashCode); } @NotNull @@ -218,7 +228,7 @@ class ClientWiredStatelessChronicleMap } public boolean isEmpty() { - return proxyReturnBoolean(isEmpty.toString()); + return proxyReturnBoolean(isEmpty); } public boolean containsKey(Object key) { @@ -235,7 +245,7 @@ class ClientWiredStatelessChronicleMap } public long longSize() { - return proxyReturnLong(longSize.toString()); + return proxyReturnLong(longSize); } @Override @@ -272,15 +282,8 @@ class ClientWiredStatelessChronicleMap @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 map) { @@ -306,7 +309,7 @@ class ClientWiredStatelessChronicleMap 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 } public void clear() { - proxyReturnVoid(() -> hub.outWire().writeDocument(false, wireOut -> wireOut.writeEventName(clear) - )); + proxyReturnVoid(clear); } @@ -705,41 +707,18 @@ class ClientWiredStatelessChronicleMap 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 @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 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 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 proxyReturnTypedObject( @@ -912,14 +869,23 @@ class ClientWiredStatelessChronicleMap @Nullable private R proxyReturnObject( - Class rClass, @NotNull final EventId methodName, Object key) { + Class 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 long tid; hub.outBytesLock().lock(); try { - tid = hub.writeHeader(startTime, channelID, hub.outWire()); sendEvent(methodName, args); - - hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java index 80f90c974..991c5f96a 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java @@ -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); }); diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java index 9e3a72f20..eda2e89aa 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java @@ -204,8 +204,6 @@ class MapWireHandler implements WireHandler, Consumer { 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 implements WireHandler, Consumer { } finally { - - // if (len > 4) if (EventGroup.IS_DEBUG) { long len = bytes.position() - SIZE_OF_SIZE; if (len == 0) {