HCOLL-167 added support for values

This commit is contained in:
Rob Austin
2014-10-28 17:18:53 +00:00
parent 64d19bd359
commit e7b502c7c6
3 changed files with 82 additions and 18 deletions
@@ -546,7 +546,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
assert buffer.capacity() == bytes.capacity();
assert bytes.limit() == bytes.capacity();
}
@@ -561,21 +560,38 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
@NotNull
public synchronized Collection<V> values() {
long sizeLocation = writeEvent(VALUES);
final long sizeLocation = writeEvent(VALUES);
final long startTime = System.currentTimeMillis();
final long transactionId = nextUniqueTransaction(startTime);
final long timeoutTime = System.currentTimeMillis() + builder.timeoutMs();
// get the data back from the server
final Bytes in = blockingFetch(sizeLocation);
Bytes in = blockingFetch0(sizeLocation, transactionId, startTime);
final long size = in.readStopBit();
final Collection<V> result = new ArrayList<V>();
if (size > Integer.MAX_VALUE)
throw new IllegalStateException("size=" + size + " is too large.");
for (; ; ) {
final ArrayList<V> result = new ArrayList<V>((int) size);
boolean hasMoreEntries = in.readBoolean();
for (int i = 0; i < size; i++) {
result.add(keyValueSerializer.readValue(in));
// number of entries in the chunk
long size = in.readInt();
for (int i = 0; i < size; i++) {
V v = keyValueSerializer.readValue(in);
result.add(v);
}
if (!hasMoreEntries)
break;
compact(in);
in = blockingFetchReadOnly(timeoutTime, transactionId);
}
return result;
}
@@ -1561,18 +1561,18 @@ class StatelessServerConnector<K, V> {
private Work values(Bytes reader, Bytes writer) {
final long sizeLocation = reflectTransactionId(reader, writer);
final Collection<V> values;
final long transactionId = reader.readLong();
Collection<V> ks;
try {
values = map.values();
ks = map.values();
} catch (RuntimeException e) {
return sendException(writer, sizeLocation, e);
return sendException(reader, writer, e);
}
writer.writeStopBit(values.size());
final Iterator<V> iterator = values.iterator();
final Iterator<V> iterator = ks.iterator();
// this allows us to write more data than the buffer will allow
return new Work() {
@@ -1580,17 +1580,24 @@ class StatelessServerConnector<K, V> {
@Override
public boolean doWork(Bytes writer) {
final long sizeLocation = header(writer, transactionId);
int count = 0;
while (iterator.hasNext()) {
// we've filled up the buffer, so lets give another channel a chance to send
// some data, we don't know the max key size, we will use the entrySize instead
if (writer.remaining() <= maxEntrySizeBytes)
if (writer.remaining() <= maxEntrySizeBytes) {
writeHeader(writer, sizeLocation, count, true);
return false;
}
count++;
writeValue(iterator.next(), writer);
}
writeSizeAndFlags(sizeLocation, false, writer);
writeHeader(writer, sizeLocation, count, false);
return true;
}
};
@@ -1601,7 +1608,6 @@ class StatelessServerConnector<K, V> {
Set<K> ks;
try {
ks = map.keySet();
} catch (RuntimeException e) {