Skip to content

Commit 6fa9a0b

Browse files
jyeminjyemin
committed
Remove unused async cursor tryNext support
JAVA-3972
1 parent 82e40ed commit 6fa9a0b

13 files changed

+21
-301
lines changed

‎driver-core/src/main/com/mongodb/internal/async/AsyncBatchCursor.java

Lines changed: 1 addition & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -38,18 +38,6 @@ public interface AsyncBatchCursor<T> extends Closeable {
3838
*/
3939
void next(SingleResultCallback<List<T>> callback);
4040

41-
/**
42-
* A special {@code next()} case that returns the next batch if available or null.
43-
*
44-
* <p>Tailable cursors are an example where this is useful. A call to {@code tryNext()} may return null, but in the future calling
45-
* {@code tryNext()} would return a new batch if a document had been added to the capped collection.</p>
46-
*
47-
* @param callback callback to receive the next batch of results
48-
* @since 3.6
49-
* @mongodb.driver.manual reference/glossary/#term-tailable-cursor Tailable Cursor
50-
*/
51-
void tryNext(SingleResultCallback<List<T>> callback);
52-
5341
/**
5442
* Sets the batch size to use when requesting the next batch. This is the number of documents to request in the next batch.
5543
*
@@ -73,7 +61,7 @@ public interface AsyncBatchCursor<T> extends Closeable {
7361

7462
/**
7563
* Implementations of {@link AsyncBatchCursor} are allowed to close themselves synchronously via methods
76-
* {@link #next(SingleResultCallback)}/{@link #tryNext(SingleResultCallback)}.
64+
* {@link #next(SingleResultCallback)}.
7765
* Self-closing behavior is discouraged because it introduces an additional burden on code that uses {@link AsyncBatchCursor}.
7866
* To help making such code simpler, this method is required to be idempotent.
7967
* <p>

‎driver-core/src/main/com/mongodb/internal/operation/AsyncChangeStreamBatchCursor.java

Lines changed: 0 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -85,17 +85,6 @@ public void apply(final AsyncAggregateResponseBatchCursor<RawBsonDocument> curso
8585
}, callback, false);
8686
}
8787

88-
@Override
89-
public void tryNext(final SingleResultCallback<List<T>> callback) {
90-
resumeableOperation(new AsyncBlock() {
91-
@Override
92-
public void apply(final AsyncAggregateResponseBatchCursor<RawBsonDocument> cursor,
93-
final SingleResultCallback<List<RawBsonDocument>> callback) {
94-
cursor.tryNext(callback);
95-
}
96-
}, callback, true);
97-
}
98-
9988
@Override
10089
public void close() {
10190
if (isClosed.compareAndSet(false, true)) {

‎driver-core/src/main/com/mongodb/internal/operation/AsyncQueryBatchCursor.java

Lines changed: 20 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,6 @@
5353
import static com.mongodb.internal.operation.OperationHelper.getMoreCursorDocumentToQueryResult;
5454
import static com.mongodb.internal.operation.QueryHelper.translateCommandException;
5555
import static com.mongodb.internal.operation.ServerVersionHelper.serverIsAtLeastVersionThreeDotTwo;
56-
import static java.lang.String.format;
5756
import static java.util.Collections.singletonList;
5857

5958
class AsyncQueryBatchCursor<T> implements AsyncAggregateResponseBatchCursor<T> {
@@ -152,12 +151,7 @@ public void close() {
152151

153152
@Override
154153
public void next(final SingleResultCallback<List<T>> callback) {
155-
next(callback, false);
156-
}
157-
158-
@Override
159-
public void tryNext(final SingleResultCallback<List<T>> callback) {
160-
next(callback, true);
154+
internalNext(callback);
161155
}
162156

163157
@Override
@@ -199,16 +193,12 @@ public int getMaxWireVersion() {
199193
return maxWireVersion;
200194
}
201195

202-
private void next(final SingleResultCallback<List<T>> callback, final boolean tryNext) {
196+
private void internalNext(final SingleResultCallback<List<T>> callback) {
203197
if (isClosed()) {
204-
callback.onResult(null, new MongoException(format("%s called after the cursor was closed.",
205-
tryNext ? "tryNext()" : "next()")));
206-
} else if (firstBatch != null && (tryNext || !firstBatch.getResults().isEmpty())) {
198+
callback.onResult(null, new MongoException("next() called after the cursor was closed."));
199+
} else if (firstBatch != null && (!firstBatch.getResults().isEmpty())) {
207200
// May be empty for a tailable cursor
208201
List<T> results = firstBatch.getResults();
209-
if (tryNext && results.isEmpty()) {
210-
results = null;
211-
}
212202
firstBatch = null;
213203
if (getServerCursor() == null) {
214204
close();
@@ -222,13 +212,12 @@ private void next(final SingleResultCallback<List<T>> callback, final boolean tr
222212
} else {
223213
synchronized (this) {
224214
if (isClosed()) {
225-
callback.onResult(null, new MongoException(format("%s called after the cursor was closed.",
226-
tryNext ? "tryNext()" : "next()")));
215+
callback.onResult(null, new MongoException("next() called after the cursor was closed."));
227216
return;
228217
}
229218
isOperationInProgress = true;
230219
}
231-
getMore(localCursor, callback, tryNext);
220+
getMore(localCursor, callback);
232221
}
233222
}
234223
}
@@ -237,9 +226,9 @@ private boolean limitReached() {
237226
return Math.abs(limit) != 0 && count.get() >= Math.abs(limit);
238227
}
239228

240-
private void getMore(final ServerCursor cursor, final SingleResultCallback<List<T>> callback, final boolean tryNext) {
229+
private void getMore(final ServerCursor cursor, final SingleResultCallback<List<T>> callback) {
241230
if (pinnedConnection != null) {
242-
getMore(pinnedConnection.retain(), cursor, callback, tryNext);
231+
getMore(pinnedConnection.retain(), cursor, callback);
243232
} else {
244233
connectionSource.getConnection(new SingleResultCallback<AsyncConnection>() {
245234
@Override
@@ -248,24 +237,23 @@ public void onResult(final AsyncConnection connection, final Throwable t) {
248237
endOperationInProgress();
249238
callback.onResult(null, t);
250239
} else {
251-
getMore(connection, cursor, callback, tryNext);
240+
getMore(connection, cursor, callback);
252241
}
253242
}
254243
});
255244
}
256245
}
257246

258-
private void getMore(final AsyncConnection connection, final ServerCursor cursor, final SingleResultCallback<List<T>> callback,
259-
final boolean tryNext) {
247+
private void getMore(final AsyncConnection connection, final ServerCursor cursor, final SingleResultCallback<List<T>> callback) {
260248
if (serverIsAtLeastVersionThreeDotTwo(connection.getDescription())) {
261249
connection.commandAsync(namespace.getDatabaseName(), asGetMoreCommandDocument(cursor.getId()), NO_OP_FIELD_NAME_VALIDATOR,
262250
ReadPreference.primary(), CommandResultDocumentCodec.create(decoder, "nextBatch"),
263251
connectionSource.getSessionContext(), connectionSource.getServerApi(),
264-
new CommandResultSingleResultCallback(connection, cursor, callback, tryNext));
252+
new CommandResultSingleResultCallback(connection, cursor, callback));
265253

266254
} else {
267255
connection.getMoreAsync(namespace, cursor.getId(), getNumberToReturn(limit, batchSize, count.get()),
268-
decoder, new QueryResultSingleResultCallback(connection, callback, tryNext));
256+
decoder, new QueryResultSingleResultCallback(connection, callback));
269257
}
270258
}
271259

@@ -342,7 +330,7 @@ private BsonDocument asKillCursorsCommandDocument(final ServerCursor localCursor
342330
}
343331

344332
private void endOperationInProgress() {
345-
boolean closePending = false;
333+
boolean closePending;
346334
synchronized (this) {
347335
isOperationInProgress = false;
348336
closePending = this.isClosePending;
@@ -353,7 +341,7 @@ private void endOperationInProgress() {
353341
}
354342

355343
private void handleGetMoreQueryResult(final AsyncConnection connection, final SingleResultCallback<List<T>> callback,
356-
final QueryResult<T> result, final boolean tryNext) {
344+
final QueryResult<T> result) {
357345
cursor.set(result.getCursor());
358346
if (isClosePending) {
359347
try {
@@ -365,8 +353,8 @@ private void handleGetMoreQueryResult(final AsyncConnection connection, final Si
365353
} finally {
366354
callback.onResult(null, null);
367355
}
368-
} else if (!tryNext && result.getResults().isEmpty() && result.getCursor() != null) {
369-
getMore(connection, result.getCursor(), callback, false);
356+
} else if (result.getResults().isEmpty() && result.getCursor() != null) {
357+
getMore(connection, result.getCursor(), callback);
370358
} else {
371359
count.addAndGet(result.getResults().size());
372360
if (limitReached()) {
@@ -392,14 +380,12 @@ private class CommandResultSingleResultCallback implements SingleResultCallback<
392380
private final AsyncConnection connection;
393381
private final ServerCursor cursor;
394382
private final SingleResultCallback<List<T>> callback;
395-
private final boolean tryNext;
396383

397384
CommandResultSingleResultCallback(final AsyncConnection connection, final ServerCursor cursor,
398-
final SingleResultCallback<List<T>> callback, final boolean tryNext) {
385+
final SingleResultCallback<List<T>> callback) {
399386
this.connection = connection;
400387
this.cursor = cursor;
401388
this.callback = errorHandlingCallback(callback, LOGGER);
402-
this.tryNext = tryNext;
403389
}
404390

405391
@Override
@@ -415,21 +401,18 @@ public void onResult(final BsonDocument result, final Throwable t) {
415401
QueryResult<T> queryResult = getMoreCursorDocumentToQueryResult(result.getDocument(CURSOR),
416402
connection.getDescription().getServerAddress());
417403
postBatchResumeToken = getPostBatchResumeTokenFromResponse(result);
418-
handleGetMoreQueryResult(connection, callback, queryResult, tryNext);
404+
handleGetMoreQueryResult(connection, callback, queryResult);
419405
}
420406
}
421407
}
422408

423409
private class QueryResultSingleResultCallback implements SingleResultCallback<QueryResult<T>> {
424410
private final AsyncConnection connection;
425411
private final SingleResultCallback<List<T>> callback;
426-
private final boolean tryNext;
427412

428-
QueryResultSingleResultCallback(final AsyncConnection connection, final SingleResultCallback<List<T>> callback,
429-
final boolean tryNext) {
413+
QueryResultSingleResultCallback(final AsyncConnection connection, final SingleResultCallback<List<T>> callback) {
430414
this.connection = connection;
431415
this.callback = errorHandlingCallback(callback, LOGGER);
432-
this.tryNext = tryNext;
433416
}
434417

435418
@Override
@@ -439,7 +422,7 @@ public void onResult(final QueryResult<T> result, final Throwable t) {
439422
endOperationInProgress();
440423
callback.onResult(null, t);
441424
} else {
442-
handleGetMoreQueryResult(connection, callback, result, tryNext);
425+
handleGetMoreQueryResult(connection, callback, result);
443426
}
444427
}
445428
}

‎driver-core/src/main/com/mongodb/internal/operation/AsyncSingleBatchQueryCursor.java

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -53,11 +53,6 @@ public void next(final SingleResultCallback<List<T>> callback) {
5353
}
5454
}
5555

56-
@Override
57-
public void tryNext(final SingleResultCallback<List<T>> callback) {
58-
next(callback);
59-
}
60-
6156
@Override
6257
public void setBatchSize(final int batchSize) {
6358
// Noop

‎driver-core/src/main/com/mongodb/internal/operation/ListCollectionsOperation.java

Lines changed: 0 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -466,20 +466,6 @@ public void onResult(final List<BsonDocument> result, final Throwable t) {
466466
});
467467
}
468468

469-
@Override
470-
public void tryNext(final SingleResultCallback<List<T>> callback) {
471-
delegate.tryNext(new SingleResultCallback<List<BsonDocument>>() {
472-
@Override
473-
public void onResult(final List<BsonDocument> result, final Throwable t) {
474-
if (t != null) {
475-
callback.onResult(null, t);
476-
} else {
477-
callback.onResult(projectFromFullNamespaceToCollectionName(result), null);
478-
}
479-
}
480-
});
481-
}
482-
483469
@Override
484470
public void setBatchSize(final int batchSize) {
485471
delegate.setBatchSize(batchSize);

‎driver-core/src/test/functional/com/mongodb/OperationFunctionalSpecification.groovy

Lines changed: 0 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -188,18 +188,6 @@ class OperationFunctionalSpecification extends Specification {
188188
}
189189
}
190190

191-
def tryNext(cursor, boolean async) {
192-
def next
193-
if (async) {
194-
def futureResultCallback = new FutureResultCallback<List<BsonDocument>>()
195-
cursor.tryNext(futureResultCallback)
196-
next = futureResultCallback.get(TIMEOUT, TimeUnit.SECONDS)
197-
} else {
198-
next = cursor.tryNext()
199-
}
200-
next
201-
}
202-
203191
def consumeAsyncResults(cursor) {
204192
def batch = next(cursor, true)
205193
while (batch != null) {

‎driver-core/src/test/functional/com/mongodb/internal/operation/AggregateOperationSpecification.groovy

Lines changed: 0 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -440,29 +440,6 @@ class AggregateOperationSpecification extends OperationFunctionalSpecification {
440440
async << [true, false]
441441
}
442442

443-
@IgnoreIf({ isStandalone() || !serverVersionAtLeast(3, 6) })
444-
def 'should be able to respect maxAwaitTime with pipeline'() {
445-
given:
446-
AggregateOperation operation = new AggregateOperation<Document>(getNamespace(), [
447-
new BsonDocument('$changeStream', new BsonDocument())
448-
], new DocumentCodec())
449-
.batchSize(2)
450-
.maxAwaitTime(10, MILLISECONDS)
451-
452-
when:
453-
def cursor = execute(operation, async)
454-
tryNext(cursor, async)
455-
456-
then:
457-
noExceptionThrown()
458-
459-
cleanup:
460-
cursor?.close()
461-
462-
where:
463-
async << [true, false]
464-
}
465-
466443
def 'should add read concern to command'() {
467444
given:
468445
def binding = Stub(ReadBinding)

‎driver-core/src/test/functional/com/mongodb/internal/operation/ChangeStreamOperationSpecification.groovy

Lines changed: 0 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -433,19 +433,11 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
433433

434434
when:
435435
def cursor = execute(operation, async)
436-
437-
then:
438-
tryNextAndClean(cursor, async) == null
439-
440-
when:
441436
def expected = insertDocuments(helper, [1, 2])
442437

443438
then:
444439
nextAndClean(cursor, async, expected.size()) == expected
445440

446-
then:
447-
tryNextAndClean(cursor, async) == null
448-
449441
when:
450442
expected = insertDocuments(helper, [3, 4])
451443

@@ -482,9 +474,6 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
482474
then:
483475
results == expected
484476

485-
then:
486-
tryNextAndClean(cursor, async) == null
487-
488477
when:
489478
expected = insertDocuments(helper, [5, 6])
490479
helper.killCursor(helper.getNamespace(), cursor.getWrapped().getServerCursor())
@@ -755,10 +744,6 @@ class ChangeStreamOperationSpecification extends OperationFunctionalSpecificatio
755744
}
756745
}
757746

758-
def tryNextAndClean(cursor, boolean async) {
759-
removeExtra(tryNext(cursor, async))
760-
}
761-
762747
def nextAndClean(cursor, boolean async, int minimumCount) {
763748
removeExtra(next(cursor, async, minimumCount))
764749
}

‎driver-core/src/test/functional/com/mongodb/internal/operation/ListCollectionsOperationSpecification.groovy

Lines changed: 0 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -85,14 +85,6 @@ class ListCollectionsOperationSpecification extends OperationFunctionalSpecifica
8585
then:
8686
callback.get() == null
8787

88-
when:
89-
cursor = executeAsync(operation)
90-
callback = new FutureResultCallback()
91-
cursor.tryNext(callback)
92-
93-
then:
94-
callback.get() == null
95-
9688
cleanup:
9789
collectionHelper.dropDatabase(madeUpDatabase)
9890
}

0 commit comments

Comments
 (0)