<commit_before>var map = require("continuable/map")
var cache = require("continuable-cache")
var through = require("through")

var mapAsync = require("./lib/map-async")

/*  type Cursor := {
        (Callback<mongodb/Cursor>) => void,
        toArray: Continuable<Array<Value>>,
        nextObject: Continuable<Value | null>,
        stream: () => Stream
    }

    createCursor := (Collection) =>
        (selector: Object, options: Object?) => Cursor
*/
module.exports = createCursor

function createCursor(collection) {
    return function (selector, options) {
        var cursor = cache(map(function (collection) {
            return collection.find(selector, options || {})
        })(collection))

        cursor.toArray = mapAsync(function (cursor, cb) {
            cursor.toArray(cb)
        })(cursor)

        cursor.nextObject = mapAsync(function (cursor, cb) {
            cursor.nextObject(cb)
        })(cursor)

        cursor.stream = createStream(cursor)

        return cursor
    }
}

function createStream(cursor) {
    return function () {
        var stream = through()

        cursor(function (err, cursor) {
            var cursorStream = cursor.stream()

            cursorStream.pipe(stream)
        })

        return stream
    }
}<commit_msg>can destroy cursor stream<commit_after>var map = require("continuable/map")
var cache = require("continuable-cache")
var through = require("through")

var mapAsync = require("./lib/map-async")

/*  type Cursor := {
        (Callback<mongodb/Cursor>) => void,
        toArray: Continuable<Array<Value>>,
        nextObject: Continuable<Value | null>,
        stream: () => Stream
    }

    createCursor := (Collection) =>
        (selector: Object, options: Object?) => Cursor
*/
module.exports = createCursor

function createCursor(collection) {
    return function (selector, options) {
        var cursor = cache(map(function (collection) {
            return collection.find(selector, options || {})
        })(collection))

        cursor.toArray = mapAsync(function (cursor, cb) {
            cursor.toArray(cb)
        })(cursor)

        cursor.nextObject = mapAsync(function (cursor, cb) {
            cursor.nextObject(cb)
        })(cursor)

        cursor.stream = createStream(cursor)

        return cursor
    }
}

function createStream(cursor) {
    return function () {
        var stream = through()

        cursor(function (err, cursor) {
            if (err) {
                return stream.emit("error", err)
            }

            var cursorStream = cursor.stream()

            stream.once("close", function () {
                cursorStream.destroy()
            })

            cursorStream.pipe(stream)
        })

        return stream
    }
}
{'fraction_non_alphanumeric': 0.1262277191706075, 'fraction_numerical': 0.0, 'mean_word_length': 2.289473684210526, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7261872', 'n_tokens_mistral': 748, 'n_tokens_neox': 720, 'n_words': 167}