Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

moved wire testing into chronicle.map

Browse Source
This commit is contained in:
Rob Austin
2015-03-10 12:50:45 +00:00
parent fb47462542
commit 08d01fce6f
4 changed files with 99 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java
+2 -15
View File
@@ -64,12 +64,13 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
@NotNull
private final OutMessageAdapter outMessageAdapter = new OutMessageAdapter();
@NotNull
private final ArrayList<BytesChronicleMap> bytesChronicleMaps = new ArrayList<>();
private final TextWire inWire = new TextWire(Bytes.elasticByteBuffer());
private final TextWire outWire = new TextWire(Bytes.elasticByteBuffer());
//private final ByteBuffer resultBuffer = ByteBuffer.allocateDirect(64);
@NotNull
private final ByteBuffer inLanByteBuffer = ByteBuffer.allocateDirect(64);
private final StringBuilder methodName = new StringBuilder();
@@ -80,7 +81,6 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
private long timestamp;
private short channelId;
@NotNull
private List<Replica> channelList;
private ReplicationHub hub;
@@ -108,8 +108,6 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
}
};
private Runnable in = null;
private byte remoteIdentifier;
private byte localIdentifer;
private Runnable out = null;
@@ -123,10 +121,6 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
private SelectionKey key;
public void setChannelList(@NotNull List<Replica> channelList) {
if (channelList == null)
throw new IllegalStateException("The StatelessWiredConnector currently only support maps " +
"that are set up via the replication channel. localIdentifier=" + localIdentifier);
this.channelList = channelList;
}
@@ -346,7 +340,6 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
return;
}
// write the transaction id
outWire.write(() -> "TRANSACTION_ID").int64(transactionId);
@@ -445,7 +438,6 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
if ("TO_STRING".contentEquals(methodName)) {
write(b -> outWire.write(() -> "RESULT").text(b.toString()));
return;
}
@@ -468,11 +460,6 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
throw new IllegalStateException("unsupported event=" + methodName);
// else if ("MAP_FOR_KEY".contentEquals(methodName)
// else if ("PUT_MAPPED".contentEquals(methodName))
// else if ("VALUE_BUILDER".contentEquals(methodName))
} finally {
src/main/java/net/openhft/chronicle/map/WiredChronicleMapStatelessClientBuilder.java
+8 -4
View File
@@ -47,6 +47,13 @@ public final class WiredChronicleMapStatelessClientBuilder<K, V> implements
this.channelID = channelID;
}
public WiredChronicleMapStatelessClientBuilder(WiredStatelessClientTcpConnectionHub hub, Class keyClass, Class valueClass, short channelID) {
this.keyClass = keyClass;
this.valueClass = valueClass;
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
@@ -56,8 +63,7 @@ public final class WiredChronicleMapStatelessClientBuilder<K, V> implements
// -- the second is really shorter.
private final InetSocketAddress remoteAddress;
private InetSocketAddress remoteAddress;
private boolean putReturnsNull = false;
private boolean removeReturnsNull = false;
private long timeoutMs = TimeUnit.SECONDS.toMillis(10);
@@ -67,7 +73,6 @@ public final class WiredChronicleMapStatelessClientBuilder<K, V> implements
private final AtomicBoolean used = new AtomicBoolean(false);
InetSocketAddress remoteAddress() {
return remoteAddress;
}
@@ -149,7 +154,6 @@ public final class WiredChronicleMapStatelessClientBuilder<K, V> implements
}
public WiredStatelessClientTcpConnectionHub hub() {
return hub;
}
src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java
+11 -10
View File
@@ -44,7 +44,6 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
private static final Logger LOG = LoggerFactory.getLogger(WiredStatelessChronicleMap.class);
// todo replace with an ENUM
public static final String METHOD_NAME = "METHOD_NAME";
public static final String RESULT_IS_NULL = "RESULT_IS_NULL";
@@ -53,7 +52,6 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
public static final String ARG_1 = "ARG_1";
public static final String ARG_2 = "ARG_2";
private final WiredStatelessClientTcpConnectionHub hub;
protected Class<K> kClass;
@@ -270,7 +268,6 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
return proxyReturnObject(vClass, removeReturnsNull ? REMOVE_WITHOUT_ACC.toString() : REMOVE.toString(), (K) key);
}
@NotNull
public void createChannel(short channelID) {
proxyReturnVoid(() -> {
writeField(METHOD_NAME, "CREATE_CHANNEL");
@@ -278,7 +275,6 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
});
}
public void putAll(@NotNull Map<? extends K, ? extends V> map) {
final long startTime = System.currentTimeMillis();
@@ -377,9 +373,6 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
} else
break;
// return result;
} finally {
hub.inBytesLock().unlock();
}
@@ -479,7 +472,17 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
assert hub.outBytesLock().isHeldByCurrentThread();
assert !hub.inBytesLock().isHeldByCurrentThread();
if (value instanceof CharSequence) {
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);
@@ -722,8 +725,6 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
}
@Nullable
private <R> R proxyReturnObject(@NotNull final String methodName, K key, V value, Class<V> resultType) {
src/test/java/net/openhft/chronicle/wire/TestWire.java → src/test/java/net/openhft/chronicle/map/TestWire.java
+78 -7
View File
@@ -16,14 +16,15 @@
* limitations under the License.
*/
package net.openhft.chronicle.wire;
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.replication.ReplicationHub;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.WiredChronicleMapStatelessClientBuilder;
import net.openhft.chronicle.map.WiredStatelessChronicleMap;
import net.openhft.chronicle.wire.Marshallable;
import net.openhft.chronicle.wire.WireIn;
import net.openhft.chronicle.wire.WireOut;
import org.junit.Assert;
import org.junit.Test;
import java.io.IOException;
@@ -39,7 +40,6 @@ import static org.junit.Assert.*;
*/
public class TestWire {
public static int SERVER_PORT = 8086;
private ChronicleMap<byte[], byte[]> map1a;
private ChronicleMap<byte[], byte[]> map2a;
@@ -47,6 +47,66 @@ public class TestWire {
private ReplicationHub hubA;
@Test
public void testUsingDetails() throws IOException, InterruptedException {
{
final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig
.of(++SERVER_PORT)
.heartBeatInterval(1, SECONDS);
hubA = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig)
.createWithId((byte) 1);
// this is how you add maps after the custer is created
map1a = of(byte[].class, byte[].class)
.instance().replicatedViaChannel(hubA.createChannel((short) 1)).create();
byte identifier = (byte) 2;
final WiredChronicleMapStatelessClientBuilder<String, Details> builder =
new WiredChronicleMapStatelessClientBuilder<>(
new InetSocketAddress("localhost", SERVER_PORT),
String.class,
Details.class,
(short) 1);
builder.identifier(identifier);
final ChronicleMap<String, Details> detailsMap = builder.create();
Details details = detailsMap.get("Colours");
// todo add some sort of cross node locking
if (details == null) {
short channelID = (short) 2;
((WiredStatelessChronicleMap) detailsMap).createChannel(channelID);
details = new Details(String.class, String.class, channelID);
detailsMap.put("Colours", details);
}
ChronicleMap<String, String> colours = new WiredChronicleMapStatelessClientBuilder<String, String>(
builder.hub(),
details.keyClass,
details.valueClass,
details.channelID).hub(builder.hub()).create();
colours.put("Rob", "Blue");
assertEquals(1, colours.size());
Details result = detailsMap.get("Colours");
Assert.assertEquals(2, result.channelID);
map1a.close();
}
}
@Test
public void testCreateChannel() throws IOException, InterruptedException {
@@ -351,8 +411,8 @@ public class TestWire {
Set<Map.Entry<String, String>> entries = statelessMap.entrySet();
assertEquals(2, values.size());
assertTrue(entries.contains(Collections.singletonMap("Key1","Value1").entrySet().iterator().next()));
assertTrue(entries.contains(Collections.singletonMap("Key2","Value2").entrySet().iterator().next()));
assertTrue(entries.contains(Collections.singletonMap("Key1", "Value1").entrySet().iterator().next()));
assertTrue(entries.contains(Collections.singletonMap("Key2", "Value2").entrySet().iterator().next()));
map1a.close();
@@ -420,6 +480,17 @@ public class TestWire {
short channelID;
public Details() {
}
public Details(Class<String> keyClass, Class<String> valueClass, short channelID) {
this.keyClass = keyClass;
this.valueClass = valueClass;
this.channelID = channelID;
}
@Override
public void writeMarshallable(WireOut wire) {
wire.write(() -> "keyClass").text(keyClass.getName());