{"action":"created","issue":{"url":"https://api.github.com/repos/luin/ioredis/issues/20","labels_url":"https://api.github.com/repos/luin/ioredis/issues/20/labels{/name}","comments_url":"https://api.github.com/repos/luin/ioredis/issues/20/comments","events_url":"https://api.github.com/repos/luin/ioredis/issues/20/events","html_url":"https://github.com/luin/ioredis/pull/20","id":72594114,"number":20,"title":"Adds handling of already buffered data","user":{"login":"AVVS","id":1713617,"avatar_url":"https://avatars.githubusercontent.com/u/1713617?v=3","gravatar_id":"","url":"https://api.github.com/users/AVVS","html_url":"https://github.com/AVVS","followers_url":"https://api.github.com/users/AVVS/followers","following_url":"https://api.github.com/users/AVVS/following{/other_user}","gists_url":"https://api.github.com/users/AVVS/gists{/gist_id}","starred_url":"https://api.github.com/users/AVVS/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/AVVS/subscriptions","organizations_url":"https://api.github.com/users/AVVS/orgs","repos_url":"https://api.github.com/users/AVVS/repos","events_url":"https://api.github.com/users/AVVS/events{/privacy}","received_events_url":"https://api.github.com/users/AVVS/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":2,"created_at":"2015-05-02T06:06:28Z","updated_at":"2015-05-02T06:28:40Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/luin/ioredis/pulls/20","html_url":"https://github.com/luin/ioredis/pull/20","diff_url":"https://github.com/luin/ioredis/pull/20.diff","patch_url":"https://github.com/luin/ioredis/pull/20.patch"},"body":"Before that using already buffered data lead to max call stack errors"},"comment":{"url":"https://api.github.com/repos/luin/ioredis/issues/comments/98318302","html_url":"https://github.com/luin/ioredis/pull/20#issuecomment-98318302","issue_url":"https://api.github.com/repos/luin/ioredis/issues/20","id":98318302,"user":{"login":"AVVS","id":1713617,"avatar_url":"https://avatars.githubusercontent.com/u/1713617?v=3","gravatar_id":"","url":"https://api.github.com/users/AVVS","html_url":"https://github.com/AVVS","followers_url":"https://api.github.com/users/AVVS/followers","following_url":"https://api.github.com/users/AVVS/following{/other_user}","gists_url":"https://api.github.com/users/AVVS/gists{/gist_id}","starred_url":"https://api.github.com/users/AVVS/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/AVVS/subscriptions","organizations_url":"https://api.github.com/users/AVVS/orgs","repos_url":"https://api.github.com/users/AVVS/repos","events_url":"https://api.github.com/users/AVVS/events{/privacy}","received_events_url":"https://api.github.com/users/AVVS/received_events","type":"User","site_admin":false},"created_at":"2015-05-02T06:28:40Z","updated_at":"2015-05-02T06:28:40Z","body":"I'm currently writing redis backend for https://github.com/pgte/abstract-skiff-persistence, which is basically a copy of https://github.com/pgte/skiff-level/blob/master/lib/index.js with a different backend, and without the proposed change `pipeline.exec(callback);` would crash when not using `hiredis` with max call stack error, and when using hiredis it would crash with smth saying that it can't read `.name` of the command or smth. I'm pretty sure it tries to parse the buffer, that was encoded as a msgpack and fails somehow during serialization/deserialization \r\n\r\nTo simulate this you can run the tests as proposed in https://github.com/pgte/abstract-skiff-persistence - basic payload is a bit incorrect - `:` instead of `,` - but otherwise it's fine to test.\r\n\r\n```js\r\nimport msgpack5 from 'msgpack5';\r\nimport SkiffPersistence from 'abstract-skiff-persistence';\r\nimport _ from 'lodash';\r\nimport async from 'neo-async';\r\n\r\nlet msgpack = msgpack5();\r\n\r\nfunction encode(val) {\r\n  return msgpack.encode(val).slice();\r\n}\r\n\r\nfunction decode(buf) {\r\n  return msgpack.decode(buf);\r\n}\r\n\r\nclass SkiffRedis extends SkiffPersistence {\r\n\r\n    constructor(options={}) {\r\n\r\n        super(options);\r\n\r\n        let { redis, namespace } = options;\r\n\r\n        // in case we use redis cluster, make sure that namespace is hashed\r\n        this._namespace = namespace || '{skiff-redis}';\r\n\r\n        this._redis = redis;\r\n\r\n        this._redis.on('error', this._onError);\r\n\r\n        this.nodes = {};\r\n    }\r\n\r\n    _onError(err) {\r\n        this.emit('error', err);\r\n    }\r\n\r\n    _key(...parts) {\r\n        if (parts.length === 0) {\r\n            throw new Error('parts must include at least one value');\r\n        }\r\n\r\n        return this._namespace + parts.join('~');\r\n    }\r\n\r\n    _saveMeta(nodeId, state, callback) {\r\n        let redis = this._redis;\r\n        let pipeline = redis.pipeline();\r\n        let log = state.log;\r\n        let logEntries = log.entries;\r\n\r\n        // do not enter hashtable mode\r\n        state.log = _.without(log, ['entries']);\r\n\r\n        pipeline.set(this._key(nodeId, 'meta'), encode(state));\r\n\r\n        let minLogIndex = logEntries.length && logEntries[0].index || 0;\r\n        let maxLogIndex = logEntries.length && logEntries[logEntries.length - 1].index || Infinity;\r\n        let logNamespace = this._key(nodeId, 'logs');\r\n        let cursor = 0;\r\n        let maxReadIndex = 0;\r\n\r\n        async.doUntil(\r\n            function zscanRedis(next) {\r\n                redis.zscanBuffer(logNamespace, cursor, 'count', 10, function zscanResponse(err, response) {\r\n                    if (err) {\r\n                        return next(err);\r\n                    }\r\n\r\n                    // update cursor\r\n                    cursor = parseInt(response[0], 10);\r\n\r\n                    // process log entries\r\n                    let entries = response[1];\r\n                    for (let i = 0, l = entries.length; i < l; i += 2) {\r\n                        let index = parseInt(entries[i + 1], 10);\r\n                        let entry = entries[i];\r\n\r\n                        if (index > maxReadIndex) {\r\n                            maxReadIndex = index;\r\n                        }\r\n\r\n                        if (index < minLogIndex || index > maxLogIndex) {\r\n                            pipeline.zrem(logNamespace, entry);\r\n                        } else {\r\n                            let correspondingNewEntry = logEntries[index - minLogIndex];\r\n                            let decodedEntry = decode(entries[i]);\r\n                            if (correspondingNewEntry.uuid !== decodedEntry.uuid) {\r\n                                pipeline.zrem(entry);\r\n                                pipeline.zadd(logNamespace, correspondingNewEntry.index, encode(correspondingNewEntry));\r\n                            }\r\n                        }\r\n\r\n                    }\r\n\r\n                    next();\r\n\r\n                });\r\n            },\r\n\r\n            function isIterationComplete() {\r\n                return cursor === 0;\r\n            },\r\n\r\n            function saveMeta(err) {\r\n                if (err) {\r\n                    return callback(err);\r\n                }\r\n\r\n                if (maxReadIndex < maxLogIndex) {\r\n                    logEntries.slice(maxReadIndex - minLogIndex + 1).forEach(function (entry) {\r\n                        pipeline.zadd(logNamespace, entry.index, encode(entry));\r\n                    });\r\n                }\r\n\r\n                pipeline.exec(callback);\r\n            }\r\n\r\n        );\r\n    }\r\n\r\n    _loadMeta(nodeId, callback) {\r\n\r\n        this._redis\r\n            .pipeline()\r\n            .getBuffer(this._key(nodeId, 'meta'))\r\n            .zrangebyscoreBuffer(this._key(nodeId, 'logs'), '-inf', '+inf')\r\n            .exec(function (___, results) {\r\n                // err is always null\r\n\r\n                let [err, meta] = results[0];\r\n                let [logerr, logEntries] = results[1];\r\n\r\n                if (err || logerr) {\r\n                    return callback(err || logerr);\r\n                }\r\n\r\n                if (meta) {\r\n                    meta = decode(meta);\r\n                    meta.log.entries = logEntries.map(decode);\r\n                }\r\n\r\n                callback(null, meta);\r\n\r\n            });\r\n\r\n    }\r\n\r\n    _applyCommand(nodeId, commitIndex, command, callback) {\r\n\r\n    }\r\n\r\n    _lastAppliedCommitIndex(nodeId, callback) {\r\n\r\n    }\r\n\r\n    _saveCommitIndex(nodeId, commitIndex, callback) {\r\n\r\n    }\r\n\r\n    _createReadStream(nodeId) {\r\n\r\n    }\r\n\r\n    _createWriteStream(nodeId) {\r\n\r\n    }\r\n\r\n    _removeAllState(nodeId, callback) {\r\n\r\n    }\r\n\r\n    _close(callback) {\r\n        this._redis.disconnect();\r\n        this._redis.removeListener('error', this._onError);\r\n        callback();\r\n    }\r\n\r\n}\r\n\r\n\r\nexport default SkiffRedis;\r\n\r\n```"}}