introduce sgvdata, use it in api and storage

Clean up and comment a bunch of code.
This commit is contained in:
Ben West
2014-07-15 15:10:49 -07:00
parent b35853bcaa
commit b1ecbe3c91
4 changed files with 205 additions and 164 deletions
+3 -3
View File
@@ -12,11 +12,11 @@ UNIX=$(date +%s $OPTS)
(
cat <<EOF
{ "sgv": $SGV,
"device": "test",
{ "sgv": "$SGV",
"device": "xxx-test-device",
"direction": "$DIRECTION",
"dateString": "$ISO",
"date": "$UNIX"
"date": $UNIX
}
EOF
) | tr -d '\n'
+144 -127
View File
@@ -2,26 +2,49 @@
var consts = require('./constants');
var es = require('event-stream');
var sgvdata = require('sgvdata');
/*
* API - Expose Nightscout HTTP API
* This api is designed to work with express.
*/
function api (env, entries, settings) {
// our globals
var express = require('express'),
api = express(),
bodyParser = require('body-parser');
api = express()
;
var verifyAuthorization = require('./middleware/verify-token')(env);
var sendJSONStatus = require('./middleware/send-json-status')( );
// some middleware
var verifyAuthorization = require('./middleware/verify-token')(env),
sendJSONStatus = require('./middleware/send-json-status')( ),
bodyParser = require('body-parser')
;
// set up express basics
api.set('title', 'Nightscout API v1');
// invoke common middleware
api.use(sendJSONStatus);
// text body types get handled as raw buffer stream
api.use(bodyParser.raw());
// json body types get handled as parsed json
api.use(bodyParser.json());
// shortcut to use extension to specify output content-type
api.use(require('express-extension-to-accept')([
'json', 'svg', 'csv', 'txt', 'png', 'html', 'tsv'
]));
// also support url-encoded content-type
api.use(bodyParser.urlencoded({
extended: true
}));
/*
* Start setting up routes
*/
// Some experiments
api.get('/authorized/:secret/test', verifyAuthorization, function (req, res, next) {
return res.json({status: 'ok'});
});
@@ -30,6 +53,7 @@ function api (env, entries, settings) {
return res.json({status: 'ok'});
});
// Status badge/text/json
api.get('/status', function (req, res, next) {
var status = {status: 'ok'};
var badge = 'http://img.shields.io/badge/Nightscout-OK-green';
@@ -52,134 +76,126 @@ function api (env, entries, settings) {
});
});
api.get('/entries', function(req, res) {
console.log('body', req.body);
console.log('params', req.params);
console.log('query', req.query);
// If "?count=" is present, use that number to decided how many to return.
var count = parseInt(req.query.count, 0) || consts.ENTRIES_DEFAULT_COUNT;
var query = { count: count };
entries.list(query, function(err, entries) {
if (err)
res.sendJSONStatus(res, consts.HTTP_INTERNAL_ERROR, 'Mongo Error', err);
else
return res.json(entries);
});
return;
});
/**********\
* Entries
\**********/
api.get('/entries/current', function(req, res) {
entries.list({count: 1}, function(err, entries) {
if (err)
res.sendJSONStatus(res, consts.HTTP_INTERNAL_ERROR, 'Mongo Error', err);
else
return res.json(entries);
});
return;
});
function insert_entries (req, res, next) {
var incoming = [ ];
if ('sgv' in req.body) {
incoming.push(req.body);
// Middleware to format any response involving entries.
function format_entries (req, res, next) {
var output = es.readArray(res.entries || [ ]);
if (res.entries_err) return res.sendJSONStatus(res, consts.HTTP_INTERNAL_ERROR, 'Mongo Error', err);
return res.format({
text: function ( ) {
es.pipeline(output, sgvdata.format( ), res);
},
json: function ( ) {
es.pipeline(output, sgvdata.lint({strict: false}), es.writeArray(function (err, out) {
res.json(out);
}));
}
if (req.body.length) {
incoming = incoming.concat(req.body);
}
function inputs ( ) {
var rec_sep = ',';
var input;
if (req.is('text/*')) {
if (req.is('text/tsv')) {
rec_sep = '\t';
}
input = es.pipeline(req,
es.split('\n'),
es.map(function (str, next) {
var p = str.split(rec_sep);
if (p.length < 4) { return next(null); }
var rec = {
dateString: p[0]
, date: parseInt(p[1])
, sgv: p[2]
, direction: p[3]
, device: p[4]
};
console.log(p);
next(null, rec);
})
);
return input;
}
return es.readArray(incoming);
}
es.pipeline(inputs( ),
entries.map( ),
es.writeArray(function (err, result) {
// incoming = incoming.concat(result);
// var records = incoming.concat(result);
console.log(result);
if (req.persist_entries) {
entries.create(result, function ( ) {
console.log("WEB DONE", arguments);
res.json(result);
});
return;
}
return res.format({
/*
html: function ( ) {
res.send("<h1>STATUS OK</h1>");
},
png: function ( ) {
res.redirect(302, badge + '.png');
},
svg: function ( ) {
res.redirect(302, badge + '.svg');
},
text: function ( ) {
res.send("STATUS OK");
},
*/
json: function ( ) {
res.json(result);
}
});
}) // ;
);
});
}
// middleware to process "uploads" of sgv data
function insert_entries (req, res, next) {
// list of incoming records
var incoming = [ ];
// Potentially a single json encoded body.
// This can happen from either an url-encoded or json content-type.
if ('sgv' in req.body) {
// add it to the incoming list
incoming.push(req.body);
}
// potentially a list of json entries
if (req.body.length) {
// add them to the list
incoming = incoming.concat(req.body);
}
/*
* inputs -> <ReadableStream>
* in node, pipe is the most interoperable interface
* inputs returns a readable stream representing all the potential
* records from the HTTP body.
* Most content-types are handled by express middeware.
* However, text/* types are given to us as a raw buffer, this
* function switches between these two variants to find the
* correct input stream.
* stream, so use svgdata to handle those.
* The inputs stream always emits sgv json objects.
*/
function inputs ( ) {
var input;
// handle all text types
if (req.is('text/*')) {
// re-use the svgdata parsing stream
input = es.pipeline(req, sgvdata.parse( ));
return input;
}
// use established list
return es.readArray(incoming);
}
// return a writable persistent storage stream
function persist (fn) {
if (req.persist_entries) {
// store everything
return entries.persist(fn);
}
// support a preview mode, just lint everything
return es.pipeline(entries.map( ), es.writeArray(fn));
}
// store results and move to the next middleware
function done (err, result) {
res.entries = result;
res.entries_err = err;
return next( );
}
// pipe everything to persistent storage
// when finished, pass to the next piece of middleware
es.pipeline(inputs( ), persist(done));
}
api.get('/entries', function(req, res, next) {
// If "?count=" is present, use that number to decided how many to return.
var query = req.query;
entries.list(query, function(err, entries) {
res.entries = entries;
res.entries_err = err;
return next( );
});
return;
}, format_entries);
api.get('/entries/current', function(req, res, next) {
entries.list({count: 1}, function(err, entries) {
res.entries = entries;
res.entries_err = err;
return next( );
});
return;
}, format_entries);
// Allow previewing your post content, just echos everything you
// posted back out.
api.post('/entries/preview', function (req, res, next) {
var status = {
params: req.params,
body: req.body,
query: req.query
};
req.persist_entries = false;
console.log(status);
next( );
return;
}, insert_entries);
req.persist_entries = false;
next( );
return;
}, insert_entries, format_entries);
api.post('/entries', function (req, res, next) {
var status = {
params: req.params,
body: req.body,
query: req.query
};
req.persist_entries = true;
console.log(status);
next( );
return;
}, insert_entries);
// Create and store new sgv entries
api.post('/entries', verifyAuthorization, function (req, res, next) {
req.persist_entries = true;
next( );
return;
}, insert_entries, format_entries);
// Fetch one entry by id
api.get('/entries/:id', function(req, res) {
console.log('body', req.body);
console.log('params', req.params);
console.log('query', req.query);
entries.getEntry(function(err, entry) {
if (err)
res.sendJSONStatus(res, consts.HTTP_INTERNAL_ERROR, 'Mongo Error', err);
@@ -188,10 +204,8 @@ function api (env, entries, settings) {
}, req.params.id);
});
// Fetch settings
api.get('/settings', function(req, res) {
console.log('body', req.body);
console.log('params', req.params);
console.log('query', req.query);
settings.getSettings(function(err, settings) {
if (err)
res.sendJSONStatus(res, consts.HTTP_INTERNAL_ERROR, 'Mongo Error', err);
@@ -200,11 +214,14 @@ function api (env, entries, settings) {
});
});
// Delete settings
api.delete('/settings', verifyAuthorization, function(req, res) {
settings.remove(function ( ) {
res.json({ });
});
});
// Replace settings
api.put('/settings', verifyAuthorization, function(req, res) {
// Retrieve the JSON formatted record.
var json = req.body;
+56 -33
View File
@@ -1,22 +1,40 @@
'use strict';
var es = require('event-stream');
var sgvdata = require('sgvdata');
/**********\
* Entries
* Encapsulate persistent storage of sgv entries.
\**********/
function entries (name, storage) {
// TODO: Code is a little redundant.
var with_collection = storage.with_collection(name);
// query for entries from storage
function list (opts, fn) {
with_collection(function (err, collection) {
// these functions, find, sort, and limit, are used to
// dynamically configure the request, based on the options we've
// been give
// determine find options
function find ( ) {
var q = opts && opts.find ? opts.find : { };
return q;
return this.find(q)
;
// return this.find(q);
}
// determine sort options
function sort ( ) {
return {"date": -1};
return this.sort({"date": -1});
// return this.sort({"date": -1});
}
// configure the limit portion of the current query
function limit ( ) {
if (opts && opts.count) {
return this.limit(parseInt(opts.count));
@@ -24,10 +42,12 @@ function entries (name, storage) {
return this;
}
// handle all the results
function toArray (err, entries) {
fn(err, entries);
}
// now just stitch them all together
limit.call(collection
.find(find( ))
.sort(sort( )))
@@ -39,39 +59,42 @@ function entries (name, storage) {
});
}
function echo (record) {
var res = {
sgv: record.sgv
, dateString: record.dateString || ''
, date: parseInt(record.date)
, device: record.device || ''
, direction: record.direction || ''
};
if (res.sgv && isFinite(res.date)) {
return res;
}
}
// return writable stream to lint each sgv record passing through it
function map ( ) {
return es.map(function (item, next) {
var record = echo(item);
if (record) next(null, echo(item));
else next(null);
});
return sgvdata.lint( );
}
// writable stream that persists all records
// takes function to call when done
function persist (fn) {
// receives entire list at end of stream
function done (err, result) {
// report any errors
if (err) return fn(err, result);
// batch insert a list of records
create(result, fn);
return;
}
// lint and store the entire list
return es.pipeline(map( ), es.writeArray(done));
}
function update (fn) {
}
function remove (fn) {
// TODO: implement
}
function remove (fn) {
// TODO: implement
}
// store new documents using the storage mechanism
function create (docs, fn) {
with_collection(function(err, collection) {
if (err) { fn(err); return; }
collection.insert(docs, function (err, stats) {
console.log('finished inserting', stats, arguments);
fn(err, stats, docs);
// potentially a batch insert
collection.insert(docs, function (err, created) {
// execute the callback
fn(err, created, docs);
});
});
}
@@ -105,11 +128,16 @@ function entries (name, storage) {
});
}
// closure to represent the API
function api ( ) {
// obtain handle usable for querying the collection associated
// with these records
return storage.pool.db.collection(name);
}
// Expose all the useful functions
api.list = list;
api.echo = echo;
api.echo = sgvdata.sync.json.echo;
api.map = map;
api.create = create;
api.getEntries = getEntries;
@@ -117,11 +145,6 @@ function entries (name, storage) {
return api;
}
// expose module
module.exports = entries;
if (!module.parent) {
var env = require('../env');
var store = require('./storage')(env);
var api = entries(env.mongo_collection, store);
}
+2 -1
View File
@@ -38,6 +38,7 @@
"mongodb": "^1.4.7",
"socket.io": "^0.9.17",
"express-extension-to-accept": "0.0.2",
"event-stream": "~3.1.5"
"event-stream": "~3.1.5",
"sgvdata": "0.0.2"
}
}