How to use the qminer.StreamAggr function in qminer

it('should should throw exception', function () {
            var aggr = new qm.StreamAggr(base, new function () {
                var updates = 0;
                this.name = 'simple';
                this.onAdd = function (rec) {
                    updates++;
                }
                this.saveJson = function (limit) {
                    return { val: updates };
                }
            });  

            var OKInput = [{ type: "recordFilterAggr", aggr: aggr.name, filters: [{ type: "trivial" }] }];
            OKInput.push({ type: "recordFilterAggr", aggr: aggr.name, filters: [{ type: "field", store: "RecordTest", field: "Int", minValue: 5 }] });
            // missing fields
            var BADInput = [{ type: "recordFilterAggr", filters: [{ type: "field", store: "RecordTest", field: "Int", minValue: 5 }] }];
            BADInput.push({ type: "recordFilterAggr", aggr: aggr.name });
            BADInput.push({ type: "recordFilterAggr", aggr: aggr.name, filters: [{ store: "RecordTest", field: "Int", minValue: 5 }] });

it('should identify the filter aggregate as the caller', function (done) {
            var outAggr = new qm.StreamAggr(base, new function () {
                this.onAdd = function (rec, agg) {
                    if (filt.name != agg.name) {
                        // assert doesn't work in this async setting
                        done(new Error('caller incorrect'));
                    }
                }
            });

            var filt = store.addStreamAggr({
                type: 'recordFilterAggr',
                aggr: outAggr.name,
                filters: [{
                    type: "field",
                    store: "RecordTest",
                    field: "Bool",
                    value: true

"fields": [
                      { "name": "User", "type": "int" },
                      { "name": "Time", "type": "datetime" },
                      { "name": "Location", type: "float_pair" },
                      { "name": "Accuracy", type: "byte", "null": true },
                      { "name": "Activities", type: "int_v", "null": true }
                    ],
                    "joins": [],
                    "keys": []
                }
            ]
        });
        // used only for schema
        // will not be used to hold records (push will not be called)
        store = base.store("GPS");
        spdAggr = new qm.StreamAggr(base, {
            type: "stayPointDetector",
            store: store,
            userField: "User",
            timeField: "Time",
            locationField: "Location",
            accuracyField: "Accuracy",
            activitiesField: "Activities",
            params: { dT: 50, tT: 300 }
        });
    });//beforeEach

_addNewStreamAggr(mergerFlag, storeName, streamName, io, socketName) {
        // creates socket emit object
        function createSocketObj(rec, merger) {
            return merger ? {
                firstValue: rec.FirstValue,
                secondValue: rec.SecondValue,
                time: rec.Time
            } : {
                value: rec.Value,
                time: rec.Time
            };
        }

        new qm.StreamAggr(this.base, new function () {
            this.name = streamName;
            this.onAdd = function (rec) {
                if (io.sockets.connected) {
                    io.sockets.emit(socketName, createSocketObj(rec, mergerFlag));
                }
            };
            this.saveJson = function (limit) {
                return {};
            };
        }, storeName);
    }

it('should filter records outside id range', function (done) {
            var aggr = new qm.StreamAggr(base, new JsAggr);

            var filt = store.addStreamAggr({
                type: 'recordFilterAggr',
                aggr: aggr.name,
                filters: [{
                    type: "recordId",
                    store: "RecordTest",
                    minRecId: 2,
                    maxRecId: 3
                }]
            });
            assertUpdateSequence("Str", ["a", "b", "c", "d", "e"], [0, 0, 1, 2, 2], store, aggr);
            done();
        });
    });

it('should create a new merger aggregator', function () {
            var aggr = {
                name: 'MergerAggr',
                type: 'stmerger',
                outStore: 'Merged',
                createStore: false,
                timestamp: 'Time',
                fields: [
                    { source: 'Cars', inField: 'NumberOfCars', outField: 'NumberOfCars', interpolation: 'linear', timestamp: 'Time' },
                    { source: 'Temperature', inField: 'Celcius', outField: 'Celcius', interpolation: 'linear', timestamp: 'Time' }
                ]
            };
            var merger = new qm.StreamAggr(base, aggr);
            assert.equal(merger.name, 'MergerAggr');
        })        
        it('should throw an exception if a key-value is not given', function () {

it('should construct a new stream aggregator for the base and People store', function () {

            var aggr = new qm.StreamAggr(base, new function () {
                var length = 0;
                this.name = 'nameLength';
                this.onAdd = function (rec) {
                    length = rec.Name.length;
                }
                this.saveJson = function (limit) {
                    return { val: length };
                }
            }, 'People');

            var id1 = base.store('People').push({ Name: "John", Gendre: "Male" });
            assert.equal(aggr.saveJson().val, 4);
        })
        it('should contruct a new stream aggregator for the People store by adding it', function () {

it('should execute the onUpdate function and return 1', function () {
            var aggr = new qm.StreamAggr(base, new function () {
                var type = null;
                this.name = 'gendreUpdateLength';
                this.onAdd = function (rec) {
                    type = null;
                }
                this.onUpdate = function (rec) {
                    type = rec.Gendre == "Male" ? 0 : 1;
                }
                this.saveJson = function (limit) {
                    return { val: type };
                }
            });

            var id1 = base.store('People').push({ Name: "John", Gendre: "Male" });
            aggr.onAdd(base.store('People')[0]);

_addStreamAggr(singleStore, opts, storeName) {
        if (singleStore) {
            this.base.store(storeName).addStreamAggr(opts);
        } else {
            new qm.StreamAggr(this.base, opts);
        }
    }