From 088d5540da851d56966256416326edd2d9bfa712 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Wed, 22 Apr 2015 14:12:52 +0100 Subject: [PATCH] providing support for the engine to access the map locally on the server --- .../map/ClientWiredStatelessChronicleMap.java | 12 +- .../chronicle/map/MapStatelessClient.java | 122 +++-------- .../chronicle/map/MapWireConnectionHub.java | 164 ++++++++++++++ .../map/MapWireConnectionHubByName.java | 143 +++++++++++++ .../openhft/chronicle/map/MapWireHandler.java | 200 +++--------------- 5 files changed, 368 insertions(+), 273 deletions(-) create mode 100644 src/main/java/net/openhft/chronicle/map/MapWireConnectionHub.java create mode 100644 src/main/java/net/openhft/chronicle/map/MapWireConnectionHubByName.java diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java index b3139eafe..f5c6ecf05 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java @@ -53,6 +53,7 @@ class ClientWiredStatelessChronicleMap extends MapStatelessClient VOID_PARAMETERS = out -> out.marshallable(AbstactStatelessClient.EMPTY); + private final Class vClass; protected Class kClass; @@ -73,6 +74,8 @@ class ClientWiredStatelessChronicleMap extends MapStatelessClient extends MapStatelessClient extends MapStatelessClient extends MapStatelessClient> entrySet() { - long cid = proxyReturnWireConsumer(entrySet, (WireIn wireIn) -> { final long[] cidRef = new long[1]; final StringBuilder type = Wires.acquireStringBuilder(); @@ -343,7 +345,7 @@ class ClientWiredStatelessChronicleMap extends MapStatelessClient(channelName, hub, cid, vClass); @@ -451,7 +453,5 @@ class ClientWiredStatelessChronicleMap extends MapStatelessClient extends - AbstactStatelessClient { +public abstract class MapStatelessClient + extends AbstactStatelessClient { - protected Class vClass; + private Class vClass; /** * @param channelName @@ -29,97 +28,10 @@ public abstract class MapStatelessClient ex @NotNull String type, long cid, @NotNull final Class vClass) { super(channelName, hub, type, cid); + this.vClass = vClass; } - protected Object readValue(WireKey argName, long tid, long startTime, final V usingValue) { - assert !hub.outBytesLock().isHeldByCurrentThread(); - long timeoutTime = startTime + hub.timeoutMs; - - hub.inBytesLock().lock(); - try { - - final Wire wireIn = hub.proxyReply(timeoutTime, tid); - checkIsData(wireIn); - - - return readV(argName, wireIn, usingValue); - - } finally { - hub.inBytesLock().unlock(); - } - } - - private Object readV(WireKey argName, Wire wireIn, V usingValue) { - return readObject(argName, wireIn, usingValue, vClass); - } - - protected E readObject(WireKey argName, Wire wireIn, E usingValue, Class clazz) { - - final ValueIn valueIn = wireIn.read(argName); - if (valueIn.isNull()) - return null; - - if (StringBuilder.class.isAssignableFrom(clazz)) { - valueIn.text((StringBuilder) usingValue); - return usingValue; - } else if (Marshallable.class.isAssignableFrom(clazz)) { - - - final E v; - if (usingValue == null) - try { - v = clazz.newInstance(); - } catch (Exception e) { - throw new RuntimeException(e); - } - else - v = usingValue; - - - valueIn.marshallable((Marshallable) v); - return v; - - } else if (String.class.isAssignableFrom(clazz)) { - //noinspection unchecked - return (E) valueIn.text(); - - } else if (Long.class.isAssignableFrom(clazz)) { - //noinspection unchecked - return (E) (Long) valueIn.int64(); - } else if (Double.class.isAssignableFrom(clazz)) { - //noinspection unchecked - return (E) (Double) valueIn.float64(); - - } else if (Integer.class.isAssignableFrom(clazz)) { - //noinspection unchecked - return (E) (Integer) valueIn.int32(); - - } else if (Float.class.isAssignableFrom(clazz)) { - //noinspection unchecked - return (E) (Float) valueIn.float32(); - - } else if (Short.class.isAssignableFrom(clazz)) { - //noinspection unchecked - return (E) (Short) valueIn.int16(); - - } else if (Character.class.isAssignableFrom(clazz)) { - //noinspection unchecked - final String text = valueIn.text(); - if (text == null || text.length() == 0) - return null; - return (E) (Character) text.charAt(0); - - } else if (Byte.class.isAssignableFrom(clazz)) { - //noinspection unchecked - return (E) (Byte) valueIn.int8(); - - - } else { - throw new IllegalStateException("unsupported type"); - } - } - @Nullable protected R proxyReturnTypedObject( @NotNull final E eventId, @@ -141,7 +53,8 @@ public abstract class MapStatelessClient ex } @Nullable - protected R proxyReturnObject(Class rClass, @NotNull final E eventId, Object field) { + protected R proxyReturnObject(@NotNull final Class rClass, + @NotNull final E eventId, Object field) { final long startTime = System.currentTimeMillis(); final long tid = sendEvent(startTime, eventId, out -> writeField(out, field)); @@ -159,7 +72,26 @@ public abstract class MapStatelessClient ex protected abstract boolean eventReturnsNull(@NotNull E methodName); + protected V readValue(WireKey argName, long tid, long startTime, final V usingValue) { + assert !hub.outBytesLock().isHeldByCurrentThread(); + long timeoutTime = startTime + hub.timeoutMs; + + hub.inBytesLock().lock(); + try { + + final Wire wireIn = hub.proxyReply(timeoutTime, tid); + checkIsData(wireIn); + return readV(argName, wireIn, usingValue); + } finally { + hub.inBytesLock().unlock(); + } + } + + + private V readV(WireKey argName, Wire wireIn, V usingValue) { + return readObject(argName, wireIn, usingValue, vClass); + } } diff --git a/src/main/java/net/openhft/chronicle/map/MapWireConnectionHub.java b/src/main/java/net/openhft/chronicle/map/MapWireConnectionHub.java new file mode 100644 index 000000000..12ecc8390 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/MapWireConnectionHub.java @@ -0,0 +1,164 @@ +package net.openhft.chronicle.map; + +import net.openhft.chronicle.hash.ChronicleHashInstanceBuilder; +import net.openhft.chronicle.hash.replication.ReplicationHub; +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; +import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Map; +import java.util.function.Supplier; + +import static java.util.concurrent.TimeUnit.SECONDS; +import static net.openhft.chronicle.map.ChronicleMapBuilder.of; + +/** + * Created by Rob Austin + */ +public class MapWireConnectionHub { + + private static final Logger LOG = LoggerFactory.getLogger(MapWireHandler.class); + public static final int MAP_SERVICE = 3; + private final byte localIdentifier; + + protected ChronicleMap channelNameToId; + private Supplier>> mapFactory; + private final Map channelMap; + private final ReplicationHub hub; + + private final ArrayList bytesChronicleMaps = new ArrayList<>(); + + + public MapWireConnectionHub( + @NotNull final Supplier>> mapFactory, + @NotNull final Supplier>> + channelNameToIdFactory, + byte localIdentifier, + int serverPort) throws IOException { + + + this.mapFactory = mapFactory; + this.localIdentifier = localIdentifier; + + final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig + .of(serverPort) + .heartBeatInterval(1, SECONDS); + + hub = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig).createWithId(localIdentifier); + + // this is how you add maps after the custer is created + of(byte[].class, byte[].class) + .instance().replicatedViaChannel(hub.createChannel((short) 1)).create(); + + channelNameToId = (ChronicleMap) channelNameToIdFactory.get() + .replicatedViaChannel(hub.createChannel(MAP_SERVICE)).create(); + + ChannelProvider provider = ChannelProvider.getProvider(hub); + channelMap = provider.chronicleChannelMap(); + + } + + + /** + * @return the next free channel id + */ + short getNextFreeChannel() { + + // todo this is a horrible hack, it slow and not processor safe, but was added to get + // todo something working for now. + + int max = 3; + for (Integer channel : channelNameToId.values()) { + max = Math.max(max, channel); + } + + return (short) (max + 1); + } + + /** + * gets the channel id for a name, or creates a new one if this name is not yet assosiated to a + * channel + * + * @param fromName the name of the channel + * @return the id associated with this name + */ + BytesChronicleMap acquireMap(@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. + + final Integer channelId = channelNameToId.get(fromName); + + if (channelId != null) + return bytesMap(channelId); + + final int nextFreeChannel = getNextFreeChannel(); + try { + + mapFactory.get().replicatedViaChannel(hub.createChannel(nextFreeChannel)).create(); + channelNameToId.put(fromName, nextFreeChannel); + + return bytesMap(nextFreeChannel); + } catch (IOException e) { + // todo send this error back to the user + LOG.error("", e); + throw e; + } + + + + } + + + /** + * this is used to push the data straight into the entry in memory + * + * @param channelId the ID of the map + * @return a BytesChronicleMap used to update the memory which holds the chronicle map + */ + @Nullable + BytesChronicleMap bytesMap(int channelId) { + + final BytesChronicleMap bytesChronicleMap = (channelId < bytesChronicleMaps.size()) + ? bytesChronicleMaps.get(channelId) + : null; + + if (bytesChronicleMap != null) + return bytesChronicleMap; + + // grow the array + for (int i = bytesChronicleMaps.size(); i <= channelId; i++) { + bytesChronicleMaps.add(null); + } + + final ReplicatedChronicleMap delegate = map(channelId); + final BytesChronicleMap element = new BytesChronicleMap(delegate); + bytesChronicleMaps.set(channelId, element); + return element; + + } + + /** + * gets the map for this channel id + * + * @param channelId the ID of the map + * @return the chronicle map with this {@code channelId} + */ + @NotNull + private ReplicatedChronicleMap map(int channelId) { + + // todo this cast is a bit of a hack, improve later + final ReplicatedChronicleMap map = + (ReplicatedChronicleMap) channelMap.get(channelId); + + if (map != null) + return map; + + throw new IllegalStateException(); + } + +} diff --git a/src/main/java/net/openhft/chronicle/map/MapWireConnectionHubByName.java b/src/main/java/net/openhft/chronicle/map/MapWireConnectionHubByName.java new file mode 100644 index 000000000..141059f50 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/MapWireConnectionHubByName.java @@ -0,0 +1,143 @@ +package net.openhft.chronicle.map; + +import net.openhft.chronicle.bytes.Bytes; +import net.openhft.chronicle.engine.client.ClientWiredStatelessTcpConnectionHub.CoreFields; +import net.openhft.chronicle.wire.TextWire; +import net.openhft.chronicle.wire.ValueOut; +import net.openhft.chronicle.wire.Wire; +import org.jetbrains.annotations.NotNull; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Collection; +import java.util.Map; +import java.util.Set; + +/** + * Created by Rob Austin + */ +public class MapWireConnectionHubByName implements Map { + + private final Class kClass; + private final Class vClass; + private final MapWireConnectionHub mapWireConnectionHub; + + + Bytes buffer = Bytes.elasticByteBuffer(); + + private final Map map; + + + Class wireType = TextWire.class; + + public MapWireConnectionHubByName(String name, + Class vClass, + Class kClass, + MapWireConnectionHub mapWireConnectionHub) throws IOException { + this.mapWireConnectionHub = mapWireConnectionHub; + final BytesChronicleMap b = this.mapWireConnectionHub.acquireMap(name); + this.map = (Map) b.delegate; + this.vClass = vClass; + this.kClass = kClass; + } + + @Override + public int size() { + return map.size(); + } + + @Override + public boolean isEmpty() { + return map.isEmpty(); + } + + + Wire toWire() { + buffer.clear(); + if (TextWire.class.isAssignableFrom(wireType)) { + return new TextWire(buffer); + } else throw new UnsupportedOperationException("todo"); + } + + + Wire toWire(byte[] bytes) { + + Bytes wrap = Bytes.wrap(bytes); + + if (TextWire.class.isAssignableFrom(wireType)) { + return new TextWire(wrap); + } else throw new UnsupportedOperationException("todo"); + } + + @Override + public boolean containsKey(Object key) { + return map.containsKey(bytes(key)); + } + + private byte[] bytes(Object key) { + ValueOut valueOut = toWire().getValueOut(); + AbstactStatelessClient.writeField(key, valueOut); + buffer.flip(); + byte[] bytes = new byte[(int) buffer.limit()]; + buffer.read(bytes); + return bytes; + } + + @Override + public boolean containsValue(Object value) { + return map.containsValue(bytes(value)); + + } + + @Override + public V get(Object key) { + byte[] bytes = map.get(bytes(key)); + + if (bytes == null) + return null; + + final Wire wire = toWire(bytes); + return AbstactStatelessClient.readObject(CoreFields.reply, wire, null, vClass); + } + + @Override + public V put(K key, V value) { + final Wire wire = toWire(map.put(bytes(key), bytes(value))); + return AbstactStatelessClient.readObject(CoreFields.reply, wire, null, vClass); + } + + @Override + public V remove(Object key) { + final Wire wire = toWire(map.remove(bytes(key))); + return AbstactStatelessClient.readObject(CoreFields.reply, wire, null, vClass); + } + + @Override + public void putAll(Map m) { + throw new UnsupportedOperationException("todo"); + } + + @Override + public void clear() { + map.clear(); + } + + @NotNull + @Override + public Set keySet() { + throw new UnsupportedOperationException("todo"); + } + + @NotNull + @Override + public Collection values() { + throw new UnsupportedOperationException("todo"); + } + + @NotNull + @Override + public Set> entrySet() { + return null; + } + +} diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java index 2a2814dca..d270c42a6 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java @@ -25,9 +25,7 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.bytes.Bytes; import net.openhft.chronicle.engine.client.ClientWiredStatelessTcpConnectionHub; import net.openhft.chronicle.engine.client.ParameterizeWireKey; -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.ClientWiredStatelessChronicleEntrySet.EntrySetEventId; import net.openhft.chronicle.network.WireHandler; import net.openhft.chronicle.network.event.EventGroup; @@ -35,19 +33,16 @@ import net.openhft.chronicle.network.event.WireHandlers; import net.openhft.chronicle.wire.*; import net.openhft.lang.io.DirectStore; import org.jetbrains.annotations.NotNull; -import org.jetbrains.annotations.Nullable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; import java.io.StreamCorruptedException; -import java.util.ArrayList; import java.util.HashMap; import java.util.Map; import java.util.concurrent.atomic.AtomicLong; 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.engine.client.ClientWiredStatelessTcpConnectionHub.CoreFields.csp; @@ -64,20 +59,19 @@ import static net.openhft.chronicle.wire.Wires.acquireStringBuilder; 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 ArrayList bytesChronicleMaps = new ArrayList<>(); - @NotNull - private final Supplier>> mapFactory; private final Map cidToCsp; + @NotNull + private final MapWireConnectionHub mapWireConnectionHub; private final Map cspToCid = new HashMap<>(); + private Wire inWire = null; private Wire outWire = null; - private void setCspTextFromCid(long cid) { cspText.setLength(0); cspText.append(cidToCsp.get(cid)); @@ -88,17 +82,13 @@ public class MapWireHandler implements WireHandler, Consumer @Override public void accept(WireIn wireIn) { - StringBuilder key = Wires.acquireStringBuilder(); + final StringBuilder key = Wires.acquireStringBuilder(); final ValueIn read = wireIn.read(key); - - if (csp.contentEquals(key)) { + if (csp.contentEquals(key)) read.text(cspText); - } else { - if (CoreFields.cid.contentEquals(key)) { - setCspTextFromCid(read.int64()); - } - } + else if (CoreFields.cid.contentEquals(key)) + setCspTextFromCid(read.int64()); final int slash = cspText.lastIndexOf("/"); final int hash = cspText.lastIndexOf("#"); @@ -107,13 +97,15 @@ public class MapWireHandler implements WireHandler, Consumer hash != -1 && hash < (cspText.length() - 1)) { final String channelStr = cspText.substring(slash + 1, hash); try { - channelId = MapWireHandler.this.acquireChannelId(channelStr); + bytesChronicleMap = mapWireConnectionHub.acquireMap(channelStr); + } catch (IOException e) { // todo send to user LOG.error("", e); } + } else - channelId = 0; + bytesChronicleMap = null; tid = inWire.read(CoreFields.tid).int64(); @@ -122,44 +114,16 @@ public class MapWireHandler implements WireHandler, Consumer private WireHandlers publishLater; - private byte localIdentifier; - private int channelId; - private Map channels; - private ReplicationHub hub; + private BytesChronicleMap bytesChronicleMap; - private ChronicleMap channelNameToId; public MapWireHandler( - @NotNull final Supplier>> mapFactory, - @NotNull final Supplier>> - channelNameToIdFactory, - @NotNull final ReplicationHub hub, byte localIdentifier, - @NotNull final Map channels, - @NotNull final Map cidToCsp) { - this(mapFactory, channelNameToIdFactory, hub, cidToCsp); - this.channels = channels; - this.localIdentifier = localIdentifier; + @NotNull final Map cidToCsp, + @NotNull final MapWireConnectionHub mapWireConnectionHub) throws IOException { - } - - public MapWireHandler( - @NotNull final Supplier>> mapFactory, - @NotNull final Supplier>> - channelNameToIdFactory, - @NotNull final ReplicationHub hub, - @NotNull final Map cidToCsp) { - - this.mapFactory = mapFactory; - this.hub = hub; this.cidToCsp = cidToCsp; - - try { - channelNameToId = (ChronicleMap) channelNameToIdFactory.get() - .replicatedViaChannel(hub.createChannel(MAP_SERVICE)).create(); - } catch (IOException e) { - e.printStackTrace(); - } + this.mapWireConnectionHub = mapWireConnectionHub; } @Override @@ -168,8 +132,6 @@ public class MapWireHandler implements WireHandler, Consumer } - - @Override public void process(@NotNull final Wire in, @NotNull final Wire out) throws StreamCorruptedException { try { @@ -182,55 +144,6 @@ public class MapWireHandler implements WireHandler, Consumer } - /** - * @return the next free channel id - */ - short getNextFreeChannel() { - - // todo this is a horrible hack, it slow and not processor safe, but was added to get - // todo something working for now. - - int max = 3; - for (Integer channel : channelNameToId.values()) { - max = Math.max(max, channel); - } - - return (short) (max + 1); - } - - - /** - * gets the channel id for a name, or creates a new one if this name is not yet assosiated to a - * channel - * - * @param fromName the name of the channel - * @return the id assosiated with this name - */ - 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. - - final Integer channelId = channelNameToId.get(fromName); - - if (channelId != null) - return channelId; - - final int nextFreeChannel = getNextFreeChannel(); - try { - - mapFactory.get().replicatedViaChannel(hub.createChannel(nextFreeChannel)).create(); - channelNameToId.put(fromName, nextFreeChannel); - - return nextFreeChannel; - } catch (IOException e) { - // todo send this error back to the user - LOG.error("", e); - throw e; - } - - } - final StringBuilder cspText = new StringBuilder(); public long tid; private final AtomicLong cid = new AtomicLong(); @@ -440,7 +353,6 @@ public class MapWireHandler implements WireHandler, Consumer } - if (replace.contentEquals(eventName)) { @@ -537,11 +449,6 @@ public class MapWireHandler implements WireHandler, Consumer }; - private void putAll(long tid) { - - // todo - } - @SuppressWarnings("SameReturnValue") private void writeValueFromBytes(final Function f) { @@ -564,7 +471,7 @@ public class MapWireHandler implements WireHandler, Consumer @NotNull private CharSequence persistedDataVersion() { - final BytesChronicleMap bytesChronicleMap = bytesMap(channelId); + if (bytesChronicleMap == null) return ""; return bytesChronicleMap.delegate.persistedDataVersion(); @@ -612,52 +519,6 @@ public class MapWireHandler implements WireHandler, Consumer return text.getBytes(); } - /** - * gets the map for this channel id - * - * @param channelId the ID of the map - * @return the chronicle map with this {@code channelId} - */ - @NotNull - private ReplicatedChronicleMap map(int channelId) { - - // todo this cast is a bit of a hack, improve later - final ReplicatedChronicleMap map = - (ReplicatedChronicleMap) channels.get(channelId); - - if (map != null) - return map; - - throw new IllegalStateException(); - } - - /** - * this is used to push the data straight into the entry in memory - * - * @param channelId the ID of the map - * @return a BytesChronicleMap used to update the memory which holds the chronicle map - */ - @Nullable - private BytesChronicleMap bytesMap(int channelId) { - - final BytesChronicleMap bytesChronicleMap = (channelId < bytesChronicleMaps.size()) - ? bytesChronicleMaps.get(channelId) - : null; - - if (bytesChronicleMap != null) - return bytesChronicleMap; - - // grow the array - for (int i = bytesChronicleMaps.size(); i <= channelId; i++) { - bytesChronicleMaps.add(null); - } - - final ReplicatedChronicleMap delegate = map(channelId); - final BytesChronicleMap element = new BytesChronicleMap(delegate); - bytesChronicleMaps.set(channelId, element); - return element; - - } @SuppressWarnings("SameReturnValue") private void writeValue(final Function, byte[]> f) { @@ -688,23 +549,23 @@ public class MapWireHandler implements WireHandler, Consumer } + @SuppressWarnings("SameReturnValue") - private void write(@NotNull Consumer c) { + void write(@NotNull Consumer c) { - final BytesChronicleMap bytesMap = bytesMap(channelId); - if (bytesMap == null) { - LOG.error("no map for channelId=" + channelId + " can be found."); + if (bytesChronicleMap == null) { + LOG.error("no map for channel."); return; } - bytesMap.output = null; + bytesChronicleMap.output = null; outWire.writeDocument(false, out -> { long position = 0; try { position = outWire.bytes().position(); - c.accept(bytesMap); + c.accept(bytesChronicleMap); } catch (Exception e) { // bytes.reset(); // the idea of wire is that is platform independent, @@ -714,9 +575,9 @@ public class MapWireHandler implements WireHandler, Consumer final WireOut o = out.write(reply) .type(e.getClass().getSimpleName()); - if (e.getMessage() != null) { + if (e.getMessage() != null) o.writeValue().text(e.getMessage()); - } + LOG.error("", e); } }); @@ -729,17 +590,16 @@ public class MapWireHandler implements WireHandler, Consumer private void writeVoid(@NotNull Consumer process) { // skip 4 bytes where we will write the size - final BytesChronicleMap bytesMap = bytesMap(channelId); - if (bytesMap == null) { - LOG.error("no map for channelId=" + channelId + " can be found."); + if (bytesChronicleMap == null) { + LOG.error("no map for channelId."); return; } try { outWire.bytes().mark(); - process.accept(bytesMap); + process.accept(bytesChronicleMap); } catch (Exception e) { outWire.writeDocument(false, out -> { @@ -758,10 +618,6 @@ public class MapWireHandler implements WireHandler, Consumer } - public void localIdentifier(byte localIdentifier) { - this.localIdentifier = localIdentifier; - } - enum Params implements WireKey { key, value,