<commit_before>var Failover = require("./failover");
var BufferReadable = require("./buffer_readable");
var stream = require("stream");

function Message(headers, body, callback){
    this.headers = headers;
    this.body = body;
    this.callback = callback || function(){};
    this.receipt = false;
}

Message.prototype.getReadable = function(){
    var readable = (typeof this.body === "function" ? this.body() : this.body);
    if(!(readable instanceof stream.Readable)){
        var buffer = readable;
        if(!(buffer instanceof Buffer)){
            buffer = new Buffer(readable);
        }
        readable = new BufferReadable(buffer);
    }
    return readable;
};

Message.prototype.send = function(client, callback){
    
    var self = this;
    
    var output = client.send(this.headers, {
        onReceipt: function(){   
            
            if(self.receipt){
                return;
            }
            
            self.receipt = true;
            
            if(callback){
                callback();
            }
            
            self.callback(null);
        }
    });
    
    var input = this.getReadable();
    
    input.pipe(output);
};

function SendPipeline(failover){
    this._failover = failover;
    this._messages = [];
    this._client = null;
    this._connecting = false;
    this._cancelled = false;
}

SendPipeline.prototype.send = function(headers, body, callback){
    
    var message = new Message(headers, body, callback);
    
    this._messages.push(message);
    
    if(this.client){
        message.send(this._client, this._disconnect.bind(this));
    }
    else{
        this._connect();
    }
    
    return this;
};

SendPipeline.prototype._connect = function(){

    if(this._connecting){
        return;
    }
    
    this._connecting = true;
    
    var self = this;
    
    var connection = this._failover.connect(function(error, client){
        
        self._connecting = false;
        
        if(self._cancelled){
            client.destroy();
            return;
        }
        
        if(error){<commit_msg>Fix typo. Make send receipt optional<commit_after>var Failover = require("./failover");
var BufferReadable = require("./buffer_readable");
var stream = require("stream");

function Message(headers, body, requestReceipt, callback){
    this.headers = headers;
    this.body = body;
    this.callback = callback || function(){};
    this.requestReceipt = requestReceipt;
    this.receipt = false;
}

Message.prototype.getReadable = function(){
    var readable = (typeof this.body === "function" ? this.body() : this.body);
    if(!(readable instanceof stream.Readable)){
        var buffer = readable;
        if(!(buffer instanceof Buffer)){
            buffer = new Buffer(readable);
        }
        readable = new BufferReadable(buffer);
    }
    return readable;
};

Message.prototype.send = function(client, callback){
    
    var self = this;
    
    var options = {};
    
    if(this.requestReceipt){
        
        options["onReceipt"] = function(){   
            
            if(self.receipt){
                return;
            }
            
            self.receipt = true;
            
            if(callback){
                callback();
            }
            
            self.callback(null);
        };
    }
    
    var output = client.send(this.headers, options);
    
    var input = this.getReadable();
    input.pipe(output);
    
    if(!this.requestReceipt){
        output.once("finish", callback);    
    }
    
    return output;
};

function SendPipeline(failover){
    this._failover = failover;
    this._messages = [];
    this._client = null;
    this._connecting = false;
    this._cancelled = false;
}

SendPipeline.prototype.send = function(headers, body, callback){
    
    var message = new Message(headers, body, true, callback);
    
    this._messages.push(message);
    
    if(this._client){
        message.send(this._client, this._disconnect.bind(this));
    }
    else{
        this._connect();
    }
    
    return this;
};

SendPipeline.prototype._connect = function(){

    if(this._connecting){
        return;
    }
    
    this._connecting = true;
    
    var self = this;
    
    var connection = this._failover.connect(function(error, client){
        
        self._connecting = false;
        
        if(self._cancelled){
            client.destroy();
            return;
        }
        
        if(error){
{'fraction_non_alphanumeric': 0.12210338680926916, 'fraction_numerical': 0.0, 'mean_word_length': 1.9301566579634464, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '7102039', 'n_tokens_mistral': 1283, 'n_tokens_neox': 1250, 'n_words': 253}