<commit_before>// stream.js
//
// A (potentially very long) stream of activities
//
// Copyright 2011,2012 StatusNet Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

var databank = require('databank'),
    _ = require('underscore'),
    DatabankObject = databank.DatabankObject,
    dateFormat = require('dateformat'),
    bcrypt  = require('bcrypt'),
    Step = require('step'),
    Person = require('./person').Person,
    Activity = require('./activity').Activity,
    IDMaker = require('../idmaker').IDMaker,
    NoSuchThingError = databank.NoSuchThingError;

var Stream = DatabankObject.subClass('stream');

exports.Stream = Stream;

Stream.SOFT_LIMIT = 1000;
Stream.HARD_LIMIT = 2000;

Stream.beforeCreate = function(props, callback) {

    var bank = Stream.bank(),
        stream = null;

    if (!props.name) {
        callback(new Error('Gotta have a name'), null);
    }

    Step(
        function() {
            bank.create('streamcount', props.name, 0, this.parallel());
            bank.create('streamsegments', props.name, [], this.parallel());
        },
        function(err, count, segments) {
            if (err) {
                callback(err, null);
            } else {
                callback(null, props);
            }
        }
    );
};

Stream.prototype.afterCreate = function(callback) {
    this.newSegment(callback);
};

// put something in the stream

var randBetween = function(min, max) {
    var diff = max - min + 1; 
    return Math.floor((Math.random() * diff) + min);
};

Stream.prototype.deliver = function(activity, callback) {

    var stream = this,
        bank = Stream.bank(),
        current = null;

    Step(
        function() {
            bank.item('streamsegments', stream.name, 0, this);
        },
        function(err, id) {
            if (err) throw err;
            current = id;
            bank.read('streamsegmentcount', current, this);
        },
        function(err, cnt) {

            if (err) throw err;

            // Once we hit the soft limit, we start thinking about 
            // a new segment. To avoid conflicts, a bit, we do it at a
            // random point between soft and hard limit. If we actually
            // hit the hard limit, force it.

            if (cnt > Stream.SOFT_LIMIT &&
                (cnt > Stream.HARD_LIMIT ||
                 randBetween(0, Stream.HARD_LIMIT - Stream.SOFT_LIMIT) === 0)) {
                stream.newSegment(this);
            } else {
                this(null, current);
            }
        },
        function(err, segmentId) {
            if (err) throw err;
            bank.prepend('streamsegment', segmentId, activity.id, this.parallel());
            bank.incr('streamsegmentcount', segmentId, this.parallel());
            bank.incr('streamcount', stream.name, this.parallel());
        },
        function(err) {
            if (err) {
                callback(err);
            } else {
                callback(null);
            }
        }
    );
};

Stream.prototype.newSegment = function(callback) {

    var bank = Stream.bank(),
        stream = this,
        id = stream.name + ':stream:' + IDMaker.makeID();

    Step(
        function() {
            bank.create('streamsegmentcount', id, 0, this.parallel());
            bank.create('streamsegment', id, [], this.parallel());
        },
        function(err, cnt, segment) {
            if (err) throw err;
            bank.prepend('streamsegments', stream.name, id, this);
        },
        function(err, segments) {
            if (err) {
                callback(err, null);
            } else {
                callback(err, id);
            }
        }
    );
};

Stream.prototype.getActivities = function(start, end, callback) {

    var bank = Stream.bank(),
        stream = this,
        getMore = function getMore(segments, start, end, callback) {

            var tip;

            if (segments.length === 0) {
                callback(null, []);
                return;
            }

            tip = segments.shift();

            Step(
                function() {
                    bank.read('streamsegmentcount', tip, this);
                },
                function(err, tipcount) {
                    var group = this.group();
                    if (err) throw err;
                    if (start < tipcount) {
                        bank.slice('streamsegment',
                                   tip,
                                   start,
                                   Math.min(end, tipcount),
                                   group());
                    }
                    if (end > tipcount) {
                        if (segments.length > 0) {
                            getMore(segments,
                                    Math.max(start - tipcount, 0),
                                    end - tipcount,
                                    group());
                        } else { // Asking for more than we have
                            // Need to trigger the rest
                            group()(null, []);
                        }
                    }
                },
                function(err, parts) {
                    if (err) {
                        callback(err, null);
                    } else { 
                        callback(null, (parts.length === 1) ? parts[0] : parts[0].concat(parts[1]));
                    }
                }
            );
        };

    Step(
        function() {
            // XXX: maybe just take slice from [0, end/HARD_LIMIT)
            bank.read('streamsegments', stream.name, this);
        },
        function(err, segments) {
            if (err) throw err;
            getMore(segments, start, end, this);
        },
        function(err, ids) {
            if (err) throw err;
            if (ids.length === 0) {
                callback(null, []);
            } else {
                bank.readAll('activity', ids, this);
            }
        },
        function(err, activityMap) {
            var activities = [], id, group;
            if (err) {
                callback(err, null);
            } else {
                for (id in activityMap) {
                    activities.push(new Activity(activityMap[id]));
                }
                activities.sort(function(a, b) {  
                    if (a.published > b.published) {
                        return -1;  
                    } else if (a.published < b.published) {
                        return 1;  
                    } else {
                        return 0;  
                    }
                });
                callback(null, activities);
            }
        }
    );
};

Stream.prototype.count = function(callback) {
    var bank = Stream.bank();
    
    bank.read('streamcount', this.name, callback);
};

Stream.schema = {
    'stream': {'pkey': 'name'},
    'streamcount': {'pkey': 'name'},
    'streamsegments': {'pkey': 'name'},

    'streamsegment': {'pkey': 'id'},
    'streamsegmentcount': {'pkey': 'id'}
};
<commit_msg>Check input parameters for getActivities()<commit_after>// stream.js
//
// A (potentially very long) stream of activities
//
// Copyright 2011,2012 StatusNet Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

var databank = require('databank'),
    _ = require('underscore'),
    DatabankObject = databank.DatabankObject,
    dateFormat = require('dateformat'),
    bcrypt  = require('bcrypt'),
    Step = require('step'),
    Person = require('./person').Person,
    Activity = require('./activity').Activity,
    IDMaker = require('../idmaker').IDMaker,
    NoSuchThingError = databank.NoSuchThingError;

var Stream = DatabankObject.subClass('stream');

exports.Stream = Stream;

Stream.SOFT_LIMIT = 1000;
Stream.HARD_LIMIT = 2000;

Stream.beforeCreate = function(props, callback) {

    var bank = Stream.bank(),
        stream = null;

    if (!props.name) {
        callback(new Error('Gotta have a name'), null);
    }

    Step(
        function() {
            bank.create('streamcount', props.name, 0, this.parallel());
            bank.create('streamsegments', props.name, [], this.parallel());
        },
        function(err, count, segments) {
            if (err) {
                callback(err, null);
            } else {
                callback(null, props);
            }
        }
    );
};

Stream.prototype.afterCreate = function(callback) {
    this.newSegment(callback);
};

// put something in the stream

var randBetween = function(min, max) {
    var diff = max - min + 1; 
    return Math.floor((Math.random() * diff) + min);
};

Stream.prototype.deliver = function(activity, callback) {

    var stream = this,
        bank = Stream.bank(),
        current = null;

    Step(
        function() {
            bank.item('streamsegments', stream.name, 0, this);
        },
        function(err, id) {
            if (err) throw err;
            current = id;
            bank.read('streamsegmentcount', current, this);
        },
        function(err, cnt) {

            if (err) throw err;

            // Once we hit the soft limit, we start thinking about 
            // a new segment. To avoid conflicts, a bit, we do it at a
            // random point between soft and hard limit. If we actually
            // hit the hard limit, force it.

            if (cnt > Stream.SOFT_LIMIT &&
                (cnt > Stream.HARD_LIMIT ||
                 randBetween(0, Stream.HARD_LIMIT - Stream.SOFT_LIMIT) === 0)) {
                stream.newSegment(this);
            } else {
                this(null, current);
            }
        },
        function(err, segmentId) {
            if (err) throw err;
            bank.prepend('streamsegment', segmentId, activity.id, this.parallel());
            bank.incr('streamsegmentcount', segmentId, this.parallel());
            bank.incr('streamcount', stream.name, this.parallel());
        },
        function(err) {
            if (err) {
                callback(err);
            } else {
                callback(null);
            }
        }
    );
};

Stream.prototype.newSegment = function(callback) {

    var bank = Stream.bank(),
        stream = this,
        id = stream.name + ':stream:' + IDMaker.makeID();

    Step(
        function() {
            bank.create('streamsegmentcount', id, 0, this.parallel());
            bank.create('streamsegment', id, [], this.parallel());
        },
        function(err, cnt, segment) {
            if (err) throw err;
            bank.prepend('streamsegments', stream.name, id, this);
        },
        function(err, segments) {
            if (err) {
                callback(err, null);
            } else {
                callback(err, id);
            }
        }
    );
};

Stream.prototype.getActivities = function(start, end, callback) {

    var bank = Stream.bank(),
        stream = this,
        getMore = function getMore(segments, start, end, callback) {

            var tip;

            if (segments.length === 0) {
                callback(null, []);
                return;
            }

            tip = segments.shift();

            Step(
                function() {
                    bank.read('streamsegmentcount', tip, this);
                },
                function(err, tipcount) {
                    var group = this.group();
                    if (err) throw err;
                    if (start < tipcount) {
                        bank.slice('streamsegment',
                                   tip,
                                   start,
                                   Math.min(end, tipcount),
                                   group());
                    }
                    if (end > tipcount) {
                        if (segments.length > 0) {
                            getMore(segments,
                                    Math.max(start - tipcount, 0),
                                    end - tipcount,
                                    group());
                        } else { // Asking for more than we have
                            // Need to trigger the rest
                            group()(null, []);
                        }
                    }
                },
                function(err, parts) {
                    if (err) {
                        callback(err, null);
                    } else { 
                        callback(null, (parts.length === 1) ? parts[0] : parts[0].concat(parts[1]));
                    }
                }
            );
        };

    if (start < 0 || end < 0 || start > end) {
        callback(new Error("Bad parameters"), null);
    }

    Step(
        function() {
            // XXX: maybe just take slice from [0, end/HARD_LIMIT)
            bank.read('streamsegments', stream.name, this);
        },
        function(err, segments) {
            if (err) throw err;
            getMore(segments, start, end, this);
        },
        function(err, ids) {
            if (err) throw err;
            if (ids.length === 0) {
                callback(null, []);
            } else {
                bank.readAll('activity', ids, this);
            }
        },
        function(err, activityMap) {
            var activities = [], id, group;
            if (err) {
                callback(err, null);
            } else {
                for (id in activityMap) {
                    activities.push(new Activity(activityMap[id]));
                }
                activities.sort(function(a, b) {  
                    if (a.published > b.published) {
                        return -1;  
                    } else if (a.published < b.published) {
                        return 1;  
                    } else {
                        return 0;  
                    }
                });
                callback(null, activities);
            }
        }
    );
};

Stream.prototype.count = function(callback) {
    var bank = Stream.bank();
    
    bank.read('streamcount', this.name, callback);
};

Stream.schema = {
    'stream': {'pkey': 'name'},
    'streamcount': {'pkey': 'name'},
    'streamsegments': {'pkey': 'name'},

    'streamsegment': {'pkey': 'id'},
    'streamsegmentcount': {'pkey': 'id'}
};

{'fraction_non_alphanumeric': 0.12349318658280922, 'fraction_numerical': 0.005110062893081761, 'mean_word_length': 1.3710779745262505, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 14, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 14, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5481349', 'n_tokens_mistral': 4093, 'n_tokens_neox': 3903, 'n_words': 1023}