Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Stream updates #221

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Stream updates #221

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a601c00
arrange deps by name
particlebanana Jan 25, 2016
7635718
update the stream method
particlebanana Jan 25, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
86 changes: 53 additions & 33 deletions lib/adapter.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@

// Dependencies
var pg = require('pg');
var QueryStream = require('pg-query-stream');
var _ = require('lodash');
var url = require('url');
var async = require('async');
Expand Down Expand Up @@ -265,12 +266,12 @@ module.exports = (function() {
// If the schema name is "public", just finish creating the table
if (schemaName == 'public') {return _define();}

// If not, attempt to create the schema first. This will succeed if
// If not, attempt to create the schema first. This will succeed if
// the schema already exists.
adapter.createSchema(connectionName, table, schemaName, _define);

function _define(errCreatingSchema) {

if (errCreatingSchema) {
cb(handleQueryError(errCreatingSchema));
}
Expand Down Expand Up @@ -333,7 +334,7 @@ module.exports = (function() {
throw new Error("No schemaName specified, and could not determined schemaname for table `" + table + "`");
}

// Build Query
// Build Query
var query = 'CREATE SCHEMA "' + schemaName + '"';

spawnConnection(connectionName, function (client, cb) {
Expand Down Expand Up @@ -994,42 +995,61 @@ module.exports = (function() {

// Stream one or more models from the collection
stream: function(connectionName, table, options, stream) {
spawnConnection(connectionName, function __STREAM__(client, cb) {

var connectionObject = connections[connectionName];
var collection = connectionObject.collections[table];
// Grab Connection Schema
var schema = {};
var connectionObject = connections[connectionName];
var collection = connectionObject.collections[table];
var tableName = table;

var client = new pg.Client(connectionObject.config);
client.connect();
Object.keys(connectionObject.collections).forEach(function(coll) {
schema[coll] = connectionObject.collections[coll].schema;
});

var schema = {};
// Build Query
var _schema = connectionObject.schema;
var processor = new Processor(_schema);

Object.keys(connectionObject.collections).forEach(function(coll) {
schema[coll] = connectionObject.collections[coll].schema;
});
// Mixin WL Next connection overrides to sqlOptions
var overrides = connectionOverrides[connectionName] || {};
var _options = _.cloneDeep(sqlOptions);
if(hop(overrides, 'wlNext')) {
_options.wlNext = overrides.wlNext;
}

// Build Query
var _schema = collection.schema;
var queryObj = new Query(_schema, schema);
var query =queryObj.find(table, options);
var sequel = new Sequel(_schema, _options);
var _query;

// Run Query
var dbStream = client.query(query.query, query.values);
// Build a query for the specific query strategy
try {
_query = sequel.find(tableName, options);
} catch(e) {
return cb(e);
}

//can stream row results back 1 at a time
dbStream.on('row', function(row) {
stream.write(row);
});
var streamQuery = new QueryStream(_query.query[0], _query.values[0]);
var dbStream = client.query(streamQuery);

// Can stream row results back 1 at a time
dbStream.on('data', function(row) {
stream.write(row);
});
//
dbStream.on('error', function(err) {
stream.end(); // End stream
cb();
});

// fired after last row is emitted
dbStream.on('end', function() {
stream.end(); // End stream
cb();
});

}, cb);

dbStream.on('error', function(err) {
stream.end(); // End stream
client.end(); // Close Connection
});

//fired after last row is emitted
dbStream.on('end', function() {
stream.end(); // End stream
client.end(); // Close Connection
});
},

// Update one or more models in the collection
Expand Down Expand Up @@ -1323,15 +1343,15 @@ module.exports = (function() {
rule: 'unique'
}];
}
}
}

else if (err.code === '3F000') {
else if (err.code === '3F000') {
formattedErr = {};
formattedErr.message = 'Attempted to create a table `' + err.table + '` in a schema `' + err.schemaName + '` that does not exist. Either create the schema manually or make sure that the `createSchemas` flag in your connection configuration is not set to `false`.';
delete err.table;
delete err.schemaName;
formattedErr.originalError = err;
}
}

else if (err.type == 'CREATING_SCHEMA') {
formattedErr = {};
Expand Down
9 changes: 5 additions & 4 deletions package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,12 +15,13 @@
"url": "git://github.com/balderdashy/sails-postgresql.git"
},
"dependencies": {
"pg": "~4.4.0",
"lodash": "~3.10.0",
"async": "~1.5.2",
"lodash": "~3.10.0",
"pg": "~4.4.0",
"pg-query-stream": "^1.0.0",
"waterline-cursor": "~0.0.6",
"waterline-errors": "~0.10.0",
"waterline-sequel": "~0.5.3",
"waterline-cursor": "~0.0.6"
"waterline-sequel": "~0.5.3"
},
"devDependencies": {
"mocha": "*",
Expand Down