mirror of
https://github.com/OpenHFT/Chronicle-Map.git
synced 2021-09-13 01:26:12 +00:00
CE-14 moved map code into the map project
This commit is contained in:
Rob Austin
@@ -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);
|
||||
}
|
||||
+164
@@ -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.lang.MemoryUnit;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
/**
|
||||
* @author Rob Austin.
|
||||
*/
|
||||
public final class ClientWiredChronicleMapStatelessClientBuilder<K, V> implements
|
||||
ChronicleHashStatelessClientBuilder<ClientWiredChronicleMapStatelessClientBuilder<K, V>,
|
||||
ChronicleMap<K, V>>,
|
||||
MapBuilder<ClientWiredChronicleMapStatelessClientBuilder<K, V>> {
|
||||
|
||||
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.<Integer, Integer>.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<K, V> putReturnsNull(boolean putReturnsNull) {
|
||||
this.putReturnsNull = putReturnsNull;
|
||||
return this;
|
||||
}
|
||||
|
||||
boolean putReturnsNull() {
|
||||
return putReturnsNull;
|
||||
}
|
||||
|
||||
@Override
|
||||
public ClientWiredChronicleMapStatelessClientBuilder<K, V> removeReturnsNull(boolean removeReturnsNull) {
|
||||
this.removeReturnsNull = removeReturnsNull;
|
||||
return this;
|
||||
}
|
||||
|
||||
boolean removeReturnsNull() {
|
||||
return removeReturnsNull;
|
||||
}
|
||||
|
||||
@Override
|
||||
public ClientWiredChronicleMapStatelessClientBuilder<K, V> timeout(long timeout, TimeUnit units) {
|
||||
this.timeoutMs = units.toMillis(timeout);
|
||||
return this;
|
||||
}
|
||||
|
||||
long timeoutMs() {
|
||||
return timeoutMs;
|
||||
}
|
||||
|
||||
@Override
|
||||
public ClientWiredChronicleMapStatelessClientBuilder<K, V> name(String name) {
|
||||
this.name = name;
|
||||
return this;
|
||||
}
|
||||
|
||||
String name() {
|
||||
return name;
|
||||
}
|
||||
|
||||
@Override
|
||||
public ClientWiredChronicleMapStatelessClientBuilder<K, V> tcpBufferSize(int tcpBufferSize) {
|
||||
this.tcpBufferSize = tcpBufferSize;
|
||||
return this;
|
||||
}
|
||||
|
||||
int tcpBufferSize() {
|
||||
return tcpBufferSize;
|
||||
}
|
||||
|
||||
@Override
|
||||
public ChronicleMap<K, V> create() throws IOException {
|
||||
|
||||
// todo clean this up
|
||||
if (hub == null)
|
||||
hub = new ClientWiredStatelessClientTcpConnectionHub(this, localIdentifier, doHandShaking);
|
||||
|
||||
if (!used.getAndSet(true)) {
|
||||
return new ClientWiredStatelessChronicleMap<K, V>(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;
|
||||
}
|
||||
|
||||
|
||||
|
||||
}
|
||||
@@ -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<K, V> implements ChronicleMap<K, V>, Cloneable, ChannelFactory {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(ClientWiredStatelessChronicleMap.class);
|
||||
|
||||
private final ClientWiredStatelessClientTcpConnectionHub hub;
|
||||
|
||||
protected Class<K> kClass;
|
||||
protected Class<V> 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<K> keyClass() {
|
||||
return kClass;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean forEachEntryWhile(Predicate<? super MapKeyContext<K, V>> predicate) {
|
||||
// TODO implement!
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void forEachEntry(Consumer<? super MapKeyContext<K, V>> action) {
|
||||
// TODO implement!
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
|
||||
public String serverApplicationVersion() {
|
||||
return hub.serverApplicationVersion(channelID);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
// todo add ref count
|
||||
}
|
||||
|
||||
@Override
|
||||
public Class<V> 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<? extends K, ? extends V> that = (Map<? extends K, ? extends V>) object;
|
||||
|
||||
final int size = size();
|
||||
|
||||
if (that.size() != size)
|
||||
return false;
|
||||
|
||||
final Set<Map.Entry<K, V>> 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<K, V> 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<K, V> 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<? extends K, ? extends V> map) {
|
||||
|
||||
final long startTime = System.currentTimeMillis();
|
||||
long transactionId = hub.nextUniqueTransaction(startTime);
|
||||
|
||||
|
||||
Set<? extends Map.Entry<? extends K, ? extends V>> entries = map.entrySet();
|
||||
Iterator<? extends Map.Entry<? extends K, ? extends V>> 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<? extends K, ? extends V> 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> R getMapped(@Nullable K key, @NotNull SerializableFunction<? super V, R> function) {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
@Nullable
|
||||
@Override
|
||||
public V putMapped(@Nullable K key, @NotNull UnaryOperator<V> unaryOperator) {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
public void clear() {
|
||||
proxyReturnVoid(() -> writeField(Fields.METHOD_NAME, CLEAR.toString()));
|
||||
}
|
||||
|
||||
|
||||
public Set<K> keySet() {
|
||||
final long startTime = System.currentTimeMillis();
|
||||
final Set<K> result = new HashSet<K>();
|
||||
// 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<V> values() {
|
||||
final long startTime = System.currentTimeMillis();
|
||||
final Set<V> result = new HashSet<V>();
|
||||
// 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<Map.Entry<K, V>> entrySet() {
|
||||
final long startTime = System.currentTimeMillis();
|
||||
final Map<K, V> result = new HashMap<K, V>();
|
||||
|
||||
// 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<K, V> 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> R proxyReturnObject(@NotNull final String methodName, K key, V value, Class<V> 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> R proxyReturnObject(Class<R> 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<K, V> {
|
||||
|
||||
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();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+648
@@ -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.AbstractChannelReplicator.SIZE_OF_SIZE;
|
||||
import static net.openhft.chronicle.map.ClientWiredStatelessChronicleMap.EventId.APPLICATION_VERSION;
|
||||
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 * TcpReplicator.TIMESTAMP_FACTOR;
|
||||
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;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,857 @@
|
||||
/*
|
||||
* 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.ByteBufferBytes;
|
||||
import net.openhft.lang.io.DirectStore;
|
||||
import net.openhft.lang.io.IByteBufferBytes;
|
||||
import net.openhft.lang.io.NativeBytes;
|
||||
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.ByteBuffer;
|
||||
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.AbstractChannelReplicator.SIZE_OF_SIZE;
|
||||
import static net.openhft.chronicle.map.MapWireHandlerBuilder.Fields.*;
|
||||
|
||||
/**
|
||||
* @author Rob Austin.
|
||||
*/
|
||||
class MapWireHandler<K, V> implements WireHandler {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(MapWireHandler.class);
|
||||
|
||||
private final NativeBytes inLangBytes = new NativeBytes(0, 0);
|
||||
@NotNull
|
||||
|
||||
private final OutMessageAdapter outMessageAdapter = new OutMessageAdapter();
|
||||
@NotNull
|
||||
|
||||
private final ArrayList<BytesChronicleMap> bytesChronicleMaps = new ArrayList<>();
|
||||
@NotNull
|
||||
|
||||
private final ByteBuffer inLanByteBuffer = ByteBuffer.allocateDirect(64);
|
||||
private final StringBuilder methodName = new StringBuilder();
|
||||
private final Supplier<ChronicleHashInstanceBuilder<ChronicleMap<K, V>>> chronicleHashInstanceBuilder;
|
||||
private Wire inWire = null;
|
||||
private Wire outWire = null;
|
||||
private final Consumer writeElement = new Consumer<Iterator<byte[]>>() {
|
||||
|
||||
@Override
|
||||
public void accept(Iterator<byte[]> iterator) {
|
||||
outWire.write(RESULT);
|
||||
outWire.bytes().write(iterator.next());
|
||||
}
|
||||
};
|
||||
private byte localIdentifier;
|
||||
private long timestamp;
|
||||
private short channelId;
|
||||
private List<Replica> channelList;
|
||||
private ReplicationHub hub;
|
||||
private byte remoteIdentifier;
|
||||
private final Consumer writeEntry = new Consumer<Iterator<Map.Entry<byte[], byte[]>>>() {
|
||||
|
||||
@Override
|
||||
public void accept(Iterator<Map.Entry<byte[], byte[]>> iterator) {
|
||||
|
||||
final Map.Entry<byte[], byte[]> entry = iterator.next();
|
||||
|
||||
outWire.write(RESULT_KEY);
|
||||
outWire.bytes().write(entry.getKey());
|
||||
|
||||
outWire.write(RESULT_VALUE);
|
||||
outWire.bytes().write(entry.getValue());
|
||||
}
|
||||
};
|
||||
private Runnable out = null;
|
||||
private SelectionKey key = null;
|
||||
|
||||
public MapWireHandler(@NotNull final Supplier<ChronicleHashInstanceBuilder<ChronicleMap<K, V>>> chronicleHashInstanceBuilder,
|
||||
@NotNull final ReplicationHub hub,
|
||||
byte localIdentifier,
|
||||
@NotNull final List<Replica> channelList) {
|
||||
this(chronicleHashInstanceBuilder, hub);
|
||||
this.channelList = channelList;
|
||||
this.localIdentifier = localIdentifier;
|
||||
}
|
||||
|
||||
|
||||
public MapWireHandler(Supplier<ChronicleHashInstanceBuilder<ChronicleMap<K, V>>> chronicleHashInstanceBuilder, ReplicationHub hub) {
|
||||
this.chronicleHashInstanceBuilder = chronicleHashInstanceBuilder;
|
||||
this.hub = hub;
|
||||
}
|
||||
|
||||
private Map<Long, Runnable> incompleteWork = new HashMap<>();
|
||||
|
||||
|
||||
@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<Map, Iterator<byte[]>> function,
|
||||
@NotNull final Consumer<Iterator<byte[]>> c) {
|
||||
|
||||
final BytesChronicleMap m = bytesMap(channelId);
|
||||
final Iterator<byte[]> 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);
|
||||
bytesMap.put(reader, reader, timestamp, identifier());
|
||||
});
|
||||
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);
|
||||
return b.put(reader, reader, timestamp, identifier());
|
||||
});
|
||||
|
||||
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);
|
||||
return bytesMap.putIfAbsent(reader, reader, timestamp, identifier());
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
if ("REMOVE_WITH_VALUE".contentEquals(methodName)) {
|
||||
write(bytesMap -> {
|
||||
final net.openhft.lang.io.Bytes reader = toReader(inWire, ARG_1, ARG_2);
|
||||
outWire.write(RESULT).bool(bytesMap.remove(reader, reader, timestamp, identifier()));
|
||||
});
|
||||
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<byte[], byte[]> 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.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<BytesChronicleMap, net.openhft.lang.io.Bytes> 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<ChronicleMap<byte[], byte[]>, 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);
|
||||
|
||||
});
|
||||
|
||||
|
||||
}
|
||||
|
||||
private void writeValueTcpRep(@NotNull Consumer<BytesChronicleMap> process) {
|
||||
|
||||
final BytesChronicleMap bytesMap = bytesMap(channelId);
|
||||
|
||||
if (bytesMap == null) {
|
||||
LOG.error("no map for channelId=" + channelId + " can be found.");
|
||||
return;
|
||||
}
|
||||
|
||||
inLanByteBuffer.clear();
|
||||
inLangBytes.clear();
|
||||
outMessageAdapter.markStartOfMessage();
|
||||
|
||||
bytesMap.output = outMessageAdapter.outBuffer;
|
||||
|
||||
try {
|
||||
process.accept(bytesMap);
|
||||
} catch (Exception e) {
|
||||
|
||||
// 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;
|
||||
}
|
||||
|
||||
outMessageAdapter.accept(outWire);
|
||||
|
||||
}
|
||||
|
||||
@SuppressWarnings("SameReturnValue")
|
||||
private void write(@NotNull Consumer<BytesChronicleMap> 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<BytesChronicleMap> 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;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* used to send data to the wired stateless client, this class converts/adapts a chronicle map
|
||||
* lang-bytes output into chronicle-bytes
|
||||
*/
|
||||
private static class OutMessageAdapter {
|
||||
|
||||
|
||||
@NotNull
|
||||
private final IByteBufferBytes outLangBytes = ByteBufferBytes.wrap(ByteBuffer.allocate(1024));
|
||||
private Bytes outChronBytes = Bytes.wrap(outLangBytes.buffer());
|
||||
|
||||
// chronicle map will write its result in here
|
||||
final MapIOBuffer outBuffer = new MapIOBuffer() {
|
||||
|
||||
@Override
|
||||
public void ensureBufferSize(long l) {
|
||||
|
||||
if (outLangBytes.remaining() >= l)
|
||||
return;
|
||||
|
||||
long size = outLangBytes.capacity() + l;
|
||||
if (size > Integer.MAX_VALUE)
|
||||
throw new BufferOverflowException();
|
||||
|
||||
// record the current position and limit
|
||||
long position = outLangBytes.position();
|
||||
|
||||
// create a new buffer and copy the data into it
|
||||
outLangBytes.clear();
|
||||
final IByteBufferBytes newOutLangBytes = ByteBufferBytes.wrap(ByteBuffer.allocate((int) size));
|
||||
newOutLangBytes.write(outLangBytes);
|
||||
|
||||
outChronBytes = Bytes.wrap(outLangBytes.buffer());
|
||||
|
||||
newOutLangBytes.limit(newOutLangBytes.capacity());
|
||||
newOutLangBytes.position(position);
|
||||
|
||||
outChronBytes.limit(outChronBytes.capacity());
|
||||
outChronBytes.position(position);
|
||||
}
|
||||
|
||||
@NotNull
|
||||
@Override
|
||||
public net.openhft.lang.io.Bytes in() {
|
||||
return outLangBytes;
|
||||
}
|
||||
|
||||
};
|
||||
|
||||
|
||||
private long startChunk;
|
||||
|
||||
void clear() {
|
||||
outChronBytes.clear();
|
||||
outLangBytes.clear();
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* adapts the chronicle out lang bytes to chroncile bytes
|
||||
*
|
||||
* @param wire the wire that we wish to append data to
|
||||
*/
|
||||
void accept(@NotNull Wire wire) {
|
||||
|
||||
wire.write(IS_EXCEPTION).bool(false);
|
||||
|
||||
// flips calls flip on this message so that we can read it
|
||||
flipMessage();
|
||||
|
||||
// set the chron-bytes and the lang-bytes to be the same
|
||||
outChronBytes.position(outLangBytes.position());
|
||||
outChronBytes.limit(outLangBytes.limit());
|
||||
|
||||
// is Null
|
||||
boolean isNull = outChronBytes.readBoolean();
|
||||
|
||||
// read the size - not used
|
||||
outChronBytes.readStopBit();
|
||||
|
||||
wire.write(RESULT_IS_NULL).bool(isNull);
|
||||
if (!isNull) {
|
||||
// write the result
|
||||
wire.write(RESULT);
|
||||
wire.bytes().write(outChronBytes);
|
||||
}
|
||||
}
|
||||
|
||||
private void flipMessage() {
|
||||
long position = outLangBytes.position();
|
||||
outLangBytes.position(startChunk);
|
||||
outLangBytes.limit(position);
|
||||
}
|
||||
|
||||
/**
|
||||
* marks the start of the message
|
||||
*/
|
||||
public void markStartOfMessage() {
|
||||
startChunk = outLangBytes.position();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
@@ -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 <K, V> WireHandler of(
|
||||
@NotNull final Supplier<ChronicleHashInstanceBuilder<ChronicleMap<K, V>>> chronicleHashInstanceBuilder,
|
||||
@NotNull final ReplicationHub hub,
|
||||
byte localIdentifier,
|
||||
@NotNull final List<Replica> channelList) {
|
||||
return new MapWireHandler<K, V>(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
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user