diff --git a/pom.xml b/pom.xml index 2b883dfce..3e2072f0d 100755 --- a/pom.xml +++ b/pom.xml @@ -63,6 +63,24 @@ 2.2.0 + + net.openhft + chronicle-wire + 1.0.1-alpha-SNAPSHOT + + + + net.openhft + chronicle-bytes + 1.0.0-alpha-SNAPSHOT + + + + net.openhft + chronicle-network + 1.0.0-alpha-SNAPSHOT + + @@ -88,6 +106,16 @@ annotations + + net.openhft + chronicle-wire + + + + net.openhft + chronicle-bytes + + @@ -108,6 +136,12 @@ pax-url-aether + + net.openhft + chronicle-network + + + diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java index 885139a28..956c4276a 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java @@ -18,8 +18,8 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.replication.ThrottlingConfig; import net.openhft.chronicle.hash.impl.util.CloseablesManager; +import net.openhft.chronicle.hash.replication.ThrottlingConfig; import net.openhft.chronicle.map.ReplicatedChronicleMap.BytesReplicatedContextFactory; import net.openhft.lang.io.ByteBufferBytes; import net.openhft.lang.io.Bytes; @@ -54,7 +54,7 @@ abstract class AbstractChannelReplicator implements Closeable { public static final int BITS_IN_A_BYTE = 8; - public static final int SIZE_OF_SIZE = 4; + static final int SIZE_OF_SIZE = 4; public static final int SIZE_OF_TRANSACTION_ID = 8; public static final int SIZE_OF_TIME_SHIFT = 2; diff --git a/src/main/java/net/openhft/chronicle/map/BytesChronicleMap.java b/src/main/java/net/openhft/chronicle/map/BytesChronicleMap.java index f552148d7..d65fbca30 100644 --- a/src/main/java/net/openhft/chronicle/map/BytesChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/BytesChronicleMap.java @@ -25,7 +25,7 @@ import java.util.Set; public class BytesChronicleMap implements AbstractChronicleMap { - private final VanillaChronicleMap delegate; + final VanillaChronicleMap delegate; TcpReplicator.TcpSocketChannelEntryWriter output; public BytesChronicleMap(VanillaChronicleMap delegate) { diff --git a/src/main/java/net/openhft/chronicle/map/ChannelFactory.java b/src/main/java/net/openhft/chronicle/map/ChannelFactory.java new file mode 100644 index 000000000..5f326f4fd --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/ChannelFactory.java @@ -0,0 +1,26 @@ +/* + * Copyright 2014 Higher Frequency Trading + * + * http://www.higherfrequencytrading.com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.map; + +/** + * Created by Rob Austin + */ +public interface ChannelFactory { + void createChannel(short channelID); +} diff --git a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java index a175ab469..25bbdb1f5 100644 --- a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java +++ b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java @@ -48,12 +48,12 @@ import static net.openhft.chronicle.map.Replica.ModificationNotifier.NOP; /** * @author Rob Austin. */ -final class ChannelProvider implements Closeable { +public final class ChannelProvider implements Closeable { private static final Logger LOG = LoggerFactory.getLogger(ChannelProvider.class.getName()); static final Map implMapping = new IdentityHashMap<>(); - static synchronized ChannelProvider getProvider(ReplicationHub hub) throws IOException { + public static synchronized ChannelProvider getProvider(ReplicationHub hub) throws IOException { ChannelProvider channelProvider = implMapping.get(hub); if (channelProvider != null) return channelProvider; @@ -411,6 +411,10 @@ final class ChannelProvider implements Closeable { replicators.add(replicator); } + public List chronicleChannelList() { + return Collections.unmodifiableList(chronicleChannelList); + } + private interface MessageHandler { void onMessage(Bytes bytes); } diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredChronicleMapStatelessClientBuilder.java b/src/main/java/net/openhft/chronicle/map/ClientWiredChronicleMapStatelessClientBuilder.java new file mode 100644 index 000000000..12481552a --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredChronicleMapStatelessClientBuilder.java @@ -0,0 +1,164 @@ +/* + * Copyright 2014 Higher Frequency Trading + * + * http://www.higherfrequencytrading.com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.map; + +import net.openhft.chronicle.hash.ChronicleHashStatelessClientBuilder; +import net.openhft.chronicle.map.ChronicleMap;import net.openhft.chronicle.map.ClientWiredStatelessChronicleMap;import net.openhft.chronicle.map.ClientWiredStatelessClientTcpConnectionHub;import net.openhft.chronicle.map.MapBuilder;import net.openhft.lang.MemoryUnit; + +import java.io.IOException; +import java.lang.Class;import java.lang.IllegalStateException;import java.lang.Override;import java.lang.String;import java.net.InetSocketAddress; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * @author Rob Austin. + */ + public final class ClientWiredChronicleMapStatelessClientBuilder implements + ChronicleHashStatelessClientBuilder, + ChronicleMap>, + MapBuilder> { + + ClientWiredStatelessClientTcpConnectionHub hub; + private Class keyClass; + private Class valueClass; + private byte localIdentifier; + private short channelID; + private boolean doHandShaking; + + public ClientWiredChronicleMapStatelessClientBuilder(InetSocketAddress remoteAddress, Class keyClass, Class valueClass, short channelID) { + this.keyClass = keyClass; + this.valueClass = valueClass; + this.remoteAddress = remoteAddress; + this.channelID = channelID; + } + + public ClientWiredChronicleMapStatelessClientBuilder(ClientWiredStatelessClientTcpConnectionHub hub, Class keyClass, Class valueClass, short channelID) { + this.keyClass = keyClass; + this.valueClass = valueClass; + this.hub = hub; + this.channelID = channelID; + } + + // This method is meaningful, despite you could call of().create(), because Java 7 doesn't + // infer type parameters if case of chained calls, like this; and then you couldn't omit class + // name. Then it is + // ChronicleMapStatelessClientBuilder..of(addr).create() + // vs + // createClientOf(addr) // static import + // -- the second is really shorter. + + + private InetSocketAddress remoteAddress; + private boolean putReturnsNull = false; + private boolean removeReturnsNull = false; + private long timeoutMs = TimeUnit.SECONDS.toMillis(10); + private String name; + private int tcpBufferSize = (int) MemoryUnit.KILOBYTES.toBytes(64); + + private final AtomicBoolean used = new AtomicBoolean(false); + + + InetSocketAddress remoteAddress() { + return remoteAddress; + } + + @Override + public ClientWiredChronicleMapStatelessClientBuilder putReturnsNull(boolean putReturnsNull) { + this.putReturnsNull = putReturnsNull; + return this; + } + + boolean putReturnsNull() { + return putReturnsNull; + } + + @Override + public ClientWiredChronicleMapStatelessClientBuilder removeReturnsNull(boolean removeReturnsNull) { + this.removeReturnsNull = removeReturnsNull; + return this; + } + + boolean removeReturnsNull() { + return removeReturnsNull; + } + + @Override + public ClientWiredChronicleMapStatelessClientBuilder timeout(long timeout, TimeUnit units) { + this.timeoutMs = units.toMillis(timeout); + return this; + } + + long timeoutMs() { + return timeoutMs; + } + + @Override + public ClientWiredChronicleMapStatelessClientBuilder name(String name) { + this.name = name; + return this; + } + + String name() { + return name; + } + + @Override + public ClientWiredChronicleMapStatelessClientBuilder tcpBufferSize(int tcpBufferSize) { + this.tcpBufferSize = tcpBufferSize; + return this; + } + + int tcpBufferSize() { + return tcpBufferSize; + } + + @Override + public ChronicleMap create() throws IOException { + + // todo clean this up + if (hub == null) + hub = new ClientWiredStatelessClientTcpConnectionHub(this, localIdentifier, doHandShaking); + + if (!used.getAndSet(true)) { + return new ClientWiredStatelessChronicleMap(this, keyClass, valueClass, channelID); + + } else { + throw new IllegalStateException( + "A stateless client has already been created using this builder. " + + "Create a new ChronicleMapStatelessClientBuilder " + + "to create a new stateless client"); + } + } + + public void identifier(byte identifier) { + this.localIdentifier = identifier; + } + + public short channelID() { + return channelID; + } + + + public ClientWiredStatelessClientTcpConnectionHub hub() { + return hub; + } + + + +} diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java new file mode 100644 index 000000000..643b0e30d --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java @@ -0,0 +1,943 @@ +/* + * Copyright 2014 Higher Frequency Trading + * + * http://www.higherfrequencytrading.com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.map; + +import net.openhft.chronicle.hash.function.SerializableFunction; +import net.openhft.chronicle.map.MapWireHandlerBuilder.Fields; +import net.openhft.chronicle.wire.Marshallable; +import net.openhft.chronicle.wire.Wire; +import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.IOException; +import java.util.*; +import java.util.function.Consumer; +import java.util.function.Predicate; + +import static java.util.Collections.emptyList; +import static net.openhft.chronicle.map.ClientWiredStatelessChronicleMap.EventId.*; + + +/** + * @author Rob Austin. + */ +class ClientWiredStatelessChronicleMap implements ChronicleMap, Cloneable, ChannelFactory { + + private static final Logger LOG = LoggerFactory.getLogger(ClientWiredStatelessChronicleMap.class); + + private final ClientWiredStatelessClientTcpConnectionHub hub; + + protected Class kClass; + protected Class vClass; + + private boolean putReturnsNull; + private boolean removeReturnsNull; + private short channelID; + + public ClientWiredStatelessChronicleMap(@NotNull final ClientWiredChronicleMapStatelessClientBuilder config, + @NotNull final Class kClass, + @NotNull final Class vClass, + short channelID) { + this.channelID = channelID; + hub = config.hub; + this.putReturnsNull = config.putReturnsNull(); + this.removeReturnsNull = config.removeReturnsNull(); + this.kClass = kClass; + this.vClass = vClass; + } + + + @SuppressWarnings("UnusedDeclaration") + void identifier(int localIdentifier) { + hub.localIdentifier = localIdentifier; + } + + @Override + public void getAll(File toFile) throws IOException { + JsonSerializer.getAll(toFile, this, emptyList()); + } + + @Override + public void putAll(File fromFile) throws IOException { + JsonSerializer.putAll(fromFile, this, emptyList()); + } + + @Override + public V newValueInstance() { + return VanillaChronicleMap.newInstance(vClass, false); + } + + @Override + public K newKeyInstance() { + return VanillaChronicleMap.newInstance(kClass, true); + } + + @Override + public Class keyClass() { + return kClass; + } + + @Override + public boolean forEachEntryWhile(Predicate> predicate) { + // TODO implement! + throw new UnsupportedOperationException(); + } + + @Override + public void forEachEntry(Consumer> action) { + // TODO implement! + throw new UnsupportedOperationException(); + } + + + public String serverApplicationVersion() { + return hub.serverApplicationVersion(channelID); + } + + @Override + public void close() { + // todo add ref count + } + + @Override + public Class valueClass() { + return vClass; + } + + + @NotNull + public File file() { + throw new UnsupportedOperationException(); + } + + @SuppressWarnings("NullableProblems") + public V putIfAbsent(K key, V value) { + + if (key == null || value == null) + throw new NullPointerException(); + + return proxyReturnObject(PUT_IF_ABSENT.toString(), key, value, vClass); + } + + @SuppressWarnings("NullableProblems") + public boolean remove(Object key, Object value) { + + if (key == null) + throw new NullPointerException(); + + return value != null && proxyReturnBoolean(REMOVE_WITH_VALUE.toString(), (K) key, (V) value); + + } + + @SuppressWarnings("NullableProblems") + public boolean replace(K key, V oldValue, V newValue) { + if (key == null || oldValue == null || newValue == null) + throw new NullPointerException(); + return proxyReturnBoolean(REPLACE_WITH_OLD_AND_NEW_VALUE.toString(), key, oldValue, newValue); + } + + @SuppressWarnings("NullableProblems") + public V replace(K key, V value) { + if (key == null || value == null) + throw new NullPointerException(); + return proxyReturnObject(REPLACE.toString(), key, value, vClass); + } + + public int size() { + return (int) longSize(); + } + + /** + * calling this method should be avoided at all cost, as the entire {@code object} is + * serialized. This equals can be used to compare map that extends ChronicleMap. So two + * Chronicle Maps that contain the same data are considered equal, even if the instances of the + * chronicle maps were of different types + * + * @param object the object that you are comparing against + * @return true if the contain the same data + */ + @Override + public boolean equals(@Nullable Object object) { + if (this == object) return true; + if (object == null || object.getClass().isAssignableFrom(Map.class)) + return false; + + final Map that = (Map) object; + + final int size = size(); + + if (that.size() != size) + return false; + + final Set> entries = entrySet(); + return that.entrySet().equals(entries); + } + + @Override + public int hashCode() { + return proxyReturnInt(HASH_CODE.toString()); + } + + @NotNull + public String toString() { + return hub.proxyReturnString(TO_STRING.toString(), channelID); + } + + @NotNull + public String serverPersistedDataVersion() { + return hub.proxyReturnString(PERSISTED_DATA_VERSION.toString(), channelID); + } + + public boolean isEmpty() { + return proxyReturnBoolean(IS_EMPTY.toString()); + } + + public boolean containsKey(Object key) { + return proxyReturnBooleanK(CONTAINS_KEY.toString(), (K) key); + } + + @NotNull + private NullPointerException keyNotNullNPE() { + return new NullPointerException("key can not be null"); + } + + public boolean containsValue(Object value) { + return proxyReturnBooleanV(CONTAINS_VALUE.toString(), (V) value); + } + + public long longSize() { + return proxyReturnLong(LONG_SIZE.toString()); + } + + @Override + public MapKeyContext context(K key) { + throw new UnsupportedOperationException("Contexts are not supported by stateless clients"); + } + + public V get(Object key) { + return proxyReturnObject(vClass, GET.toString(), (K) key); + } + + @Nullable + public V getUsing(K key, V usingValue) { + throw new UnsupportedOperationException(); + } + + @NotNull + public V acquireUsing(@NotNull K key, V usingValue) { + throw new UnsupportedOperationException( + "acquireUsing() is not supported for stateless clients"); + } + + @NotNull + @Override + public MapKeyContext acquireContext(@NotNull K key, @NotNull V usingValue) { + throw new UnsupportedOperationException("Contexts are not supported by stateless clients"); + } + + public V remove(Object key) { + if (key == null) + throw keyNotNullNPE(); + return proxyReturnObject(vClass, removeReturnsNull ? REMOVE_WITHOUT_ACC.toString() : REMOVE.toString(), (K) key); + } + + @Override + public void createChannel(short channelID) { + proxyReturnVoid( + () -> { + writeField(Fields.METHOD_NAME, "CREATE_CHANNEL"); + hub.outWire().write(Fields.ARG_1).int16(channelID); + }); + } + + public void putAll(@NotNull Map map) { + + final long startTime = System.currentTimeMillis(); + long transactionId = hub.nextUniqueTransaction(startTime); + + + Set> entries = map.entrySet(); + Iterator> iterator = entries.iterator(); + + boolean hasNext; + + do { + + hasNext = iterator.hasNext(); + + hub.outBytesLock().lock(); + try { + assert hub.outBytesLock().isHeldByCurrentThread(); + assert !hub.inBytesLock().isHeldByCurrentThread(); + + Wire wire = hub.outWire(); + + assert hub.outBytesLock().isHeldByCurrentThread(); + hub.markSize(wire); + hub.startTime(startTime); + + wire.write(Fields.TRANSACTION_ID).int64(transactionId); + wire.write(Fields.TIME_STAMP).int64(startTime); + wire.write(Fields.CHANNEL_ID).int16(channelID); + + writeField(Fields.METHOD_NAME, "PUT_ALL"); + + hub.outWire().write(Fields.HAS_NEXT).bool(hasNext); + + if (hasNext) { + Map.Entry e = iterator.next(); + writeField(Fields.ARG_1, e.getKey()); + writeField(Fields.ARG_2, e.getValue()); + } + + hub.writeSocket(hub.outWire()); + + } finally { + hub.outBytesLock().unlock(); + } + + } while (hasNext); + + // wait for the transaction id to be received, this will only be received once the + // last chunk has been processed + readVoid(transactionId, startTime); + } + + + public V put(K key, V value) { + if (key == null || value == null) + throw new NullPointerException(); + return proxyReturnObject(putReturnsNull ? PUT_WITHOUT_ACC.toString() : PUT.toString(), key, value, vClass); + } + + @Nullable + public R getMapped(@Nullable K key, @NotNull SerializableFunction function) { + throw new UnsupportedOperationException(); + } + + @Nullable + @Override + public V putMapped(@Nullable K key, @NotNull UnaryOperator unaryOperator) { + throw new UnsupportedOperationException(); + } + + public void clear() { + proxyReturnVoid(() -> writeField(Fields.METHOD_NAME, CLEAR.toString())); + } + + + public Set keySet() { + final long startTime = System.currentTimeMillis(); + final Set result = new HashSet(); + // send + final long transactionId = proxySend("KEY_SET", startTime); + assert !hub.outBytesLock().isHeldByCurrentThread(); + final long timeoutTime = startTime + hub.timeoutMs; + + for (; ; ) { + + // receive + hub.inBytesLock().lock(); + try { + final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); + + + if (wireIn.read(Fields.HAS_NEXT).bool()) { + K k = readK(Fields.RESULT, wireIn, null); + result.add(k); + } else + break; + + } finally { + hub.inBytesLock().unlock(); + } + } + return result; + } + + + public Collection values() { + final long startTime = System.currentTimeMillis(); + final Set result = new HashSet(); + // send + final long transactionId = proxySend("VALUES", startTime); + assert !hub.outBytesLock().isHeldByCurrentThread(); + final long timeoutTime = startTime + hub.timeoutMs; + + for (; ; ) { + + // receive + hub.inBytesLock().lock(); + try { + final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); + + + if (wireIn.read(Fields.HAS_NEXT).bool()) { + V v = readV(Fields.RESULT, wireIn, null); + result.add(v); + } else + break; + + } finally { + hub.inBytesLock().unlock(); + } + } + return result; + } + + @NotNull + public Set> entrySet() { + final long startTime = System.currentTimeMillis(); + final Map result = new HashMap(); + + // send + final long transactionId = proxySend("ENTRY_SET", startTime); + assert !hub.outBytesLock().isHeldByCurrentThread(); + final long timeoutTime = startTime + hub.timeoutMs; + + for (; ; ) { + + // receive + hub.inBytesLock().lock(); + try { + final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); + + if (wireIn.read(Fields.HAS_NEXT).bool()) { + result.put( + readK(Fields.RESULT_KEY, wireIn, null), + readV(Fields.RESULT_VALUE, wireIn, null)); + } else + break; + + } finally { + hub.inBytesLock().unlock(); + } + } + return result.entrySet(); + } + + /** + * @param callback each entry is passed to the callback + */ + void entrySet(@NotNull MapEntryCallback callback) { + throw new UnsupportedOperationException(); + } + + + private long readLong(long transactionId, long startTime) { + assert !hub.outBytesLock().isHeldByCurrentThread(); + final long timeoutTime = startTime + hub.timeoutMs; + + // receive + hub.inBytesLock().lock(); + try { + return hub.proxyReply(timeoutTime, transactionId).read(Fields.RESULT).int64(); + } finally { + hub.inBytesLock().unlock(); + } + } + + + private void writeField(Fields fieldName, Object value) { + writeField(fieldName, value, hub.outWire()); + } + + private void writeField(Fields fieldName, Object value, Wire wire) { + + assert hub.outBytesLock().isHeldByCurrentThread(); + assert !hub.inBytesLock().isHeldByCurrentThread(); + + if (value instanceof Byte) + wire.write(fieldName).int8((Byte) value); + else if (value instanceof Character) + wire.write(fieldName).text(((Character) value).toString()); + else if (value instanceof Short) + wire.write(fieldName).int16((Short) value); + else if (value instanceof Integer) + wire.write(fieldName).int32((Integer) value); + else if (value instanceof Long) + wire.write(fieldName).int64((Long) value); + else if (value instanceof CharSequence) { + wire.write(fieldName).text((CharSequence) value); + } else if (value instanceof Marshallable) { + wire.write(fieldName).marshallable((Marshallable) value); + } else { + throw new IllegalStateException("type=" + value.getClass() + " is unsupported, it must either be of type Marshallable or CharSequence"); + } + } + + + private V readValue(Fields argName, long transactionId, long startTime, final V usingValue) { + assert !hub.outBytesLock().isHeldByCurrentThread(); + long timeoutTime = startTime + hub.timeoutMs; + + hub.inBytesLock().lock(); + try { + + final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); + + if (wireIn.read(Fields.RESULT_IS_NULL).bool()) + return null; + + return readV(argName, wireIn, usingValue); + + } finally { + hub.inBytesLock().unlock(); + } + } + + private V readV(Fields argName, Wire wireIn, V usingValue) { + if (StringBuilder.class.isAssignableFrom(vClass)) { + wireIn.read(argName).text((StringBuilder) usingValue); + return usingValue; + } else if (Marshallable.class.isAssignableFrom(vClass)) { + + if (usingValue == null) + try { + V v = vClass.newInstance(); + wireIn.read(argName).marshallable((Marshallable) v); + return v; + } catch (Exception e) { + throw new RuntimeException(e); + } + + wireIn.read(argName).marshallable((Marshallable) usingValue); + return usingValue; + + } else if (String.class.isAssignableFrom(vClass)) { + //noinspection unchecked + return (V) wireIn.read(argName).text(); + + } else if (Long.class.isAssignableFrom(vClass)) { + //noinspection unchecked + return (V) (Long) wireIn.read(argName).int64(); + } else if (Double.class.isAssignableFrom(vClass)) { + //noinspection unchecked + return (V) (Double) wireIn.read(argName).float64(); + + } else if (Integer.class.isAssignableFrom(vClass)) { + //noinspection unchecked + return (V) (Integer) wireIn.read(argName).int32(); + + } else if (Float.class.isAssignableFrom(vClass)) { + //noinspection unchecked + return (V) (Float) wireIn.read(argName).float32(); + + } else if (Short.class.isAssignableFrom(vClass)) { + //noinspection unchecked + return (V) (Short) wireIn.read(argName).int16(); + + } else if (Character.class.isAssignableFrom(vClass)) { + //noinspection unchecked + final String text = wireIn.read(argName).text(); + if (text == null || text.length() == 0) + return null; + return (V) (Character) text.charAt(0); + + } else if (Byte.class.isAssignableFrom(vClass)) { + //noinspection unchecked + return (V) (Byte) wireIn.read(argName).int8(); + + + } else { + throw new IllegalStateException("unsupported type"); + } + } + + private K readK(Fields argName, Wire wireIn, K usingValue) { + if (StringBuilder.class.isAssignableFrom(kClass)) { + wireIn.read(argName).text((StringBuilder) usingValue); + return usingValue; + } else if (Marshallable.class.isAssignableFrom(kClass)) { + + if (usingValue == null) + try { + K v = kClass.newInstance(); + wireIn.read(argName).marshallable((Marshallable) v); + return v; + } catch (Exception e) { + throw new RuntimeException(e); + } + + wireIn.read(argName).marshallable((Marshallable) usingValue); + return usingValue; + + } else if (String.class.isAssignableFrom(vClass)) { + //noinspection unchecked + return (K) wireIn.read(argName).text(); + + } else { + throw new IllegalStateException("unsupported type"); + } + } + + private boolean readBoolean(long transactionId, long startTime) { + assert !hub.outBytesLock().isHeldByCurrentThread(); + + long timeoutTime = startTime + hub.timeoutMs; + + // receive + hub.inBytesLock().lock(); + try { + final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); + return wireIn.read(Fields.RESULT).bool(); + } finally { + hub.inBytesLock().unlock(); + } + } + + private int readInt(long transactionId, long startTime) { + assert !hub.outBytesLock().isHeldByCurrentThread(); + + long timeoutTime = startTime + hub.timeoutMs; + + // receive + hub.inBytesLock().lock(); + try { + final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); + return wireIn.read(Fields.RESULT).int32(); + } finally { + hub.inBytesLock().unlock(); + } + } + + @SuppressWarnings("SameParameterValue") + private boolean proxyReturnBoolean(@NotNull final String methodName, K key, V value) { + final long startTime = System.currentTimeMillis(); + + long transactionId; + + hub.outBytesLock().lock(); + try { + transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); + writeField(Fields.METHOD_NAME, methodName); + writeField(Fields.ARG_1, key); + writeField(Fields.ARG_2, value); + + hub.writeSocket(hub.outWire()); + } finally { + hub.outBytesLock().unlock(); + } + + return readBoolean(transactionId, startTime); + + } + + @SuppressWarnings("SameParameterValue") + private boolean proxyReturnBoolean(@NotNull final String methodName, K key, V value1, V value2) { + final long startTime = System.currentTimeMillis(); + + long transactionId; + hub.outBytesLock().lock(); + try { + transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); + writeField(Fields.METHOD_NAME, methodName); + writeField(Fields.ARG_1, value1); + writeField(Fields.ARG_2, value2); + hub.writeSocket(hub.outWire()); + } finally { + hub.outBytesLock().unlock(); + } + + return readBoolean(transactionId, startTime); + + } + + @SuppressWarnings("SameParameterValue") + private boolean proxyReturnBooleanV(@NotNull final String methodName, V value) { + final long startTime = System.currentTimeMillis(); + return readBoolean(proxySend(methodName, startTime, value), startTime); + } + + @SuppressWarnings("SameParameterValue") + private boolean proxyReturnBooleanK(@NotNull final String methodName, K key) { + final long startTime = System.currentTimeMillis(); + + return readBoolean(proxySend(methodName, startTime, key), startTime); + } + + @SuppressWarnings("SameParameterValue") + private long proxyReturnLong(@NotNull final String methodName) { + final long startTime = System.currentTimeMillis(); + + long transactionId; + + hub.outBytesLock().lock(); + try { + transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); + writeField(Fields.METHOD_NAME, methodName); + hub.writeSocket(hub.outWire()); + } finally { + hub.outBytesLock().unlock(); + } + + return readLong(transactionId, startTime); + } + + @SuppressWarnings("SameParameterValue") + private boolean proxyReturnBoolean(@NotNull final String methodName) { + final long startTime = System.currentTimeMillis(); + + long transactionId; + + hub.outBytesLock().lock(); + try { + transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); + writeField(Fields.METHOD_NAME, methodName); + hub.writeSocket(hub.outWire()); + } finally { + hub.outBytesLock().unlock(); + } + + return readBoolean(transactionId, startTime); + } + + @SuppressWarnings("SameParameterValue") + private void proxyReturnVoid(Runnable r) { + + + final long startTime = System.currentTimeMillis(); + long transactionId; + + hub.outBytesLock().lock(); + try { + transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); + r.run(); + hub.writeSocket(hub.outWire()); + } finally { + hub.outBytesLock().unlock(); + } + + readVoid(transactionId, startTime); + } + + + @SuppressWarnings("SameParameterValue") + private int proxyReturnInt(@NotNull final String methodName) { + final long startTime = System.currentTimeMillis(); + + long transactionId; + + hub.outBytesLock().lock(); + try { + transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); + writeField(Fields.METHOD_NAME, methodName); + hub.writeSocket(hub.outWire()); + } finally { + hub.outBytesLock().unlock(); + } + + return readInt(transactionId, startTime); + } + + + @Nullable + private R proxyReturnObject(@NotNull final String methodName, K key, V value, Class resultType) { + + final long startTime = System.currentTimeMillis(); + long transactionId; + + hub.outBytesLock().lock(); + try { + assert hub.outBytesLock().isHeldByCurrentThread(); + assert !hub.inBytesLock().isHeldByCurrentThread(); + + transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); + writeField(Fields.METHOD_NAME, methodName); + writeField(Fields.ARG_1, key); + writeField(Fields.ARG_2, value); + hub.writeSocket(hub.outWire()); + + } finally { + hub.outBytesLock().unlock(); + } + + if (eventReturnsNull(methodName)) + return null; + + if (resultType == vClass) + return (R) readValue(Fields.RESULT, transactionId, startTime, null); + + else + throw new UnsupportedOperationException("class of type class=" + resultType + " is not " + + "supported"); + } + + @Nullable + private R proxyReturnObject(Class rClass, @NotNull final String methodName, Object key) { + final long startTime = System.currentTimeMillis(); + + long transactionId; + + transactionId = proxySend(methodName, startTime, key); + + if (eventReturnsNull(methodName)) + return null; + + if (rClass == vClass) + return (R) readValue(Fields.RESULT, transactionId, startTime, null); + else + throw new UnsupportedOperationException("class of type class=" + rClass + " is not " + + "supported"); + } + + private long proxySend(String methodName, long startTime, Object arg1) { + long transactionId; + hub.outBytesLock().lock(); + try { + transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); + writeField(Fields.METHOD_NAME, methodName); + writeField(Fields.ARG_1, arg1); + hub.writeSocket(hub.outWire()); + } finally { + hub.outBytesLock().unlock(); + } + return transactionId; + } + + + private long proxySend(String methodName, long startTime) { + long transactionId; + hub.outBytesLock().lock(); + try { + transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); + writeField(Fields.METHOD_NAME, methodName); + hub.writeSocket(hub.outWire()); + } finally { + hub.outBytesLock().unlock(); + } + return transactionId; + } + + private boolean eventReturnsNull(@NotNull String methodName) { + + switch (methodName) { + case "PUT_ALL_WITHOUT_ACC": + case "PUT_WITHOUT_ACC": + case "REMOVE_WITHOUT_ACC": + return true; + default: + return false; + } + + } + + private void readVoid(long transactionId, long startTime) { + long timeoutTime = startTime + hub.timeoutMs; + + // receive + hub.inBytesLock().lock(); + try { + hub.proxyReply(timeoutTime, transactionId); + } finally { + hub.inBytesLock().unlock(); + } + } + + + static enum EventId { + HEARTBEAT, + STATEFUL_UPDATE, + LONG_SIZE, + SIZE, + IS_EMPTY, + CONTAINS_KEY, + CONTAINS_VALUE, + GET, + PUT, + PUT_WITHOUT_ACC, + REMOVE, + REMOVE_WITHOUT_ACC, + CLEAR, + KEY_SET, + VALUES, + ENTRY_SET, + REPLACE, + REPLACE_WITH_OLD_AND_NEW_VALUE, + PUT_IF_ABSENT, + REMOVE_WITH_VALUE, + TO_STRING, + APPLICATION_VERSION, + PERSISTED_DATA_VERSION, + PUT_ALL, + PUT_ALL_WITHOUT_ACC, + HASH_CODE, + MAP_FOR_KEY, + PUT_MAPPED, + KEY_BUILDER, + VALUE_BUILDER + } + + class Entry implements Map.Entry { + + final K key; + final V value; + + /** + * Creates new entry. + */ + Entry(K k1, V v) { + value = v; + key = k1; + } + + public final K getKey() { + return key; + } + + public final V getValue() { + return value; + } + + public final V setValue(V newValue) { + V oldValue = value; + ClientWiredStatelessChronicleMap.this.put(getKey(), newValue); + return oldValue; + } + + public final boolean equals(Object o) { + if (!(o instanceof Map.Entry)) + return false; + final Map.Entry e = (Map.Entry) o; + final Object k1 = getKey(); + final Object k2 = e.getKey(); + if (k1 == k2 || (k1 != null && k1.equals(k2))) { + Object v1 = getValue(); + Object v2 = e.getValue(); + if (v1 == v2 || (v1 != null && v1.equals(v2))) + return true; + } + return false; + } + + public final int hashCode() { + return (key == null ? 0 : key.hashCode()) ^ + (value == null ? 0 : value.hashCode()); + } + + @NotNull + public final String toString() { + return getKey() + "=" + getValue(); + } + } +} + diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessClientTcpConnectionHub.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessClientTcpConnectionHub.java new file mode 100644 index 000000000..f6875e91e --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessClientTcpConnectionHub.java @@ -0,0 +1,648 @@ +/* + * Copyright 2014 Higher Frequency Trading + * + * http://www.higherfrequencytrading.com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.map; + +import com.sun.jdi.connect.spi.ClosedConnectionException; +import net.openhft.chronicle.bytes.Bytes; +import net.openhft.chronicle.hash.RemoteCallTimeoutException; +import net.openhft.chronicle.hash.impl.util.BuildVersion; +import net.openhft.chronicle.hash.impl.util.CloseablesManager; +import net.openhft.chronicle.network2.event.EventGroup; +import net.openhft.chronicle.wire.TextWire; +import net.openhft.chronicle.wire.Wire; +import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.nio.ByteBuffer; +import java.nio.channels.SocketChannel; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.locks.ReentrantLock; + +import static net.openhft.chronicle.map.ClientWiredStatelessChronicleMap.EventId.APPLICATION_VERSION; +import static net.openhft.chronicle.map.MapWireHandler.SIZE_OF_SIZE; +import static net.openhft.chronicle.map.MapWireHandlerBuilder.Fields.*; + +/** + * Created by Rob Austin + */ +public class ClientWiredStatelessClientTcpConnectionHub { + + private static final Logger LOG = LoggerFactory.getLogger(StatelessChronicleMap.class); + + protected final String name; + protected final InetSocketAddress remoteAddress; + protected final long timeoutMs; + protected final int tcpBufferSize; + private final ReentrantLock inBytesLock = new ReentrantLock(true); + private final ReentrantLock outBytesLock = new ReentrantLock(); + + + @NotNull + private final AtomicLong transactionID = new AtomicLong(0); + private final ClientWiredChronicleMapStatelessClientBuilder config; + @Nullable + protected CloseablesManager closeables; + + private final Wire outWire = new TextWire(Bytes.elasticByteBuffer()); + + long largestChunkSoFar = 0; + private final Wire intWire = new TextWire(Bytes.elasticByteBuffer()); + + // used by the enterprise version + protected int localIdentifier; + + + private SocketChannel clientChannel; + // this is a transaction id and size that has been read by another thread. + private volatile long parkedTransactionId; + + private volatile long parkedTransactionTimeStamp; + private long limitOfLast = 0; + + // set up in the header + private long startTime; + private boolean doHandShaking; + + + public ClientWiredStatelessClientTcpConnectionHub(ClientWiredChronicleMapStatelessClientBuilder config, byte localIdentifier, boolean doHandShaking) { + this.localIdentifier = localIdentifier; + this.doHandShaking = doHandShaking; + this.remoteAddress = config.remoteAddress(); + this.tcpBufferSize = config.tcpBufferSize(); + this.config = config; + this.name = config.name(); + this.timeoutMs = config.timeoutMs(); + attemptConnect(remoteAddress); + checkVersion(config.channelID()); + } + + private synchronized void attemptConnect(final InetSocketAddress remoteAddress) { + + // ensures that the excising connection are closed + closeExisting(); + + try { + SocketChannel socketChannel = AbstractChannelReplicator.openSocketChannel(closeables); + + if (socketChannel.connect(remoteAddress)) { + + clientChannel = socketChannel; + } + + } catch (IOException e) { + if (closeables != null) closeables.closeQuietly(); + clientChannel = null; + } + } + + ReentrantLock inBytesLock() { + return inBytesLock; + } + + ReentrantLock outBytesLock() { + return outBytesLock; + } + + protected void checkVersion(short channelID) { + + final String serverVersion = serverApplicationVersion(channelID); + final String clientVersion = clientVersion(); + + if (!serverVersion.equals(clientVersion)) { + LOG.warn("DIFFERENT CHRONICLE-MAP VERSIONS: The Chronicle-Map-Server and " + + "Stateless-Client are on different " + + "versions, " + + " we suggest that you use the same version, server=" + serverApplicationVersion(channelID) + ", " + + "client=" + clientVersion); + } + } + + + private void checkTimeout(long timeoutTime) { + if (timeoutTime < System.currentTimeMillis()) + throw new RemoteCallTimeoutException(); + } + + protected synchronized void lazyConnect(final long timeoutMs, + final InetSocketAddress remoteAddress) { + if (clientChannel != null) + return; + + if (LOG.isDebugEnabled()) + LOG.debug("attempting to connect to " + remoteAddress + " ,name=" + name); + + SocketChannel result; + + long timeoutAt = System.currentTimeMillis() + timeoutMs; + + for (; ; ) { + checkTimeout(timeoutAt); + + // ensures that the excising connection are closed + closeExisting(); + + try { + result = AbstractChannelReplicator.openSocketChannel(closeables); + if (!result.connect(remoteAddress)) { + try { + Thread.sleep(100); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + continue; + } + + result.socket().setTcpNoDelay(true); + if (doHandShaking) + + break; + } catch (IOException e) { + if (closeables != null) closeables.closeQuietly(); + } catch (Exception e) { + if (closeables != null) closeables.closeQuietly(); + throw e; + } + } + clientChannel = result; + + checkVersion(config.channelID()); + + } + + + /** + * closes the existing connections and establishes a new closeables + */ + protected void closeExisting() { + // ensure that any excising connection are first closed + if (closeables != null) + closeables.closeQuietly(); + + closeables = new CloseablesManager(); + } + + + + public synchronized void close() { + + if (closeables != null) + closeables.closeQuietly(); + + closeables = null; + clientChannel = null; + + } + + /** + * the transaction id are generated as unique timestamps + * + * @param time in milliseconds + * @return a unique transactionId + */ + long nextUniqueTransaction(long time) { + long id = time; + for (; ; ) { + long old = transactionID.get(); + if (old >= id) id = old + 1; + if (transactionID.compareAndSet(old, id)) + break; + } + return id; + } + + @NotNull + public String serverApplicationVersion(short channelID) { + TextWire wire = new TextWire(Bytes.elasticByteBuffer()); + String result = proxyReturnString(APPLICATION_VERSION.toString(), channelID, wire); + return (result == null) ? "" : result; + } + + @SuppressWarnings("WeakerAccess") + @NotNull + String clientVersion() { + return BuildVersion.version(); + } + + + /** + * sends data to the server via TCP/IP + * + * @param wire the {@code wire} containing the outbound data + */ + void writeSocket(@NotNull final Wire wire) { + + assert outBytesLock().isHeldByCurrentThread(); + assert !inBytesLock().isHeldByCurrentThread(); + + + final long timeoutTime = startTime + this.timeoutMs; + try { + + for (; ; ) { + if (clientChannel == null) + lazyConnect(timeoutMs, remoteAddress); + try { + writeLength(wire); + + // send out all the bytes + writeSocket(wire, timeoutTime); + break; + + } catch (@NotNull java.nio.channels.ClosedChannelException | ClosedConnectionException e) { + checkTimeout(timeoutTime); + lazyConnect(timeoutMs, remoteAddress); + } + } + } catch (IOException e) { + close(); + throw new IORuntimeException(e); + } catch (Exception e) { + close(); + throw e; + } + } + + private void writeLength(Wire outWire) { + assert outBytesLock().isHeldByCurrentThread(); + long position = outWire.bytes().position(); + if (position > Integer.MAX_VALUE || position < Integer.MIN_VALUE) + throw new IllegalStateException("message too large"); + + long pos = outWire.bytes().position(); + try { + outWire.bytes().reset(); + int size = (int) (position - outWire.bytes().position()); + outWire.bytes().writeUnsignedShort(size - SIZE_OF_SIZE); + } finally { + outWire.bytes().position(pos); + } + } + + protected Wire proxyReply(long timeoutTime, final long transactionId) { + + assert inBytesLock().isHeldByCurrentThread(); + + try { + + final Wire wire = proxyReplyThrowable(timeoutTime, transactionId); + + // handle an exception if the message contains the IS_EXCEPTION field + if (wire.read(() -> "IS_EXCEPTION").bool()) { + final String text = wire.read(() -> "EXCEPTION").text(); + throw new RuntimeException(text); + } + return wire; + } catch (IOException e) { + close(); + throw new IORuntimeException(e); + } catch (RuntimeException e) { + close(); + throw e; + } catch (Exception e) { + close(); + throw new RuntimeException(e); + } catch (AssertionError e) { + LOG.error("name=" + name, e); + throw e; + } + } + + + private Wire proxyReplyThrowable(long timeoutTime, long transactionId) throws IOException { + + assert inBytesLock().isHeldByCurrentThread(); + + for (; ; ) { + + // read the next item from the socket + if (parkedTransactionId == 0) { + + assert parkedTransactionTimeStamp == 0; + + // if we have processed all the bytes that we have read in + if (inWireByteBuffer().position() == intWire.bytes().position()) + inWireClear(); + + // todo change the size to include the meta data bit + // reads just the size + readSocket(SIZE_OF_SIZE, timeoutTime); + + final int messageSize = intWire.bytes().readUnsignedShort(intWire.bytes().position()); + + + assert messageSize > 0 : "Invalid message size " + messageSize; + assert messageSize < 16 << 20 : "Invalid message size " + messageSize; + + final int remainingBytes0 = messageSize; + readSocket(remainingBytes0, timeoutTime); + + + intWire.bytes().skip(SIZE_OF_SIZE); + intWire.bytes().limit(intWire.bytes().position() + messageSize); + + System.out.println("\n--------------------------------\n" + + "client read:\n\n" + Bytes.toDebugString(intWire.bytes())); + + long transactionId0 = intWire.read(() -> "TRANSACTION_ID").int64(); + + + // check the transaction id is reasonable + /* assert transactionId0 > 1410000000000L * TcpReplicator.TIMESTAMP_FACTOR : + "TransactionId too small " + transactionId0 + " messageSize " + messageSize; + assert transactionId0 < 2100000000000L * TcpReplicator.TIMESTAMP_FACTOR : + "TransactionId too large " + transactionId0 + " messageSize " + messageSize; +*/ + // if the transaction id is for this thread process it + if (transactionId0 == transactionId) { + clearParked(); + return intWire; + + } else { + + // if the transaction id is not for this thread, park it + // and allow another thread to pick it up + parkedTransactionTimeStamp = System.currentTimeMillis(); + parkedTransactionId = transactionId0; + pause(); + continue; + } + } + + // the transaction id was read by another thread, but is for this thread, process it + if (parkedTransactionId == transactionId) { + clearParked(); + return intWire; + } + + // time out the old transaction id + if (System.currentTimeMillis() - timeoutTime > + parkedTransactionTimeStamp) { + + LOG.error("name=" + name, new IllegalStateException("Skipped Message with " + + "transaction-id=" + + parkedTransactionTimeStamp + + ", this can occur when you have another thread which has called the " + + "stateless client and terminated abruptly before the message has been " + + "returned from the server and hence consumed by the other thread.")); + + // read the the next message + clearParked(); + pause(); + } + + } + + } + + /** + * clears the wire and its underlying byte buffer + */ + private void inWireClear() { + inWireByteBuffer().clear(); + intWire.bytes().clear(); + } + + private void clearParked() { + assert inBytesLock().isHeldByCurrentThread(); + parkedTransactionId = 0; + parkedTransactionTimeStamp = 0; + } + + private void pause() { + + assert !outBytesLock().isHeldByCurrentThread(); + assert inBytesLock().isHeldByCurrentThread(); + + /// don't call inBytesLock.isHeldByCurrentThread() as it not atomic + inBytesLock().unlock(); + + // allows another thread to enter hear + inBytesLock().lock(); + } + + /** + * reads up to the number of byte in {@code requiredNumberOfBytes} from the socket + * + * @param requiredNumberOfBytes the number of bytes to read + * @param timeoutTime timeout in milliseconds + * @return bytes read from the TCP/IP socket + * @throws java.io.IOException socket failed to read data + */ + @SuppressWarnings("UnusedReturnValue") + private void readSocket(int requiredNumberOfBytes, long timeoutTime) throws IOException { + + assert inBytesLock().isHeldByCurrentThread(); + ByteBuffer buffer = inWireByteBuffer(); + + buffer.limit(buffer.position() + requiredNumberOfBytes); + long start = buffer.position(); + + while (buffer.position() - start < requiredNumberOfBytes) { + + int len = clientChannel.read(buffer); + + if (len == -1) + throw new IORuntimeException("Disconnection to server"); + + checkTimeout(timeoutTime); + } + } + + private ByteBuffer inWireByteBuffer() { + return (ByteBuffer) intWire.bytes().underlyingObject(); + } + + /** + * writes the bytes to the socket + * + * @param outWire the data that you wish to write + * @param timeoutTime how long before a we timeout + * @throws java.io.IOException + */ + private void writeSocket(Wire outWire, long timeoutTime) throws IOException { + + assert outBytesLock().isHeldByCurrentThread(); + assert !inBytesLock().isHeldByCurrentThread(); + + long outBytesPosition = outWire.bytes().position(); + + + // if we have other threads waiting to send and the buffer is not full, let the other threads + // write to the buffer + if (outBytesLock().hasQueuedThreads() && + outBytesPosition + largestChunkSoFar <= tcpBufferSize) { + return; + } + + final ByteBuffer outBuffer = (ByteBuffer) outWire.bytes().underlyingObject(); + outBuffer.limit((int) outWire.bytes().position()); + outBuffer.position(SIZE_OF_SIZE); + + if (EventGroup.IS_DEBUG ) { + System.out.println("\n--------------------------------------------\n" + + "client wrote:\n\n" + Bytes.toDebugString(outBuffer)); + } + + outBuffer.position(0); + + upateLargestChunkSoFarSize(outBuffer); + + while (outBuffer.remaining() > 0) { + + int len = clientChannel.write(outBuffer); + + if (len == -1) + throw new IORuntimeException("Disconnection to server"); + + + // if we have queued threads then we don't have to write all the bytes as the other + // threads will write the remains bytes. + if (outBuffer.remaining() > 0 && outBytesLock().hasQueuedThreads() && + outBuffer.remaining() + largestChunkSoFar <= tcpBufferSize) { + + if (LOG.isDebugEnabled()) + LOG.debug("continuing - without all the data being written to the buffer as " + + "it will be written by the next thread"); + outBuffer.compact(); + outWire.bytes().limit(outBuffer.limit()); + outWire.bytes().position(outBuffer.position()); + return; + } + + checkTimeout(timeoutTime); + + } + + outBuffer.clear(); + outWire.bytes().clear(); + + } + + /** + * calculates the size of each chunk + * + * @param outBuffer + */ + private void upateLargestChunkSoFarSize(ByteBuffer outBuffer) { + int sizeOfThisChunk = (int) (outBuffer.limit() - limitOfLast); + if (largestChunkSoFar < sizeOfThisChunk) + largestChunkSoFar = sizeOfThisChunk; + + limitOfLast = outBuffer.limit(); + } + + + private long proxySend(@NotNull final String methodName, + final long startTime, + short channelID, + @NotNull final Wire wire) { + + assert outBytesLock().isHeldByCurrentThread(); + assert !inBytesLock().isHeldByCurrentThread(); + + // send + outBytesLock().lock(); + try { + long transactionId = writeHeader(startTime, channelID, wire); + wire.write(() -> "METHOD_NAME").text(methodName); + writeSocket(wire); + return transactionId; + } finally { + outBytesLock().unlock(); + } + } + + @SuppressWarnings("SameParameterValue") + @Nullable + String proxyReturnString(@NotNull final String messageId, short channelID) { + return proxyReturnString(messageId, channelID, outWire); + } + + @SuppressWarnings("SameParameterValue") + @Nullable + String proxyReturnString(@NotNull final String messageId, short channelID, Wire outWire) { + final long startTime = System.currentTimeMillis(); + long transactionId; + + outBytesLock().lock(); + try { + transactionId = proxySend(messageId, startTime, channelID, outWire); + } finally { + outBytesLock().unlock(); + } + + long timeoutTime = startTime + this.timeoutMs; + + // receive + inBytesLock().lock(); + try { + return proxyReply(timeoutTime, transactionId).read(() -> "RESULT").text(); + } catch (Exception e) { + throw new RuntimeException(e); + } finally { + inBytesLock().unlock(); + } + } + + + Wire outWire() { + assert outBytesLock().isHeldByCurrentThread(); + return outWire; + } + + long writeHeader(long startTime, short channelID, Wire wire) { + + assert outBytesLock().isHeldByCurrentThread(); + markSize(wire); + startTime(startTime); + + long transactionId = nextUniqueTransaction(startTime); + wire.write(TYPE).text("MAP"); + wire.write(TRANSACTION_ID).int64(transactionId); + wire.write(TIME_STAMP).int64(startTime); + wire.write(CHANNEL_ID).int16(channelID); + + return transactionId; + } + + + /** + * mark the location of the outWire size + * + * @param outWire + */ + void markSize(Wire outWire) { + + assert outBytesLock().isHeldByCurrentThread(); + + // this is where the size will go + outWire.bytes().mark(); + + // skip the 2 bytes for the size + outWire.bytes().skip(2); + } + + void startTime(long startTime) { + this.startTime = startTime; + } +} diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java new file mode 100644 index 000000000..cfe97d5a9 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java @@ -0,0 +1,714 @@ +/* + * Copyright 2014 Higher Frequency Trading + * + * http://www.higherfrequencytrading.com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.map; + +/** + * Created by Rob Austin + */ + +import net.openhft.chronicle.bytes.Bytes; +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.network2.WireHandler; +import net.openhft.chronicle.network2.event.EventGroup; +import net.openhft.chronicle.wire.ValueIn; +import net.openhft.chronicle.wire.Wire; +import net.openhft.chronicle.wire.WireKey; +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.PrintWriter; +import java.io.StreamCorruptedException; +import java.io.StringWriter; +import java.nio.BufferOverflowException; +import java.nio.channels.SelectionKey; +import java.util.*; +import java.util.concurrent.Callable; +import java.util.function.Consumer; +import java.util.function.Function; +import java.util.function.Supplier; + +import static net.openhft.chronicle.map.MapWireHandlerBuilder.Fields.*; + +/** + * @author Rob Austin. + */ +class MapWireHandler implements WireHandler { + + private static final Logger LOG = LoggerFactory.getLogger(MapWireHandler.class); + public static final int SIZE_OF_SIZE = 2; + + private final ArrayList bytesChronicleMaps = new ArrayList<>(); + @NotNull + + private final StringBuilder methodName = new StringBuilder(); + private final Supplier>> chronicleHashInstanceBuilder; + private Wire inWire = null; + private Wire outWire = null; + private final Consumer writeElement = new Consumer>() { + + @Override + public void accept(Iterator iterator) { + outWire.write(RESULT); + outWire.bytes().write(iterator.next()); + } + }; + private final Consumer writeEntry = new Consumer>>() { + + @Override + public void accept(Iterator> iterator) { + + final Map.Entry entry = iterator.next(); + + outWire.write(RESULT_KEY); + outWire.bytes().write(entry.getKey()); + + outWire.write(RESULT_VALUE); + outWire.bytes().write(entry.getValue()); + } + }; + private byte localIdentifier; + private long timestamp; + private short channelId; + private List channelList; + private ReplicationHub hub; + private byte remoteIdentifier; + private Runnable out = null; + private SelectionKey key = null; + private Map incompleteWork = new HashMap<>(); + + public MapWireHandler(@NotNull final Supplier>> chronicleHashInstanceBuilder, + @NotNull final ReplicationHub hub, + byte localIdentifier, + @NotNull final List channelList) { + this(chronicleHashInstanceBuilder, hub); + this.channelList = channelList; + this.localIdentifier = localIdentifier; + } + + public MapWireHandler(Supplier>> chronicleHashInstanceBuilder, ReplicationHub hub) { + this.chronicleHashInstanceBuilder = chronicleHashInstanceBuilder; + this.hub = hub; + } + + @Override + public void process(Wire in, Wire out) throws StreamCorruptedException { + + + // if(LOG.isDebugEnabled()) { + System.out.println("--------------------------------------------\nserver read:\n\n" + Bytes.toDebugString(in.bytes())); + // } + + this.inWire = in; + this.outWire = out; + onEvent(); + } + + private void writeChunked(long transactionId, + @NotNull final Function> function, + @NotNull final Consumer> c) { + + final BytesChronicleMap m = bytesMap(channelId); + final Iterator iterator = function.apply(m); + + // this allows us to write more data than the buffer will allow + out = () -> { + + // each chunk has its own transaction-id + outWire.write(TRANSACTION_ID).int64(transactionId); + + write(map -> { + + boolean hasNext = iterator.hasNext(); + outWire.write(HAS_NEXT).bool(hasNext); + + if (hasNext) + c.accept(iterator); + else + // setting out to NULL denotes that there are no more chunks + out = null; + }); + + }; + + out.run(); + + } + + @SuppressWarnings("UnusedReturnValue") + void onEvent() { + + // it is assumed by this point that the buffer has all the bytes in it for this message + + long transactionId = inWire.read(TRANSACTION_ID).int64(); + timestamp = inWire.read(TIME_STAMP).int64(); + channelId = inWire.read(CHANNEL_ID).int16(); + inWire.read(METHOD_NAME).text(methodName); + + if ("PUT_WITHOUT_ACC".contentEquals(methodName)) { + writeVoid(bytesMap -> { + final net.openhft.lang.io.Bytes reader = toReader(inWire, ARG_1, ARG_2); + // todo bytesMap.put(reader, reader, timestamp, identifier()); + bytesMap.put(reader, reader); + }); + return; + } + + try { + + if ("KEY_SET".contentEquals(methodName)) { + writeChunked(transactionId, map -> map.keySet().iterator(), writeElement); + return; + } + + if ("VALUES".contentEquals(methodName)) { + writeChunked(transactionId, map -> map.values().iterator(), writeElement); + return; + } + + if ("ENTRY_SET".contentEquals(methodName)) { + writeChunked(transactionId, map -> map.entrySet().iterator(), writeEntry); + return; + } + + if ("PUT_ALL".contentEquals(methodName)) { + putAll(transactionId); + return; + } + + // write the transaction id + outWire.write(() -> "TRANSACTION_ID").int64(transactionId); + + if ("CREATE_CHANNEL".contentEquals(methodName)) { + writeVoid(() -> { + short channelId1 = inWire.read(ARG_1).int16(); + chronicleHashInstanceBuilder.get().replicatedViaChannel(hub.createChannel(channelId1)).create(); + return null; + }); + return; + } + + + if ("REMOTE_IDENTIFIER".contentEquals(methodName)) { + this.remoteIdentifier = inWire.read(RESULT).int8(); + return; + } + + if ("LONG_SIZE".contentEquals(methodName)) { + write(b -> outWire.write(RESULT).int64(b.longSize())); + return; + } + + if ("IS_EMPTY".contentEquals(methodName)) { + write(b -> outWire.write(RESULT).bool(b.isEmpty())); + return; + } + + if ("CONTAINS_KEY".contentEquals(methodName)) { + write(b -> outWire.write(RESULT). + bool(b.containsKey(toReader(inWire, ARG_1)))); + return; + } + + if ("CONTAINS_VALUE".contentEquals(methodName)) { + write(b -> outWire.write(RESULT). + bool(b.containsKey(toReader(inWire, ARG_1)))); + return; + } + + if ("GET".contentEquals(methodName)) { + + writeValueUsingDelegate(map -> { + byte[] key1 = this.toByteArray(inWire, ARG_1); + return map.get(key1); + }); + + return; + } + + if ("PUT".contentEquals(methodName)) { + writeValue(b -> { + final net.openhft.lang.io.Bytes reader = toReader(inWire, ARG_1, ARG_2); + + // todo call return b.put(reader, reader, timestamp, identifier()); + return b.put(reader, reader); + }); + + return; + } + + if ("REMOVE".contentEquals(methodName)) { + writeValue(b -> b.remove(toReader(inWire, ARG_1))); + return; + } + + if ("CLEAR".contentEquals(methodName)) { + writeVoid(BytesChronicleMap::clear); + return; + } + + if ("REPLACE".contentEquals(methodName)) { + writeValue(bytesMap -> { + final net.openhft.lang.io.Bytes reader = toReader(inWire, ARG_1, ARG_2); + return bytesMap.replace(reader, reader); + }); + return; + } + + if ("REPLACE_WITH_OLD_AND_NEW_VALUE".contentEquals(methodName)) { + + writeValue(bytesMap -> { + final net.openhft.lang.io.Bytes reader = toReader(inWire, ARG_1, ARG_2, ARG_3); + return bytesMap.replace(reader, reader); + }); + + return; + } + + if ("PUT_IF_ABSENT".contentEquals(methodName)) { + writeValue(bytesMap -> { + final net.openhft.lang.io.Bytes reader = toReader(inWire, ARG_1, ARG_2); + // todo call bytesMap.putIfAbsent(reader, reader, timestamp, identifier()); + return bytesMap.putIfAbsent(reader, reader); + }); + return; + } + + if ("REMOVE_WITH_VALUE".contentEquals(methodName)) { + write(bytesMap -> { + final net.openhft.lang.io.Bytes reader = toReader(inWire, ARG_1, ARG_2); + // todo call outWire.write(RESULT).bool(bytesMap.remove(reader, reader, timestamp, identifier())); + outWire.write(RESULT).bool(bytesMap.remove(reader, reader)); + }); + return; + } + + if ("TO_STRING".contentEquals(methodName)) { + write(b -> outWire.write(RESULT).text(b.toString())); + return; + } + + if ("APPLICATION_VERSION".contentEquals(methodName)) { + write(b -> outWire.write(RESULT).text(applicationVersion())); + return; + } + + if ("PERSISTED_DATA_VERSION".contentEquals(methodName)) { + write(b -> outWire.write(RESULT).text(persistedDataVersion())); + + return; + } + + if ("HASH_CODE".contentEquals(methodName)) { + write(b -> outWire.write(RESULT).int32(b.hashCode())); + return; + } + + throw new IllegalStateException("unsupported event=" + methodName); + + } finally { + + // if (len > 4) + if (EventGroup.IS_DEBUG) { + long len = outWire.bytes().position() - SIZE_OF_SIZE; + System.out.println("--------------------------------------------\nserver wrote:\n\n" + Bytes.toDebugString(outWire.bytes(), SIZE_OF_SIZE, len)); + } + } + + + } + + private byte[] toByteArray(net.openhft.lang.io.Bytes bytes) { + if (bytes == null || bytes.remaining() == 0) + return new byte[]{}; + + if (bytes.remaining() > Integer.MAX_VALUE) + throw new BufferOverflowException(); + + byte[] result = new byte[(int) bytes.remaining()]; + bytes.write(result); + return result; + } + + private byte[] toBytes(WireKey fieldName) { + + final Wire wire = inWire; + System.out.println(Bytes.toDebugString(inWire.bytes())); + + final ValueIn read = wire.read(fieldName); + final long l = read.readLength(); + + if (l > Integer.MAX_VALUE) + throw new BufferOverflowException(); + + final int fieldLength = (int) l; + + final long endPos = wire.bytes().position() + fieldLength; + final long limit = wire.bytes().limit(); + + try { + byte[] bytes = new byte[fieldLength]; + + wire.bytes().read(bytes); + return bytes; + } finally { + wire.bytes().position(endPos); + wire.bytes().limit(limit); + } + } + + private void putAll(long transactionId) { + + final BytesChronicleMap bytesMap = bytesMap(MapWireHandler.this.channelId); + + if (bytesMap == null) + return; + + Runnable runnable = incompleteWork.get(transactionId); + + if (runnable != null) { + runnable.run(); + return; + } + + runnable = new Runnable() { + + // we should try and collect the data and then apply it atomically as quickly possible + final Map collectData = new HashMap<>(); + + @Override + public void run() { + if (inWire.read(HAS_NEXT).bool()) { + collectData.put(toBytes(ARG_1), toBytes(ARG_2)); + } else { + // the old code assumed that all the data would fit into a single buffer + // this assumption is invalid + if (!collectData.isEmpty()) { + bytesMap.delegate.putAll((Map) collectData); + incompleteWork.remove(transactionId); + outWire.write(TRANSACTION_ID).int64(transactionId); + + // todo handle the case where there is an exception + outWire.write(IS_EXCEPTION).bool(false); + + } + } + } + }; + + incompleteWork.put(transactionId, runnable); + runnable.run(); + + } + + private byte identifier() { + // if the client provides a remote identifier then we will use that otherwise we will use the local identifier + return remoteIdentifier == 0 ? localIdentifier : remoteIdentifier; + } + + @NotNull + private CharSequence persistedDataVersion() { + final BytesChronicleMap bytesChronicleMap = bytesMap(channelId); + if (bytesChronicleMap == null) + return ""; + return bytesChronicleMap.delegate.persistedDataVersion(); + } + + @NotNull + private CharSequence applicationVersion() { + return BuildVersion.version(); + } + + /** + * creates a lang buffer that holds just the payload of the args + * + * @param wire the inbound wire + * @param args the key names of the {@code wire} args + * @return a new lang buffer containing the bytes of the args + */ + private net.openhft.lang.io.Bytes toReader(@NotNull Wire wire, @NotNull WireKey... args) { + + final long inSize = wire.bytes().limit(); + final net.openhft.lang.io.Bytes bytes = DirectStore.allocate(inSize).bytes(); + + // copy the bytes to the reader + for (final WireKey field : args) { + + final ValueIn read = wire.read(field); + final long fieldLength = read.readLength(); + + final long endPos = wire.bytes().position() + fieldLength; + final long limit = wire.bytes().limit(); + + try { + + final Bytes source = wire.bytes(); + source.limit(endPos); + + // write the size + bytes.writeStopBit(source.remaining()); + + while (source.remaining() > 0) { + if (source.remaining() >= 8) + bytes.writeLong(source.readLong()); + else + bytes.writeByte(source.readByte()); + } + + } finally { + wire.bytes().position(endPos); + wire.bytes().limit(limit); + } + + } + + return bytes.flip(); + } + + /** + * creates a lang buffer that holds just the payload of the args + * + * @param wire the inbound wire + * @return a new lang buffer containing the bytes of the args + */ + + // todo remove this method - just added to get it to work for now + private byte[] toByteArray(@NotNull Wire wire, @NotNull WireKey field) { + + final ValueIn read = wire.read(field); + final long fieldLength = read.readLength(); + + final long endPos = wire.bytes().position() + fieldLength; + final long limit = wire.bytes().limit(); + byte[] result = new byte[]{}; + try { + + final Bytes source = wire.bytes(); + source.limit(endPos); + + if (source.remaining() > Integer.MAX_VALUE) + throw new BufferOverflowException(); + + // write the size + result = new byte[(int) source.remaining()]; + source.read(result); + + + } finally { + wire.bytes().position(endPos); + wire.bytes().limit(limit); + } + + + return result; + } + + /** + * 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(short channelId) { + + // todo this cast is a bit of a hack, improve later + final ReplicatedChronicleMap replicas = + (ReplicatedChronicleMap) channelList.get(channelId); + + if (replicas != null) + return replicas; + + 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(short 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 f) { + + write(b -> { + + byte[] fromBytes = toByteArray(f.apply(b)); + boolean isNull = fromBytes.length == 0; + outWire.write(RESULT_IS_NULL).bool(isNull); + if (isNull) + return; + + outWire.write(RESULT); + outWire.bytes().write(fromBytes); + + }); + + + } + + @SuppressWarnings("SameReturnValue") + private void writeValueUsingDelegate(final Function, byte[]> f) { + + write(b -> { + + byte[] result = f.apply((ChronicleMap) b.delegate); + boolean isNull = result == null; + + outWire.write(RESULT_IS_NULL).bool(isNull); + if (isNull) + return; + + isNull = result.length == 0; + + outWire.write(RESULT); + outWire.bytes().write(result); + + }); + + + } + + + @SuppressWarnings("SameReturnValue") + private void write(@NotNull Consumer c) { + + final BytesChronicleMap bytesMap = bytesMap(channelId); + + if (bytesMap == null) { + LOG.error("no map for channelId=" + channelId + " can be found."); + return; + } + + bytesMap.output = null; + outWire.bytes().mark(); + outWire.write(IS_EXCEPTION).bool(false); + + try { + c.accept(bytesMap); + } catch (Exception e) { + outWire.bytes().reset(); + // the idea of wire is that is platform independent, + // so we wil have to send the exception as a String + outWire.write(IS_EXCEPTION).bool(true); + outWire.write(EXCEPTION).text(toString(e)); + LOG.error("", e); + return; + } + + } + + @SuppressWarnings("SameReturnValue") + private void writeVoid(@NotNull Callable r) { + + final BytesChronicleMap bytesMap = bytesMap(channelId); + + if (bytesMap == null) { + LOG.error("no map for channelId=" + channelId + " can be found."); + return; + } + + bytesMap.output = null; + outWire.bytes().mark(); + outWire.write(IS_EXCEPTION).bool(false); + + try { + r.call(); + } catch (Exception e) { + outWire.bytes().reset(); + // the idea of wire is that is platform independent, + // so we wil have to send the exception as a String + outWire.write(IS_EXCEPTION).bool(true); + outWire.write(EXCEPTION).text(toString(e)); + LOG.error("", e); + return; + } + + } + + /** + * only used for debugging + */ + @SuppressWarnings("UnusedDeclaration") + private void showOutWire() { + System.out.println("pos=" + outWire.bytes().position() + ",bytes=" + Bytes.toDebugString(outWire.bytes(), 0, outWire.bytes().position())); + } + + @SuppressWarnings("SameReturnValue") + 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."); + return; + } + + bytesMap.output = null; + try { + process.accept(bytesMap); + } catch (Exception e) { + LOG.error("", e); + } + + } + + /** + * converts the exception into a String, so that it can be sent to c# clients + */ + private String toString(@NotNull Throwable t) { + final StringWriter sw = new StringWriter(); + final PrintWriter pw = new PrintWriter(sw); + t.printStackTrace(pw); + return sw.toString(); + } + + public void localIdentifier(byte localIdentifier) { + this.localIdentifier = localIdentifier; + } + + +} diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java b/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java new file mode 100644 index 000000000..41e41b72d --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java @@ -0,0 +1,61 @@ +/* + * Copyright 2014 Higher Frequency Trading + * + * http://www.higherfrequencytrading.com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.map; + +import net.openhft.chronicle.hash.ChronicleHashInstanceBuilder; +import net.openhft.chronicle.hash.replication.ReplicationHub; +import net.openhft.chronicle.network2.WireHandler; +import net.openhft.chronicle.wire.WireKey; +import org.jetbrains.annotations.NotNull; + +import java.util.List; +import java.util.function.Supplier; + +/** + * Created by Rob Austin + */ +public class MapWireHandlerBuilder { + + public static WireHandler of( + @NotNull final Supplier>> chronicleHashInstanceBuilder, + @NotNull final ReplicationHub hub, + byte localIdentifier, + @NotNull final List channelList) { + return new MapWireHandler(chronicleHashInstanceBuilder, hub, localIdentifier, channelList); + } + + public static enum Fields implements WireKey { + HAS_NEXT, + TIME_STAMP, + CHANNEL_ID, + METHOD_NAME, + TYPE, + TRANSACTION_ID, + RESULT, + RESULT_KEY, + RESULT_VALUE, + ARG_1, + ARG_2, + ARG_3, + IS_EXCEPTION, + EXCEPTION, + RESULT_IS_NULL + } +} + diff --git a/src/main/java/net/openhft/chronicle/map/Replica.java b/src/main/java/net/openhft/chronicle/map/Replica.java index 5f233fac0..c0b973ec9 100644 --- a/src/main/java/net/openhft/chronicle/map/Replica.java +++ b/src/main/java/net/openhft/chronicle/map/Replica.java @@ -26,7 +26,7 @@ import java.io.Closeable; /** * @author Rob Austin. */ -interface Replica extends Closeable { +public interface Replica extends Closeable { /** * Provides the unique Identifier associated with this map instance.

An identifier is used diff --git a/src/main/java/net/openhft/chronicle/map/ServiceDescriptor.java b/src/main/java/net/openhft/chronicle/map/ServiceDescriptor.java new file mode 100644 index 000000000..aa82953e1 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/ServiceDescriptor.java @@ -0,0 +1,80 @@ +/* + * Copyright 2014 Higher Frequency Trading + * + * http://www.higherfrequencytrading.com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.map; + +import net.openhft.chronicle.wire.Marshallable; +import net.openhft.chronicle.wire.WireIn; +import net.openhft.chronicle.wire.WireOut; + +public class ServiceDescriptor implements Marshallable { + + public Class keyClass; + public Class valueClass; + public short channelID; + + public ServiceDescriptor() { + } + + public ServiceDescriptor(Class keyClass, Class valueClass, short channelID) { + this.keyClass = keyClass; + this.valueClass = valueClass; + this.channelID = channelID; + } + + + @Override + public void writeMarshallable(WireOut wire) { + wire.write(() -> "keyClass").text(keyClass.getName()); + wire.write(() -> "valueClass").text(valueClass.getName()); + wire.write(() -> "channelID").int16(channelID); + } + + @Override + public void readMarshallable(WireIn wire) throws IllegalStateException { + try { + keyClass = (Class) Class.forName(wire.read(() -> "keyClass").text()); + valueClass = (Class) Class.forName(wire.read(() -> "valueClass").text()); + channelID = wire.read(() -> "channelID").int16(); + } catch (ClassNotFoundException e) { + throw new RuntimeException(e); + } + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + ServiceDescriptor serviceDescriptor = (ServiceDescriptor) o; + + if (channelID != serviceDescriptor.channelID) return false; + if (!keyClass.equals(serviceDescriptor.keyClass)) return false; + if (!valueClass.equals(serviceDescriptor.valueClass)) return false; + + return true; + } + + @Override + public int hashCode() { + int result = keyClass.hashCode(); + result = 31 * result + valueClass.hashCode(); + result = 31 * result + (int) channelID; + return result; + } + } \ No newline at end of file