<commit_before>var redis = require("redis");

/*
This function needs to take a filter. When events come in on the filter,
publish them to the queue via the Redis URL. Every time we publish an event,
if the block number has changed since the previous block we record the block
number in redis.

When we first start up, if the process we need to start in "resumption mode".
In resumption mode, our filter should go from the block stored in redis to
"latest". Resumption mode must queue up all of the messages it processes until
resumption mode finishes. When resumption mode starts, a timeout should be set
while messages are being processed. Each time a message is processed, it should
reset the timeout. When the timeout executes we can assume that resumption
has finished, so the queue should be flushed, then the block number should be
updated.
*/

var publishers = require("./publishers");

// TODO: Make sure filter will be recreated if the backend RPC server changes

module.exports = function(redisClient, notificationChannel, filterCreator, web3, transform){
    var blockKey = notificationChannel + "::blocknumber";
    channel = publishers.FromURI(redisClient, notificationChannel);
    return new Promise((resolve, reject) => {
        redisClient.get(blockKey, function(err, data) {
            if(err){
                reject(err);
            } else {
                resolve(data);
            }
        })
    }).then((blockNumber) => {
        var resuming;
        var resumptionTimeout;
        if(blockNumber === null) {
            blockNumber = "latest";
            resuming = false;
        } else {
            // blockNumber from Redis is the last completed block, so we
            // want to resume from the next block.
            blockNumber = parseInt(blockNumber) + 1;
            resuming = true;
        }
        var watcher = filterCreator({fromBlock: blockNumber, toBlock: "latest"});
        lastBlockNumber = blockNumber;
        watcher.watch((err, data) => {
            if(err) {
                console.log(err);
                var stack = new Error().stack
                console.log( stack )
                // If we get an error, exit. Docker should restart the process,
                // and it can go through the resumption process.
                //
                // We may need to recover more gracefully, but we'll try this
                // for now.
                process.exit(1);
            }
            if(resuming){
                clearTimeout(resumptionTimeout);
                resumptionTimeout = setTimeout(function(){
                    web3.eth.getBlockNumber((currentBlock) => {
                        resuming = false;
                        channel.FlushQueue();
                        redisClient.set(blockKey, currentBlock);
                        lastBlockNumber = currentBlock;
                    })
                }, 5000);
                channel.QueueMessage(JSON.stringify(transform(data)));
            } else {
                web3.eth.getBlockNumber((currentBlock) => {
                    var payload = JSON.stringify(transform(data));
                    console.log(`Block ${currentBlock} - published '${payload}'`);
                    channel.Publish(payload);
                    if(lastBlockNumber != currentBlock) {
                        redisClient.set(blockKey, currentBlock);
                        lastBlockNumber = currentBlock;
                    }
                });
            }

        });
    }).catch((err) => {
        console.log("Failed to get initial block number: " + err);
    })
}<commit_msg>First argument of getBlockNumber should be error<commit_after>var redis = require("redis");

/*
This function needs to take a filter. When events come in on the filter,
publish them to the queue via the Redis URL. Every time we publish an event,
if the block number has changed since the previous block we record the block
number in redis.

When we first start up, if the process we need to start in "resumption mode".
In resumption mode, our filter should go from the block stored in redis to
"latest". Resumption mode must queue up all of the messages it processes until
resumption mode finishes. When resumption mode starts, a timeout should be set
while messages are being processed. Each time a message is processed, it should
reset the timeout. When the timeout executes we can assume that resumption
has finished, so the queue should be flushed, then the block number should be
updated.
*/

var publishers = require("./publishers");

// TODO: Make sure filter will be recreated if the backend RPC server changes

module.exports = function(redisClient, notificationChannel, filterCreator, web3, transform){
    var blockKey = notificationChannel + "::blocknumber";
    channel = publishers.FromURI(redisClient, notificationChannel);
    return new Promise((resolve, reject) => {
        redisClient.get(blockKey, function(err, data) {
            if(err){
                reject(err);
            } else {
                resolve(data);
            }
        })
    }).then((blockNumber) => {
        var resuming;
        var resumptionTimeout;
        if(blockNumber === null) {
            blockNumber = "latest";
            resuming = false;
        } else {
            // blockNumber from Redis is the last completed block, so we
            // want to resume from the next block.
            blockNumber = parseInt(blockNumber) + 1;
            resuming = true;
        }
        var watcher = filterCreator({fromBlock: blockNumber, toBlock: "latest"});
        lastBlockNumber = blockNumber;
        watcher.watch((err, data) => {
            if(err) {
                console.log(err);
                var stack = new Error().stack
                console.log( stack )
                // If we get an error, exit. Docker should restart the process,
                // and it can go through the resumption process.
                //
                // We may need to recover more gracefully, but we'll try this
                // for now.
                process.exit(1);
            }
            if(resuming){
                clearTimeout(resumptionTimeout);
                resumptionTimeout = setTimeout(function(){
                    web3.eth.getBlockNumber((err, currentBlock) => {
                        resuming = false;
                        channel.FlushQueue();
                        redisClient.set(blockKey, currentBlock);
                        lastBlockNumber = currentBlock;
                    })
                }, 5000);
                channel.QueueMessage(JSON.stringify(transform(data)));
            } else {
                web3.eth.getBlockNumber((err, currentBlock) => {
                    var payload = JSON.stringify(transform(data));
                    console.log(`Block ${currentBlock} - published '${payload}'`);
                    channel.Publish(payload);
                    if(lastBlockNumber != currentBlock && currentBlock != null) {
                        redisClient.set(blockKey, currentBlock);
                        lastBlockNumber = currentBlock;
                    }
                });
            }

        });
    }).catch((err) => {
        console.log("Failed to get initial block number: " + err);
    })
}
{'fraction_non_alphanumeric': 0.08678986995208761, 'fraction_numerical': 0.0024640657084188913, 'mean_word_length': 1.8372815533980582, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 15, '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': '7005834', 'n_tokens_mistral': 1794, 'n_tokens_neox': 1690, 'n_words': 622}