Skip to content

Commit 43f9a01

Browse files
authored
Minor object store improvement - complete faster (#1237)
* Minor object store improvement - complete faster * validation order * short circuit
1 parent 30b264c commit 43f9a01

File tree

2 files changed

+23
-11
lines changed

2 files changed

+23
-11
lines changed

src/main/java/io/nats/client/impl/NatsFeatureBase.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@
2929
public class NatsFeatureBase {
3030

3131
protected final NatsJetStream js;
32-
protected final JetStreamManagement jsm;
32+
protected final NatsJetStreamManagement jsm;
3333
protected String streamName;
3434

3535
NatsFeatureBase(NatsConnection connection, FeatureOptions fo) throws IOException {

src/main/java/io/nats/client/impl/NatsObjectStore.java

Lines changed: 22 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@
2222
import java.io.*;
2323
import java.nio.file.Files;
2424
import java.security.NoSuchAlgorithmException;
25-
import java.time.Duration;
2625
import java.util.ArrayList;
2726
import java.util.Arrays;
2827
import java.util.List;
@@ -218,6 +217,7 @@ public ObjectInfo get(String objectName, OutputStream out) throws IOException, J
218217
Digester digester = new Digester();
219218
long totalBytes = 0;
220219
long totalChunks = 0;
220+
long expectedChunks = oi.getChunks();
221221

222222
// if there is one chunk, just go get the message directly and we're done.
223223
if (oi.getChunks() == 1) {
@@ -236,30 +236,42 @@ public ObjectInfo get(String objectName, OutputStream out) throws IOException, J
236236
JetStreamSubscription sub = js.subscribe(rawChunkSubject(oi.getNuid()),
237237
PushSubscribeOptions.builder().stream(streamName).ordered(true).build());
238238

239-
Message m = sub.nextMessage(Duration.ofSeconds(1));
239+
Message m = sub.nextMessage(jsm.getTimeout());
240240
while (m != null) {
241+
// track the byte count and chunks
242+
long pending = m.metaData().pendingCount();
243+
if (expectedChunks != pending + (++totalChunks)) {
244+
throw OsGetChunksMismatch.instance(); // short circuit, we already know there are not enough chunks.
245+
}
246+
241247
byte[] data = m.getData();
248+
totalBytes += data.length;
242249

243-
// track the byte count and chunks
244250
// update the digest
245-
// write the bytes to the output file
246-
totalBytes += data.length;
247-
totalChunks++;
248251
digester.update(data);
252+
253+
// write the bytes to the output file
249254
out.write(data);
250255

251256
// read until the subject is complete
252-
m = sub.nextMessage(Duration.ofSeconds(1));
257+
if (pending == 0) {
258+
break;
259+
}
260+
m = sub.nextMessage(jsm.getTimeout());
253261
}
254262

255-
sub.unsubscribe();
263+
try {
264+
sub.unsubscribe();
265+
}
266+
catch (RuntimeException ignore) {}
256267
}
257-
out.flush();
258268

259-
if (totalBytes != oi.getSize()) { throw OsGetSizeMismatch.instance(); }
260269
if (totalChunks != oi.getChunks()) { throw OsGetChunksMismatch.instance(); }
270+
if (totalBytes != oi.getSize()) { throw OsGetSizeMismatch.instance(); }
261271
if (!digester.matches(oi.getDigest())) { throw OsGetDigestMismatch.instance(); }
262272

273+
out.flush(); // moved after validation, no need if invalid
274+
263275
return oi;
264276
}
265277

0 commit comments

Comments
 (0)