HCOLL-167 more refactors statelss client

This commit is contained in:
Rob Austin
2014-10-27 19:11:28 +00:00
parent 9f5ee2da1a
commit 03236f31e5
3 changed files with 108 additions and 55 deletions
@@ -602,41 +602,36 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
@NotNull
public synchronized Set<Map.Entry<K, V>> entrySet() {
long sizeLocation = writeEvent(ENTRY_SET);
long startTime = System.currentTimeMillis();
long transactionId = nextUniqueTransaction(startTime);
final long sizeLocation = writeEvent(ENTRY_SET);
final long startTime = System.currentTimeMillis();
final long transactionId = nextUniqueTransaction(startTime);
final long timeoutTime = System.currentTimeMillis() + builder.timeoutMs();
// get the data back from the server
Bytes in = blockingFetch0(sizeLocation, transactionId, startTime);
final Set<Map.Entry<K, V>> result = new HashSet<Map.Entry<K, V>>();
boolean hasMoreEntries;
do {
hasMoreEntries = in.readBoolean();
for (; ; ) {
boolean hasMoreEntries = in.readBoolean();
LOG.info("entrySet - hasMoreEntries=" + hasMoreEntries);
// number of entries in the chunk
long size = in.readInt();
LOG.info("entrySet - size=" + size);
for (int i = 0; i < size; i++) {
K k = keyValueSerializer.readKey(in);
V v = keyValueSerializer.readValue(in);
// LOG.info("v="+v);
result.add(new Entry(k, v));
}
if (hasMoreEntries)
try {
in = blockingFetch(System.currentTimeMillis() + builder.timeoutMs(), transactionId);
} catch (IOException e) {
e.printStackTrace();
}
if (!hasMoreEntries)
break;
} while (hasMoreEntries);
in = blockingFetchReadOnly(timeoutTime, transactionId);
}
return result;
}
@@ -767,6 +762,21 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
}
}
private Bytes blockingFetchReadOnly(long timeoutTime, final long transactionId) {
try {
return blockingFetch(timeoutTime, transactionId);
} catch (IOException e) {
close();
throw new IORuntimeException(e);
} catch (Exception e) {
close();
throw e;
}
}
private Bytes blockingFetch(long timeoutTime, long transactionId) throws IOException {
bytes.clear();
bytes.buffer().clear();
@@ -1667,18 +1667,7 @@ class StatelessServerConnector<K, V> {
@Override
public boolean doWork(Bytes writer) {
final long sizeLocation = writer.position();
writer.skip(AbstractChannelReplicator.SIZE_OF_SIZE + 1); // SIZE_OF_SIZE + is
// exception
writer.writeLong(transactionId);
// hasAnotherChunk
writer.skip(1);
// count
writer.skip(4);
final long sizeLocation = header(writer, transactionId);
int count = 0;
while (iterator.hasNext()) {
@@ -1703,26 +1692,7 @@ class StatelessServerConnector<K, V> {
return true;
}
private void writeHeader(Bytes writer, long sizeLocation, int count, final boolean hasAnotherChunk) {
final long end = writer.position();
final int size = (int) (end - sizeLocation);
writer.position(sizeLocation);
// size in bytes
writer.writeInt(size);
// is exception
writer.writeBoolean(false);
//transaction id;
writer.skip(8);
writer.writeBoolean(hasAnotherChunk);
// count
writer.writeInt(count);
writer.position(end);
}
};
}
@@ -1734,6 +1704,43 @@ class StatelessServerConnector<K, V> {
}
private long header(Bytes writer, final long transactionId) {
final long sizeLocation = writer.position();
writer.skip(AbstractChannelReplicator.SIZE_OF_SIZE + 1); // SIZE_OF_SIZE + is
// exception
writer.writeLong(transactionId);
// hasAnotherChunk
writer.skip(1);
// count
writer.skip(4);
return sizeLocation;
}
private void writeHeader(Bytes writer, long sizeLocation, int count, final boolean hasAnotherChunk) {
final long end = writer.position();
final int size = (int) (end - sizeLocation);
writer.position(sizeLocation);
// size in bytes
writer.writeInt(size);
// is exception
writer.writeBoolean(false);
//transaction id;
writer.skip(8);
writer.writeBoolean(hasAnotherChunk);
// count
writer.writeInt(count);
writer.position(end);
}
private Work putIfAbsent(Bytes reader, Bytes writer) {
final K key = readKey(reader);
final V v = readValue(reader);