diff --git a/src/main/java/net/openhft/chronicle/map/MapStatelessClient.java b/src/main/java/net/openhft/chronicle/map/MapStatelessClient.java index 50f6d296f..bdb34b59d 100644 --- a/src/main/java/net/openhft/chronicle/map/MapStatelessClient.java +++ b/src/main/java/net/openhft/chronicle/map/MapStatelessClient.java @@ -41,7 +41,7 @@ public abstract class MapStatelessClient< E extends ParameterizeWireKey> extends - +/* protected E proxyReturnE(@NotNull final WireKey eventId, @Nullable final O arg, @NotNull final Class eClass) { @@ -58,7 +58,7 @@ public abstract class MapStatelessClient< E extends ParameterizeWireKey> extends return proxyReturnWireConsumer(eventId, f -> f.read(CoreFields.reply) .object(eClass)); - } + } */ } diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandlerProcessor.java b/src/main/java/net/openhft/chronicle/map/MapWireHandlerProcessor.java index f718179a7..019536140 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandlerProcessor.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandlerProcessor.java @@ -37,7 +37,9 @@ import java.util.HashMap; import java.util.Map; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.atomic.AtomicLong; +import java.util.function.BiConsumer; import java.util.function.Consumer; +import java.util.function.Function; import static net.openhft.chronicle.map.MapWireHandlerProcessor.EventId.*; import static net.openhft.chronicle.map.MapWireHandlerProcessor.Params.*; @@ -48,13 +50,28 @@ import static net.openhft.chronicle.wire.Wires.acquireStringBuilder; /** * @author Rob Austin. */ -public class MapWireHandlerProcessor implements MapWireHandler>, Consumer { +public class MapWireHandlerProcessor implements MapWireHandler, K, V>, + Consumer { private CharSequence csp; + private BiConsumer vToWire; + private Function wireToK; + private Function wireToV; + @Override - public void process(Wire in, Wire out, ConcurrentMap map, CharSequence csp) throws - StreamCorruptedException { + public void process(@NotNull final Wire in, + @NotNull final Wire out, + @NotNull final ConcurrentMap map, + @NotNull final CharSequence csp, + @NotNull final BiConsumer vToWire, + @NotNull final Function kFromWire, + @NotNull final Function vFromWire) throws StreamCorruptedException { + + + this.vToWire = vToWire; + this.wireToK = kFromWire; + this.wireToV = vFromWire; try { this.inWire = in; @@ -68,6 +85,7 @@ public class MapWireHandlerProcessor implements MapWireHandler map; - - - private WireHandlers publishLater; - + private ConcurrentMap map; public MapWireHandlerProcessor(@NotNull final Map cidToCsp) throws IOException { this.cidToCsp = cidToCsp; @@ -164,7 +178,7 @@ public class MapWireHandlerProcessor implements MapWireHandler valueIn.marshallable(wire -> data.put( - wire.read(put.params()[0]).bytes(), - wire.read(put.params()[1]).bytes()))); + wireToK.apply(wire.read(put.params()[0])), + wireToV.apply(wire.read(put.params()[1]))))); } map.putAll(data); @@ -236,10 +250,11 @@ public class MapWireHandlerProcessor implements MapWireHandler { final Params[] params = putIfAbsent.params(); - final byte[] key = wire.read(params[0]).bytes(); - final byte[] value = wire.read(params[1]).bytes(); - final ValueOut v = outWire.write(reply); - writeValue((map).putIfAbsent(key, value)); + final K key = wireToK.apply(wire.read(params[0])); + final V value = wireToV.apply(wire.read(params[1])); + + final V v = map.putIfAbsent(key, value); + vToWire.accept(outWire.write(reply), v); }); @@ -301,19 +316,20 @@ public class MapWireHandlerProcessor implements MapWireHandler { final Params[] params = getAndPut.params(); + final K key = wireToK.apply(wire.read(params[0])); + final V value = wireToV.apply(wire.read(params[1])); - byte[] k = wire.read(params[0]).bytes(); - byte[] v = wire.read(params[1]).bytes(); - outWire.write(reply); - writeValue(map.put(k, v)); + vToWire.accept(outWire.write(reply), + map.put(key, value)); }); @@ -335,7 +351,8 @@ public class MapWireHandlerProcessor implements MapWireHandler { final Params[] params = replace.params(); - final byte[] key = wire.read(params[0]).bytes(); - final byte[] value = wire.read(params[1]).bytes(); - outWire.write(reply); - writeValue(map.replace(key, value)); + final K key = wireToK.apply(wire.read(params[0])); + final V value = wireToV.apply(wire.read(params[1])); + + vToWire.accept(outWire.write(reply), + map.replace(key, value)); }); @@ -357,9 +375,9 @@ public class MapWireHandlerProcessor implements MapWireHandler { final Params[] params = replaceWithOldAndNewValue.params(); - final byte[] key = wire.read(params[0]).bytes(); - final byte[] oldValue = wire.read(params[1]).bytes(); - final byte[] newValue = wire.read(params[2]).bytes(); + final K key = wireToK.apply(wire.read(params[0])); + final V oldValue = wireToV.apply(wire.read(params[1])); + final V newValue = wireToV.apply(wire.read(params[2])); outWire.write(reply).bool(map.replace(key, oldValue, newValue)); }); @@ -369,10 +387,10 @@ public class MapWireHandlerProcessor implements MapWireHandler { final Params[] params = putIfAbsent.params(); - final byte[] key = wire.read(params[0]).bytes(); - final byte[] value = wire.read(params[1]).bytes(); - outWire.write(reply); - writeValue(map.putIfAbsent(key, value)); + final K key = wireToK.apply(wire.read(params[0])); + final V value = wireToV.apply(wire.read(params[1])); + vToWire.accept(outWire.write(reply), + map.putIfAbsent(key, value)); }); @@ -429,22 +447,6 @@ public class MapWireHandlerProcessor implements MapWireHandler