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

fixed issue with proxyString read and

Browse Source 
Merge branch 'master' of https://github.com/OpenHFT/Chronicle-Wire

Conflicts:
	src/main/java/net/openhft/chronicle/wire/TextWire.java
This commit is contained in:
Rob Austin
2015-03-06 22:50:04 +00:00
parent a530bc7e3c
commit f2664210fe
5 changed files with 64 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pom.xml
+10
View File
@@ -69,6 +69,11 @@
<version>1.0.1-alpha-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>net.openhft</groupId>
<artifactId>chronicle-bytes</artifactId>
<version>1.0.0-alpha-SNAPSHOT</version>
</dependency>
</dependencies>
@@ -101,6 +106,11 @@
<artifactId>chronicle-wire</artifactId>
</dependency>
<dependency>
<groupId>net.openhft</groupId>
<artifactId>chronicle-bytes</artifactId>
</dependency>
<!-- optional dependencies -->
<dependency>
src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java
+1
View File
@@ -119,6 +119,7 @@ public final class ReplicationHub extends AbstractReplication {
return channel;
}
/**
* Builder of {@link ReplicationHub}s.
*/
src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java
+3 -5
View File
@@ -21,7 +21,6 @@ package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.function.SerializableFunction;
import net.openhft.chronicle.wire.Marshallable;
import net.openhft.chronicle.wire.Wire;
import net.openhft.lang.threadlocal.ThreadLocalCopies;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
@@ -42,7 +41,7 @@ import static net.openhft.chronicle.map.WiredStatelessChronicleMap.EventId.*;
/**
* @author Rob Austin.
*/
class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Cloneable {
public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Cloneable {
private static final Logger LOG = LoggerFactory.getLogger(WiredStatelessChronicleMap.class);
private final WiredStatelessClientTcpConnectionHub hub;
@@ -414,7 +413,7 @@ class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Cloneable
}
}
private String readString(long transactionId, long startTime) {
/* private String readString(long transactionId, long startTime) {
assert !hub.outBytesLock().isHeldByCurrentThread();
long timeoutTime = startTime + hub.timeoutMs;
@@ -427,7 +426,7 @@ class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Cloneable
} finally {
hub.inBytesLock().unlock();
}
}
}*/
@SuppressWarnings("SameParameterValue")
@@ -622,7 +621,6 @@ class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Cloneable
@Nullable
private <R> R proxyReturnObject(Class<R> rClass, @NotNull final String methodName, Object key) {
final long startTime = System.currentTimeMillis();
ThreadLocalCopies copies;
long transactionId;
src/main/java/net/openhft/chronicle/map/WiredStatelessClientTcpConnectionHub.java
+5 -21
View File
@@ -92,7 +92,7 @@ public class WiredStatelessClientTcpConnectionHub {
this.name = config.name();
this.timeoutMs = config.timeoutMs();
attemptConnect(remoteAddress);
// checkVersion(config.channelID());
checkVersion(config.channelID());
}
private synchronized void attemptConnect(final InetSocketAddress remoteAddress) {
@@ -264,6 +264,7 @@ public class WiredStatelessClientTcpConnectionHub {
/**
* sends data to the server via TCP/IP
*
* @param wire the {@code wire} containing the outbound data
*/
void writeSocket(@NotNull final Wire wire) {
@@ -375,7 +376,7 @@ public class WiredStatelessClientTcpConnectionHub {
intWire.bytes().skip(4);
intWire.bytes().limit(messageSize);
long transactionId0 = intWire.read(() -> "TRANSACTION_ID").int64();
long transactionId0 = intWire.read(() -> "TRANSACTION_ID").int64();
// check the transaction id is reasonable
@@ -555,24 +556,6 @@ public class WiredStatelessClientTcpConnectionHub {
}
private long proxySend(@NotNull final String methodName, final long startTime, short channelID) {
assert outBytesLock().isHeldByCurrentThread();
assert !inBytesLock().isHeldByCurrentThread();
// send
outBytesLock().lock();
try {
long transactionId = writeHeader(startTime, channelID, outWire());
outWire.write(() -> "METHOD_NAME").text(methodName);
writeSocket(outWire);
return transactionId;
} finally {
outBytesLock().unlock();
}
}
private long proxySend(@NotNull final String methodName,
final long startTime,
short channelID,
@@ -617,7 +600,7 @@ public class WiredStatelessClientTcpConnectionHub {
// receive
inBytesLock().lock();
try {
return proxyReply(timeoutTime, transactionId).read(() -> messageId).text();
return proxyReply(timeoutTime, transactionId).read(() -> "RESULT").text();
} catch (Exception e) {
throw new RuntimeException(e);
} finally {
@@ -649,6 +632,7 @@ public class WiredStatelessClientTcpConnectionHub {
/**
* mark the location of the outWire size
*
* @param outWire
*/
void markSize(Wire outWire) {
src/test/java/net/openhft/chronicle/wire/TestWire.java
+45 -3
View File
@@ -23,6 +23,7 @@ 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 org.junit.Test;
import java.io.IOException;
@@ -80,12 +81,51 @@ public class TestWire {
assertFalse(statelessMap.isEmpty());
}
map1a.close();
}
}
@Test
public void testVersion() 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();
short channelID = (short) 1;
byte identifier = (byte) 2;
final WiredChronicleMapStatelessClientBuilder<String, String> builder =
new WiredChronicleMapStatelessClientBuilder<>(
new InetSocketAddress("localhost", SERVER_PORT),
String.class,
String.class,
channelID);
builder.identifier(identifier);
builder.putReturnsNull(true);
try (ChronicleMap<String, String> statelessMap = builder.create()) {
String s = ((WiredStatelessChronicleMap) statelessMap).serverApplicationVersion();
Assert.assertNotNull(s);
}
map1a.close();
}
@Test
public void testLongSize() throws IOException, InterruptedException {
@@ -120,7 +160,7 @@ public class TestWire {
assertEquals(1, statelessMap.longSize());
assertFalse(statelessMap.isEmpty());
}
map1a.close();
}
@@ -173,6 +213,9 @@ public class TestWire {
Assert.assertEquals(expected, actual);
map1a.close();
map2a.close();
map3a.close();
}
}
@@ -208,13 +251,12 @@ public class TestWire {
final String actual = statelessMap.get("FirstMap");
Assert.assertEquals(null, actual);
map1a.close();
}
}
public static class Details implements Marshallable {
Class keyClass;