private static <T> T executeAndGet(Consumer<SingleResultCallback<T>> execution) throws ExecutionException, TimeoutException, InterruptedException {
    CompletableFuture<T> future = new CompletableFuture<>();
    SingleResultCallback<T> callback = getCallback(future);
    execution.accept(callback);
    int TIMEOUT_IN_MILLIS = 550;
    return future.get(TIMEOUT_IN_MILLIS, TimeUnit.MILLISECONDS);
}
 

private static <T> SingleResultCallback<T> getCallback(final CompletableFuture<T> future) {
    return new SingleResultCallback<>() {
        @Override
        public void onResult(final T result, final Throwable t) {
            if (t != null) {
                future.completeExceptionally(t);
            } else {
                future.complete(result);
            }
        }
    };
}
 

@Override
public Observable<Integer> write(final ByteBuffer src) {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<Integer>>() {
        @Override
        public void apply(final SingleResultCallback<Integer> callback) {
            wrapped.write(src, callback);
        }
    }), observableAdapter);
}
 

@Override
public Observable<Success> close() {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<Success>>() {
        @Override
        public void apply(final SingleResultCallback<Success> callback) {
            wrapped.close(voidToSuccessCallback(callback));
        }
    }), observableAdapter);
}
 

@Override
public Observable<Success> abort() {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<Success>>() {
        @Override
        public void apply(final SingleResultCallback<Success> callback) {
            wrapped.abort(voidToSuccessCallback(callback));
        }
    }), observableAdapter);
}
 

@Override
public Observable<Long> count(final Bson filter, final CountOptions options) {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<Long>>() {
        @Override
        public void apply(final SingleResultCallback<Long> callback) {
            wrapped.count(filter, options, callback);
        }
    }), observableAdapter);
}
 

@Override
public Observable<BulkWriteResult> bulkWrite(final List<? extends WriteModel<? extends TDocument>> requests,
                                            final BulkWriteOptions options) {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<BulkWriteResult>>() {
        @Override
        public void apply(final SingleResultCallback<BulkWriteResult> callback) {
            wrapped.bulkWrite(requests, options, callback);
        }
    }), observableAdapter);
}
 

@Override
public Observable<Success> insertOne(final TDocument document) {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<Success>>() {
        @Override
        public void apply(final SingleResultCallback<Success> callback) {
            wrapped.insertOne(document, voidToSuccessCallback(callback));
        }
    }), observableAdapter);
}
 

@Override
public Observable<Success> insertOne(final TDocument document, final InsertOneOptions options) {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<Success>>() {
        @Override
        public void apply(final SingleResultCallback<Success> callback) {
            wrapped.insertOne(document, options, voidToSuccessCallback(callback));
        }
    }), observableAdapter);
}
 

@Override
public Observable<Success> insertMany(final List<? extends TDocument> documents, final InsertManyOptions options) {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<Success>>() {
        @Override
        public void apply(final SingleResultCallback<Success> callback) {
            wrapped.insertMany(documents, options, voidToSuccessCallback(callback));
        }
    }), observableAdapter);
}
 

@Override
public Observable<DeleteResult> deleteOne(final Bson filter) {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<DeleteResult>>() {
        @Override
        public void apply(final SingleResultCallback<DeleteResult> callback) {
            wrapped.deleteOne(filter, callback);
        }
    }), observableAdapter);
}
 

@Override
public Observable<DeleteResult> deleteOne(final Bson filter, final DeleteOptions options) {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<DeleteResult>>() {
        @Override
        public void apply(final SingleResultCallback<DeleteResult> callback) {
            wrapped.deleteOne(filter, options, callback);
        }
    }), observableAdapter);
}
 

@Override
public Observable<DeleteResult> deleteMany(final Bson filter) {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<DeleteResult>>() {
        @Override
        public void apply(final SingleResultCallback<DeleteResult> callback) {
            wrapped.deleteMany(filter, callback);
        }
    }), observableAdapter);
}
 

@Override
public Observable<DeleteResult> deleteMany(final Bson filter, final DeleteOptions options) {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<DeleteResult>>() {
        @Override
        public void apply(final SingleResultCallback<DeleteResult> callback) {
            wrapped.deleteMany(filter, options, callback);
        }
    }), observableAdapter);
}
 

@Override
public Observable<UpdateResult> replaceOne(final Bson filter, final TDocument replacement, final UpdateOptions options) {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<UpdateResult>>() {
        @Override
        public void apply(final SingleResultCallback<UpdateResult> callback) {
            wrapped.replaceOne(filter, replacement, options, callback);
        }
    }), observableAdapter);
}
 

@Override
public Observable<UpdateResult> updateOne(final Bson filter, final Bson update, final UpdateOptions options) {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<UpdateResult>>() {
        @Override
        public void apply(final SingleResultCallback<UpdateResult> callback) {
            wrapped.updateOne(filter, update, options, callback);
        }
    }), observableAdapter);
}
 

@Override
public Observable<UpdateResult> updateMany(final Bson filter, final Bson update, final UpdateOptions options) {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<UpdateResult>>() {
        @Override
        public void apply(final SingleResultCallback<UpdateResult> callback) {
            wrapped.updateMany(filter, update, options, callback);
        }
    }), observableAdapter);
}
 

@Override
public Observable<TDocument> findOneAndDelete(final Bson filter, final FindOneAndDeleteOptions options) {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<TDocument>>() {
        @Override
        public void apply(final SingleResultCallback<TDocument> callback) {
            wrapped.findOneAndDelete(filter, options, callback);
        }
    }), observableAdapter);
}
 

@Override
public Observable<TDocument> findOneAndReplace(final Bson filter, final TDocument replacement, final FindOneAndReplaceOptions options) {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<TDocument>>() {
        @Override
        public void apply(final SingleResultCallback<TDocument> callback) {
            wrapped.findOneAndReplace(filter, replacement, options, callback);
        }
    }), observableAdapter);
}
 

@Override
public Observable<TDocument> findOneAndUpdate(final Bson filter, final Bson update, final FindOneAndUpdateOptions options) {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<TDocument>>() {
        @Override
        public void apply(final SingleResultCallback<TDocument> callback) {
            wrapped.findOneAndUpdate(filter, update, options, callback);
        }
    }), observableAdapter);
}
 

@Override
public Observable<Success> drop() {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<Success>>() {
        @Override
        public void apply(final SingleResultCallback<Success> callback) {
            wrapped.drop(voidToSuccessCallback(callback));
        }
    }), observableAdapter);
}
 

@Override
public Observable<String> createIndex(final Bson key, final IndexOptions options) {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<String>>() {
        @Override
        public void apply(final SingleResultCallback<String> callback) {
            wrapped.createIndex(key, options, callback);
        }
    }), observableAdapter);
}
 

@Override
public Observable<String> createIndexes(final List<IndexModel> indexes) {
    return RxObservables.create(Observables.observeAndFlatten(new Block<SingleResultCallback<List<String>>>() {
        @Override
        public void apply(final SingleResultCallback<List<String>> callback) {
            wrapped.createIndexes(indexes, callback);
        }
    }), observableAdapter);
}
 

@Override
public Observable<Success> dropIndex(final String indexName) {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<Success>>() {
        @Override
        public void apply(final SingleResultCallback<Success> callback) {
            wrapped.dropIndex(indexName, voidToSuccessCallback(callback));
        }
    }), observableAdapter);
}
 

@Override
public Observable<Success> dropIndex(final Bson keys) {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<Success>>() {
        @Override
        public void apply(final SingleResultCallback<Success> callback) {
            wrapped.dropIndex(keys, voidToSuccessCallback(callback));
        }
    }), observableAdapter);
}
 

@Override
public Observable<Success> renameCollection(final MongoNamespace newCollectionNamespace, final RenameCollectionOptions options) {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<Success>>() {
        @Override
        public void apply(final SingleResultCallback<Success> callback) {
            wrapped.renameCollection(newCollectionNamespace, options, voidToSuccessCallback(callback));
        }
    }), observableAdapter);
}
 

@Override
public Observable<GridFSFile> first() {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<GridFSFile>>(){
        @Override
        public void apply(final SingleResultCallback<GridFSFile> callback) {
            wrapped.first(callback);
        }
    }), observableAdapter);
}
 

@Override
public <TResult> Observable<TResult> runCommand(final Bson command, final Class<TResult> clazz) {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<TResult>>() {
        @Override
        public void apply(final SingleResultCallback<TResult> callback) {
            wrapped.runCommand(command, clazz, callback);
        }
    }), observableAdapter);
}
 

@Override
public <TResult> Observable<TResult> runCommand(final Bson command, final ReadPreference readPreference,
                                                   final Class<TResult> clazz) {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<TResult>>() {
        @Override
        public void apply(final SingleResultCallback<TResult> callback) {
            wrapped.runCommand(command, readPreference, clazz, callback);
        }
    }), observableAdapter);
}
 

@Override
public Observable<Success> drop() {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<Success>>() {
        @Override
        public void apply(final SingleResultCallback<Success> callback) {
            wrapped.drop(voidToSuccessCallback(callback));
        }
    }), observableAdapter);
}