public SourceReader createReader()

in flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/source/MongoSource.java [136:157]

• 19 lines of code
• 1 McCabe index (conditional complexity)

    public SourceReader<OUT, MongoSourceSplit> createReader(SourceReaderContext readerContext) {
        FutureCompletingBlockingQueue<RecordsWithSplitIds<BsonDocument>> elementsQueue =
                new FutureCompletingBlockingQueue<>();

        MongoSourceReaderContext mongoReaderContext =
                new MongoSourceReaderContext(readerContext, limit);

        Supplier<SplitReader<BsonDocument, MongoSourceSplit>> splitReaderSupplier =
                () ->
                        new MongoScanSourceSplitReader(
                                connectionOptions,
                                readOptions,
                                projectedFields,
                                filter,
                                mongoReaderContext);

        return new MongoSourceReader<>(
                elementsQueue,
                splitReaderSupplier,
                new MongoRecordEmitter<>(deserializationSchema),
                mongoReaderContext);
    }