diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java index 2547cab2d..a48958023 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java @@ -20,8 +20,9 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.bytes.Bytes; import net.openhft.chronicle.engine.client.ClientWiredStatelessTcpConnectionHub; +import net.openhft.chronicle.engine.client.ClientWiredStatelessTcpConnectionHub.CoreFields; import net.openhft.chronicle.hash.function.SerializableFunction; -import net.openhft.chronicle.map.MapWireHandlerBuilder.Fields; +import net.openhft.chronicle.map.MapWireHandler.Fields; import net.openhft.chronicle.wire.*; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; @@ -308,8 +309,8 @@ class ClientWiredStatelessChronicleMap assert hub.outBytesLock().isHeldByCurrentThread(); hub.markSize(wire); hub.startTime(startTime); - wire.write(Fields.csp).text("MAP"); - wire.write(Fields.tid).int64(tid); + wire.write(MapWireHandler.Fields.csp).text("MAP"); + wire.write(CoreFields.tid).int64(tid); // wire.write(Fields.channelId).int16(channelID); @@ -320,7 +321,7 @@ class ClientWiredStatelessChronicleMap Map.Entry e = iterator.next(); - hub.outWire().write(Fields.hasNext).bool(iterator.hasNext()); + hub.outWire().write(MapWireHandler.Fields.hasNext).bool(iterator.hasNext()); // todo // writeField(wireOut, Fields.arg1, e.getKey()); @@ -404,9 +405,9 @@ class ClientWiredStatelessChronicleMap final Bytes bytes = wireIn.bytes(); while (bytes.remaining() > 0) { - boolean bool = wireIn.read(Fields.hasNext).bool(); + boolean bool = wireIn.read(MapWireHandler.Fields.hasNext).bool(); if (bool) { - usingCollection.add(readObject(Fields.reply, wireIn, null, vClass1)); + usingCollection.add(readObject(MapWireHandler.Fields.reply, wireIn, null, vClass1)); } else break OUTER; @@ -453,11 +454,11 @@ class ClientWiredStatelessChronicleMap checkIsData(wireIn); final Bytes bytes = wireIn.bytes(); while (bytes.remaining() > 0) { - boolean bool = wireIn.read(Fields.hasNext).bool(); + boolean bool = wireIn.read(MapWireHandler.Fields.hasNext).bool(); if (bool) { result.put( - readK(Fields.resultKey, wireIn, null), - readV(Fields.resultValue, wireIn, null)); + readK(MapWireHandler.Fields.resultKey, wireIn, null), + readV(MapWireHandler.Fields.resultValue, wireIn, null)); } else break OUTER; @@ -501,19 +502,19 @@ class ClientWiredStatelessChronicleMap checkIsData(wireIn); final Bytes bytes = wireIn.bytes(); while (bytes.remaining() > 0) { - boolean bool = wireIn.read(Fields.hasNext).bool(); + boolean bool = wireIn.read(MapWireHandler.Fields.hasNext).bool(); if (bool) { result.put( - readK(Fields.resultKey, wireIn, null), - readV(Fields.resultValue, wireIn, null)); + readK(MapWireHandler.Fields.resultKey, wireIn, null), + readV(MapWireHandler.Fields.resultValue, wireIn, null)); } else break OUTER; if (bytes.remaining() > 0) { - boolean isException = wireIn.read(Fields.isException).bool(); + boolean isException = wireIn.read(MapWireHandler.Fields.isException).bool(); if (isException) - throw new RuntimeException(wireIn.read(Fields.reply).text()); + throw new RuntimeException(wireIn.read(MapWireHandler.Fields.reply).text()); } } @@ -543,7 +544,7 @@ class ClientWiredStatelessChronicleMap try { final Wire wire = hub.proxyReply(timeoutTime, tid); checkIsData(wire); - return wire.read(Fields.reply).int64(); + return wire.read(MapWireHandler.Fields.reply).int64(); } finally { hub.inBytesLock().unlock(); } @@ -685,7 +686,7 @@ class ClientWiredStatelessChronicleMap try { final Wire wireIn = hub.proxyReply(timeoutTime, tid); checkIsData(wireIn); - return wireIn.read(Fields.reply).bool(); + return wireIn.read(MapWireHandler.Fields.reply).bool(); } finally { hub.inBytesLock().unlock(); } @@ -701,7 +702,7 @@ class ClientWiredStatelessChronicleMap try { final Wire wireIn = hub.proxyReply(timeoutTime, tid); checkIsData(wireIn); - return wireIn.read(Fields.reply).int32(); + return wireIn.read(MapWireHandler.Fields.reply).int32(); } finally { hub.inBytesLock().unlock(); } @@ -836,7 +837,7 @@ class ClientWiredStatelessChronicleMap return null; if (resultType == vClass) - return (R) readValue(Fields.reply, tid, startTime, null); + return (R) readValue(MapWireHandler.Fields.reply, tid, startTime, null); else throw new UnsupportedOperationException("class of type class=" + resultType + @@ -854,7 +855,7 @@ class ClientWiredStatelessChronicleMap return null; if (rClass == vClass) - return (R) readValue(Fields.reply, tid, startTime, null); + return (R) readValue(MapWireHandler.Fields.reply, tid, startTime, null); else throw new UnsupportedOperationException("class of type class=" + rClass + " is not " + "supported"); @@ -879,7 +880,7 @@ class ClientWiredStatelessChronicleMap hub.outBytesLock().lock(); try { tid = writeHeader(startTime); - hub.outWire().write(Fields.eventName).text(methodName.toString()); + hub.outWire().write(MapWireHandler.Fields.eventName).text(methodName.toString()); hub.outWire().writeDocument(false, wireOut -> { final ValueOut out = wireOut.writeEventName(methodName); diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java index b45cc41a3..f5eca0a81 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java @@ -27,7 +27,6 @@ import net.openhft.chronicle.engine.client.ClientWiredStatelessTcpConnectionHub; import net.openhft.chronicle.hash.ChronicleHashInstanceBuilder; import net.openhft.chronicle.hash.impl.util.BuildVersion; import net.openhft.chronicle.hash.replication.ReplicationHub; -import net.openhft.chronicle.map.MapWireHandlerBuilder.Fields; import net.openhft.chronicle.network.WireHandler; import net.openhft.chronicle.network.event.EventGroup; import net.openhft.chronicle.network.event.WireHandlers; @@ -52,25 +51,24 @@ import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Supplier; +import static net.openhft.chronicle.engine.client.ClientWiredStatelessTcpConnectionHub.CoreFields; import static net.openhft.chronicle.map.MapWireHandler.EventId.*; +import static net.openhft.chronicle.map.MapWireHandler.Fields.*; import static net.openhft.chronicle.map.MapWireHandler.Params.*; -import static net.openhft.chronicle.map.MapWireHandlerBuilder.Fields.*; /** * @author Rob Austin. */ -class MapWireHandler implements WireHandler, Consumer { - +public class MapWireHandler implements WireHandler, Consumer { private static final Logger LOG = LoggerFactory.getLogger(MapWireHandler.class); public static final int MAP_SERVICE = 3; public static final int SIZE_OF_SIZE = ClientWiredStatelessTcpConnectionHub.SIZE_OF_SIZE; - private final Map incompleteWork = new HashMap(); + private final Map incompleteWork = new HashMap<>(); private final ArrayList bytesChronicleMaps = new ArrayList<>(); @NotNull private final Supplier>> mapFactory; - private final Supplier>> channelNameToIdFactory; private Wire inWire = null; private Wire outWire = null; @@ -85,6 +83,32 @@ class MapWireHandler implements WireHandler, Consumer { } }; + private final Consumer metaDataConsumer = new Consumer() { + + @Override + public void accept(WireIn wireIn) { + + wireIn.read(csp).text(cspText); + + final int i = cspText.lastIndexOf("/"); + + if (i != -1 && i < (cspText.length() - 1)) { + final String channelStr = cspText.substring(i + 1, cspText.length() - "#MAP".length()); + try { + channelId = MapWireHandler.this.acquireChannelId(channelStr); + } catch (IOException e) { + // todo send to user + LOG.error("", e); + } + } else + channelId = 0; + + tid = inWire.read(CoreFields.tid).int64(); + + } + }; + + private final Consumer writeEntry = new Consumer>>() { @Override @@ -130,9 +154,8 @@ class MapWireHandler implements WireHandler, Consumer { @NotNull final ReplicationHub hub) { this.mapFactory = mapFactory; - this.channelNameToIdFactory = channelNameToIdFactory; - this.hub = hub; + try { channelNameToId = (ChronicleMap) channelNameToIdFactory.get() .replicatedViaChannel(hub.createChannel(MAP_SERVICE)).create(); @@ -142,19 +165,25 @@ class MapWireHandler implements WireHandler, Consumer { } @Override - public void accept(WireHandlers wireHandlers) { + public void accept(@NotNull final WireHandlers wireHandlers) { this.publishLater = wireHandlers; } + + private void writeChunked( + @NotNull final Function function, + @NotNull final Consumer c) throws StreamCorruptedException { + writeChunked(function, c, Long.MAX_VALUE); + } + /** - * @param tid the transaction id - * @param function provides that returns items bases on a BytesChronicleMap - * @param c an iterator that contains items - * @param maxEntries the maximum number of items that can be written + * @param function provides that returns items bases on a BytesChronicleMap + * @param c an iterator that contains items + * @param maxEntries the maximum number of items that can be written * @throws StreamCorruptedException */ private void writeChunked( - long tid, @NotNull final Function function, + @NotNull final Function function, @NotNull final Consumer c, long maxEntries) throws StreamCorruptedException { final BytesChronicleMap m = bytesMap(channelId); @@ -165,7 +194,7 @@ class MapWireHandler implements WireHandler, Consumer { @Override public void process(Wire in, Wire out) throws StreamCorruptedException { - outWire.write(Fields.tid).int64(tid); + outWire.write(CoreFields.tid).int64(tid); // this allows us to write more data than the buffer will allow for (int count = 0; ; count++) { @@ -203,20 +232,13 @@ class MapWireHandler implements WireHandler, Consumer { } - private void writeChunked( - long tid, - @NotNull final Function function, - @NotNull final Consumer c) throws StreamCorruptedException { - writeChunked(tid, function, c, Long.MAX_VALUE); - } - @Override - public void process(Wire in, Wire out) throws StreamCorruptedException { + public void process(@NotNull final Wire in, @NotNull final Wire out) throws StreamCorruptedException { try { this.inWire = in; this.outWire = out; - onEvent(); + inWire.readDocument(metaDataConsumer, dataConsumer); } catch (Exception e) { LOG.error("", e); } @@ -247,7 +269,7 @@ class MapWireHandler implements WireHandler, Consumer { * @param fromName the name of the channel * @return the id assosiated with this name */ - int acquireChannelId(@NotNull String fromName) throws IOException { + int acquireChannelId(@NotNull final String fromName) throws IOException { // todo this is a horrible hack, it slow and NOT processor safe, but was added to get // todo something working for now. @@ -272,274 +294,251 @@ class MapWireHandler implements WireHandler, Consumer { } + final StringBuilder cspText = new StringBuilder(); + public long tid; + private final Consumer dataConsumer = new Consumer() { - @SuppressWarnings("UnusedReturnValue") - void onEvent() throws IOException { + @Override + public void accept(WireIn wireIn) { - StringBuilder cspText = Wires.acquireStringBuilder(); - // it is assumed by this point that the buffer has all the bytes in it for this message + final StringBuilder eventName = Wires.acquireStringBuilder(); + final ValueIn valueIn = inWire.readEventName(eventName); - inWire.read(csp).text(cspText); - - final int i = cspText.lastIndexOf("/"); - if (i != -1 && i < (cspText.length() - 1)) { - final String channelStr = cspText.substring(i + 1, cspText.length() - "#MAP".length()); - channelId = acquireChannelId(channelStr); - - } else - channelId = 0; - - final long tid = inWire.read(Fields.tid).int64(); - - - final Bytes bytes1 = inWire.bytes(); - - final int body = bytes1.readVolatileInt(); - assert Wires.isData(body); - - final StringBuilder eventName = Wires.acquireStringBuilder(); - final ValueIn valueIn = inWire.readEventName(eventName); - - if (!incompleteWork.isEmpty()) { - Runnable runnable = incompleteWork.get(tid); - if (runnable != null) { - runnable.run(); - return; - } - } - - final Bytes outBytes = outWire.bytes(); - final Bytes bytes = outBytes; - - try { - - if (put.contentEquals(eventName)) { - - valueIn.marshallable(wire -> { - - final Params[] params = putIfAbsent.params; - final byte[] key = wire.read(params[0]).bytes(); - final byte[] value = wire.read(params[1]).bytes(); - - writeVoid(b -> ((Map) b.delegate).putIfAbsent(key, value)); - - }); - - return; + if (!incompleteWork.isEmpty()) { + Runnable runnable = incompleteWork.get(CoreFields.tid); + if (runnable != null) { + runnable.run(); + return; + } } + final Bytes outBytes = outWire.bytes(); + final Bytes bytes = outBytes; - if (keySet.contentEquals(eventName)) { - writeChunked(tid, map -> map.keySet().iterator(), writeElement); - return; - } + try { - if (values.contentEquals(eventName)) { - writeChunked(tid, map -> map.delegate.values().iterator(), writeElement); - return; - } + if (put.contentEquals(eventName)) { - if (entrySet.contentEquals(eventName)) { - writeChunked(tid, m -> m.delegate.entrySet().iterator(), writeEntry); - return; - } + valueIn.marshallable(wire -> { - if (entrySetRestricted.contentEquals(eventName)) { - long maxEntries = inWire.read(arg1).int64(); - writeChunked(tid, m -> m.delegate.entrySet().iterator(), writeEntry,maxEntries); - return; - } + final Params[] params = putIfAbsent.params; + final byte[] key = wire.read(params[0]).bytes(); + final byte[] value = wire.read(params[1]).bytes(); - if (putAll.contentEquals(eventName)) { - putAll(tid); - return; - } + writeVoid(b -> ((Map) b.delegate).putIfAbsent(key, value)); - outWire.writeDocument(true, wire -> outWire.write(Fields.tid).int64(tid)); - - // write the transaction id - if (createChannel.contentEquals(eventName)) { - writeVoid(() -> { - - short channelId1 = valueIn.int16(); - mapFactory.get().replicatedViaChannel(hub.createChannel(channelId1)).create(); - return null; - }); - return; - } - - if (EventId.remoteIdentifier.contentEquals(eventName)) { - this.remoteIdentifier = valueIn.int8(); - return; - } - - if (longSize.contentEquals(eventName)) { - write(b -> outWire.write(reply).int64(b.longSize())); - return; - } - - if (isEmpty.contentEquals(eventName)) { - write(b -> outWire.write(reply).bool(b.isEmpty())); - return; - } - - if (containsKey.contentEquals(eventName)) { - // todo remove the toByteArray(..) - write(b -> outWire.write(reply) - .bool(b.delegate.containsKey(toByteArray(valueIn)))); - return; - } - - if (containsValue.contentEquals(eventName)) { - // todo remove the toByteArray(..) - write(b -> outWire.write(reply) - .bool(b.delegate.containsValue(toByteArray(valueIn)))); - return; - } - - if (get.contentEquals(eventName)) { - // todo remove the toByteArray(..) - writeValueUsingDelegate(map -> { - final byte[] key = toByteArray(valueIn); - final byte[] value = map.get(key); - return value; - }); - return; - } - - if (getAndPut.contentEquals(eventName)) { - - valueIn.marshallable(wire -> { - - final Params[] params = getAndPut.params; - final byte[] key1 = wire.read(params[0]).bytes(); - - final byte[] value1 = wire.read(params[1]).bytes(); - - MapWireHandler.this.writeValue(b -> { - final byte[] result = b.put(key1, value1); - return result; }); - }); + return; + } - return; - } - if (remove.contentEquals(eventName)) { + if (keySet.contentEquals(eventName)) { + writeChunked(map -> map.keySet().iterator(), writeElement); + return; + } - MapWireHandler.this.writeValue(b -> { - return b.remove(toByteArray(valueIn)); - }); + if (values.contentEquals(eventName)) { + writeChunked(map -> map.delegate.values().iterator(), writeElement); + return; + } - return; - } + if (entrySet.contentEquals(eventName)) { + writeChunked(new Function() { + @Override + public Iterator apply(BytesChronicleMap m) { + return m.delegate.entrySet().iterator(); + } + }, writeEntry); + return; + } - if (clear.contentEquals(eventName)) { - writeVoid(BytesChronicleMap::clear); - return; - } + if (entrySetRestricted.contentEquals(eventName)) { + long maxEntries = inWire.read(arg1).int64(); + writeChunked(m -> m.delegate.entrySet().iterator(), writeEntry, maxEntries); + return; + } - if (replace.contentEquals(eventName)) { + if (putAll.contentEquals(eventName)) { + putAll(tid); + return; + } + + outWire.writeDocument(true, wire -> outWire.write(CoreFields.tid).int64(tid)); + + // write the transaction id + if (createChannel.contentEquals(eventName)) { + writeVoid(() -> { + + short channelId1 = valueIn.int16(); + mapFactory.get().replicatedViaChannel(hub.createChannel(channelId1)).create(); + return null; + }); + return; + } + + + if (longSize.contentEquals(eventName)) { + write(b -> outWire.write(reply).int64(b.longSize())); + return; + } + + if (isEmpty.contentEquals(eventName)) { + write(b -> outWire.write(reply).bool(b.isEmpty())); + return; + } + + if (containsKey.contentEquals(eventName)) { + // todo remove the toByteArray(..) + write(b -> outWire.write(reply) + .bool(b.delegate.containsKey(toByteArray(valueIn)))); + return; + } + + if (containsValue.contentEquals(eventName)) { + // todo remove the toByteArray(..) + write(b -> outWire.write(reply) + .bool(b.delegate.containsValue(toByteArray(valueIn)))); + return; + } + + if (get.contentEquals(eventName)) { + // todo remove the toByteArray(..) + writeValueUsingDelegate(map -> { + final byte[] key = toByteArray(valueIn); + final byte[] value = map.get(key); + return value; + }); + return; + } + + if (getAndPut.contentEquals(eventName)) { + + valueIn.marshallable(wire -> { + + final Params[] params = getAndPut.params; + final byte[] key1 = wire.read(params[0]).bytes(); + + final byte[] value1 = wire.read(params[1]).bytes(); + + MapWireHandler.this.writeValue(b -> { + final byte[] result = b.put(key1, value1); + return result; + }); + + }); + + return; + } + + if (remove.contentEquals(eventName)) { + + MapWireHandler.this.writeValue(b -> b.remove(toByteArray(valueIn))); + + return; + } + + if (clear.contentEquals(eventName)) { + writeVoid(BytesChronicleMap::clear); + return; + } + + if (replace.contentEquals(eventName)) { // todo fix this this is a hack to get to work for now. // todo may use something like : // todo bytesMap.replace(reader, reader, timestamp, identifier()); - valueIn.marshallable(wire -> { + valueIn.marshallable(wire -> { + final Params[] params = replace.params; + final byte[] key = wire.read(params[0]).bytes(); + final byte[] value = wire.read(params[1]).bytes(); - final Params[] params = replace.params; - final byte[] key = wire.read(params[0]).bytes(); - final byte[] value = wire.read(params[1]).bytes(); + writeValueFromBytes(b -> ((Map) b.delegate).replace(key, value)); - writeValueFromBytes(b -> ((Map) b.delegate).replace(key, value)); - - }); + }); - return; - } + return; + } - if (replaceWithOldAndNewValue.contentEquals(eventName)) { + if (replaceWithOldAndNewValue.contentEquals(eventName)) { + write(bytesMap -> { + final net.openhft.lang.io.Bytes reader = toReader(valueIn, + replaceWithOldAndNewValue.params); + boolean result = bytesMap.replace(reader, reader, reader); + outWire.write(Fields.reply).bool(result); + }); - write(bytesMap -> { - final net.openhft.lang.io.Bytes reader = toReader(valueIn, - replaceWithOldAndNewValue.params); - boolean result = bytesMap.replace(reader, reader, reader); - outWire.write(Fields.reply).bool(result); - }); + return; + } - return; - } + if (putIfAbsent.contentEquals(eventName)) { + valueIn.marshallable(wire -> { + final Params[] params = putIfAbsent.params; + final byte[] key = wire.read(params[0]).bytes(); + final byte[] value = wire.read(params[1]).bytes(); - if (putIfAbsent.contentEquals(eventName)) { + writeValueFromBytes(b -> ((Map) b.delegate).putIfAbsent(key, value)); - valueIn.marshallable(wire -> { + }); - final Params[] params = putIfAbsent.params; - final byte[] key = wire.read(params[0]).bytes(); - final byte[] value = wire.read(params[1]).bytes(); + return; + } - writeValueFromBytes(b -> ((Map) b.delegate).putIfAbsent(key, value)); - - }); - - return; - } - - if (removeWithValue.contentEquals(eventName)) { - write(bytesMap -> { - final net.openhft.lang.io.Bytes reader = toReader(valueIn, removeWithValue.params); - // todo call outWire.write(result) - // .bool(bytesMap.remove(reader, reader, timestamp, identifier())); - outWire.write(reply).bool(bytesMap.remove(reader, reader)); - }); - return; - } + if (removeWithValue.contentEquals(eventName)) { + write(bytesMap -> { + final net.openhft.lang.io.Bytes reader = toReader(valueIn, removeWithValue.params); + // todo call outWire.write(result) + // .bool(bytesMap.remove(reader, reader, timestamp, identifier())); + outWire.write(reply).bool(bytesMap.remove(reader, reader)); + }); + return; + } - if (getApplicationVersion.contentEquals(eventName)) { - write(b -> outWire.write(reply).text(applicationVersion())); - return; - } + if (getApplicationVersion.contentEquals(eventName)) { + write(b -> outWire.write(reply).text(applicationVersion())); + return; + } - if (persistedDataVersion.contentEquals(eventName)) { - write(b -> outWire.write(reply).text(persistedDataVersion())); + if (persistedDataVersion.contentEquals(eventName)) { + write(b -> outWire.write(reply).text(persistedDataVersion())); + return; + } - return; - } + if (hashCode.contentEquals(eventName)) { + write(b -> outWire.write(reply).int32(b.hashCode())); + return; + } - if (hashCode.contentEquals(eventName)) { - write(b -> outWire.write(reply).int32(b.hashCode())); - return; - } + throw new IllegalStateException("unsupported event=" + eventName); - throw new IllegalStateException("unsupported event=" + eventName); + } catch (Exception e) { + LOG.error("", e); + } finally { - } finally { - - if (EventGroup.IS_DEBUG) { - long len = bytes.position() - SIZE_OF_SIZE; - if (len == 0) { - System.out.println("--------------------------------------------\n" + - "server writes:\n\n"); - } else { + if (EventGroup.IS_DEBUG) { + long len = bytes.position() - SIZE_OF_SIZE; + if (len == 0) { + System.out.println("--------------------------------------------\n" + + "server writes:\n\n"); + } else { - System.out.println("--------------------------------------------\n" + + System.out.println("--------------------------------------------\n" + "server writes:\n\n" + Wires.fromSizePrefixedBlobs(bytes, SIZE_OF_SIZE, len)); + } } } + } - - - } - + }; // todo remove private byte[] toBytes(WireKey fieldName) { @@ -611,7 +610,7 @@ class MapWireHandler implements WireHandler, Consumer { incompleteWork.remove(tid); - outWire.write(Fields.tid).int64(tid); + outWire.write(CoreFields.tid).int64(tid); writeVoid(() -> { bytesMap.delegate.putAll((Map) collectData); @@ -672,7 +671,6 @@ class MapWireHandler implements WireHandler, Consumer { /** * creates a lang buffer that holds just the payload of the args * - * @return a new lang buffer containing the bytes of the args */ private net.openhft.lang.io.Bytes toReader(ValueIn valueIn, Params... params) { @@ -703,14 +701,12 @@ class MapWireHandler implements WireHandler, Consumer { } - private byte[] toByteArray(ValueIn valueIn) { final String text = valueIn.text(); return text.getBytes(); - } /** @@ -816,7 +812,7 @@ class MapWireHandler implements WireHandler, Consumer { // so we wil have to send the exception as a String outWire.bytes().reset(); - WireOut o = out.write(reply) + final WireOut o = out.write(reply) .type(e.getClass().getSimpleName()); if (e.getMessage() != null) { @@ -873,13 +869,12 @@ class MapWireHandler implements WireHandler, Consumer { } + try { + outWire.bytes().mark(); + process.accept(bytesMap); - try { - outWire.bytes().mark(); - process.accept(bytesMap); - - } catch (Exception e) { - outWire.writeDocument(false, out -> { + } catch (Exception e) { + outWire.writeDocument(false, out -> { // bytes.reset(); // the idea of wire is that is platform independent, // so we wil have to send the exception as a String @@ -966,5 +961,22 @@ class MapWireHandler implements WireHandler, Consumer { } + // note : peter has asked for these to be in camel case + enum Fields implements WireKey { + hasNext, + timeStamp, + channelId, + eventName, + csp, + reply, + resultKey, + resultValue, + arg1, + arg2, + arg3, + isException, + exception, + resultIsNull + } } diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java b/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java index d78c88c79..90936d365 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java @@ -33,39 +33,5 @@ import java.util.function.Supplier; */ public class MapWireHandlerBuilder { - public static WireHandler of( - @NotNull Supplier>> mapFactory, - @NotNull Supplier>> - channelNameToIdFactory, - @NotNull ReplicationHub hub, - byte localIdentifier, - @NotNull Map channels) { - - return new MapWireHandler<>( - mapFactory, - channelNameToIdFactory, - hub, - localIdentifier, - channels); - } - - // note : peter has asked for these to be in camel case - public enum Fields implements WireKey { - hasNext, - timeStamp, - channelId, - eventName, - csp, - tid, // transactionId - reply, - resultKey, - resultValue, - arg1, - arg2, - arg3, - isException, - exception, - resultIsNull - } }