Skip to content

Commit

Permalink
use make-fetch-happen to replace request
Browse files Browse the repository at this point in the history
  • Loading branch information
@stuartf
stuartf committed Jun 3, 2019
1 parent e27c00f commit d569e9e
Show file tree
Hide file tree
Showing 6 changed files with 1,996 additions and 704 deletions.
18 changes: 18 additions & 0 deletions .eslintrc.js
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,18 @@
module.exports = {
extends: [
"google",
"plugin:prettier/recommended"
],
rules: {
"no-undef": "error"
},
env: {
es6: true,
node: true,
jest: true
},
parserOptions: {
sourceType: "module",
ecmaVersion: 2018
}
};
1 change: 1 addition & 0 deletions .gitignore
View file
Original file line number Diff line number Diff line change
@@ -1,2 +1,3 @@
node_modules/
coverage/
.nyc_output/
245 changes: 100 additions & 145 deletions index.js
View file
Original file line number Diff line number Diff line change
@@ -1,108 +1,89 @@
const https = require("https");
const isEmpty = require("lodash.isempty");
const debug = require("debug")("buzzapi");
const hyperid = require("hyperid")({ fixedLength: true, urlSafe: true });
const os = require("os");
let request = require("requestretry");
const util = require("util");
const BuzzAPIError = require("./buzzApiError");

// Enable cookies
request = request.defaults({ jar: true });

const BuzzAPI = function(config) {
const that = this;

this.options = {
api_app_id: config.apiUser,
api_app_password: Buffer.from(config.apiPassword).toString("base64"),
api_request_mode: config.sync ? "sync" : "async",
ticket: /.*/,
ticket: "",
api_receive_timeout: config.api_receive_timeout || 900000
};

const server = config.server || "https://api.gatech.edu";

let openReqs = 0;
const queuedReqs = [];
const fetch = require("make-fetch-happen").defaults({
retry: { retries: 5, randomize: true },
onRetry: () => debug("Retrying a request"),
agent: https.globalAgent
});

const unresolved = {};
let gettingResult = false;
let ticket = "";

this.post = function(resource, operation, data, callback) {
if (openReqs >= 20) {
return new Promise((res, rej) => {
queuedReqs.push({ args: arguments, res: res, rej: rej });
});
} else {
return doPost(resource, operation, data, callback);
}
};

const doPost = function(resource, operation, data, callback) {
this.post = function(resource, operation, data) {
debug("Options: " + JSON.stringify(that.options));
return new Promise((res, rej) => {
openReqs++;
if (typeof data === "function") {
callback = data;
data = {};
}
const myOpts = {};
myOpts.url = `${server}/apiv3/${resource}/${operation}`;
myOpts.api_client_request_handle =
data.api_client_request_handle ||
`${process.pid}@${os.hostname()}-${hyperid()}`;
myOpts.json = Object.assign(data, that.options);
const myOpts = {
method: "POST",
body: JSON.stringify({
...that.options,
...data,
api_client_request_handle:
data.api_client_request_handle ||
`${process.pid}@${os.hostname()}-${hyperid()}`
})
};
debug("Requesting %s", JSON.stringify(myOpts));
request.post(myOpts, function(err, response, body) {
if (response && response.attempts && response.attempts > 1) {
debug("Request took multiple attempts %s", response.attempts);
}
if (err || response.statusCode > 299 || body.api_error_info) {
if (body) {
const error = new BuzzAPIError(err, body.api_error_info, body);
return callback ? callback(error, null, body) : rej(error);
} else {
const error = new BuzzAPIError(err);
return callback ? callback(error) : rej(error);
return fetch(`${server}/apiv3/${resource}/${operation}`, myOpts).then(
response => {
if (!response.ok || response.status > 299) {
return rej(new BuzzAPIError(null, null, response.statusText));
}
} else if (that.options.api_request_mode === "sync") {
debug("Sync was set, returning the result");
resolve();
return callback
? callback(null, body.api_result_data, body)
: res(body.api_result_data);
} else {
debug(
"Got messageId: %s for %s",
body.api_result_data,
myOpts.api_client_request_handle
);
unresolved[body.api_result_data] = {
resolve: res,
reject: rej,
callback: callback,
initTime: new Date()
};
ticket = body.api_app_ticket;
return getResult();
response
.json()
.then(json => {
if (json.api_error_info) {
const error = new BuzzAPIError(
new Error(),
json.api_error_info,
json
);
return rej(error);
} else if (that.options.api_request_mode === "sync") {
debug("Sync was set, returning the result");
return res(json.api_result_data);
} else {
debug(
"Got messageId: %s for %s",
json.api_result_data,
myOpts.body.api_client_request_handle
);
unresolved[json.api_result_data] = {
resolve: res,
reject: rej,
initTime: new Date()
};
that.options.ticket = json.api_app_ticket;
return getResult();
}
})
.catch(err => rej(new BuzzAPIError()));
}
});
);
});
};

const resolve = function(messageId) {
openReqs--;
debug("queued: %s open: %s", queuedReqs.length, openReqs);
if (messageId) {
delete unresolved[messageId];
}
if (queuedReqs[0]) {
const next = queuedReqs.pop();
return doPost
.apply(null, next.args)
.then(next.res)
.catch(next.rej);
}
};

const cleanupExpired = function() {
Expand All @@ -112,11 +93,7 @@ const BuzzAPI = function(config) {
that.options.api_receive_timeout
) {
const err = new Error("Request timed out for: " + messageId);
if (unresolved[messageId].callback) {
unresolved[messageId].callback(err);
} else {
unresolved[messageId].reject(err);
}
unresolved[messageId].reject(err);
resolve(messageId);
}
});
Expand All @@ -141,112 +118,90 @@ const BuzzAPI = function(config) {
}
const handle = `${process.pid}@${os.hostname()}-${hyperid()}`;
debug("Asking for result of %s using handle %s", messageIds, handle);
request.post(
{
url: util.format("%s/apiv3/api.my_messages", server),
json: {
api_operation: "read",
api_app_ticket: ticket,
api_pull_response_to: messageIds,
api_receive_timeout: 5000,
api_client_request_handle: handle
}
},
(err, response, body) => {
gettingResult = false;
if (response && response.attempts && response.attempts > 1) {
debug("Request took multiple attempts %s", response.attempts);
}
return fetch(`${server}/apiv3/api.my_messages`, {
method: "POST",
body: JSON.stringify({
api_operation: "read",
api_app_ticket: that.options.ticket,
api_pull_response_to: messageIds,
api_receive_timeout: 5000,
api_client_request_handle: handle
})
}).then(response => {
gettingResult = false;
if (!response.ok || response.status > 299) {
messageIds.map(messageId => {
const message = unresolved[messageId];
message.reject(new BuzzAPIError(null, null, response.statusText));
resolve(messageId);
});
return Promise.reject(
new BuzzAPIError(null, null, response.statusText)
);
}

response.json().then(json => {
if (
err ||
response.statusCode > 299 ||
body.api_error_info ||
(body.api_result_data && body.api_result_data.api_error_info)
json.api_error_info ||
(json.api_result_data && json.api_result_data.api_error_info)
) {
if (!body) {
messageIds.map(messageId => {
const message = unresolved[messageId];
if (message.callback) {
message.callback(new BuzzAPIError(err), null);
} else {
message.reject(new BuzzAPIError(err));
}
resolve(messageId);
});
return;
}
if (body.api_error_info) {
if (json.api_error_info) {
const message =
unresolved[body.api_error_info.api_request_messageid];
unresolved[json.api_error_info.api_request_messageid];
const err = new BuzzAPIError(
"BuzzApi returned error_info",
body.api_error_info,
body
json.api_error_info,
json
);
debug(unresolved);
debug(body);
debug(json);
if (message) {
return message.callback
? message.callback(err, null)
: message.reject(err);
resolve(json.api_error_info.api_request_messageid);
return message.reject(err);
} else {
// We don't know which message threw the error, so invalidate all of them
messageIds.map(messageId => {
const m = unresolved[messageId];
if (m.callback) {
m.callback(err, null);
} else {
m.reject(err);
}
m.reject(err);
resolve(messageId);
});
return;
}
} else if (body.api_result_data) {
const messageId = body.api_result_data.api_request_messageid;
} else if (json.api_result_data) {
const messageId = json.api_result_data.api_request_messageid;
const message = unresolved[messageId];
resolve(messageId);
const err = new BuzzAPIError(
"BuzzApi returned error_info",
body.api_result_data.api_error_info,
body
json.api_result_data.api_error_info,
json
);
return message.callback
? message.callback(err, null)
: message.reject(err);
return message.reject(err);
} else {
messageIds.map(messageId => {
const error = new BuzzAPIError(err, body, body);
const error = new BuzzAPIError(new Error(), json, json);
const message = unresolved[messageId];
if (message.callback) {
message.callback(error, null);
} else {
message.reject(error);
}
resolve(messageId);
return message.reject(error);
});
return;
}
} else if (isEmpty(body.api_result_data)) {
} else if (isEmpty(json.api_result_data)) {
// Empty result_data here means our data isn't ready, wait 1 to 5 seconds and try again
debug("Result not ready for " + messageIds);
return scheduleRetry();
} else {
const messageId = body.api_result_data.api_request_messageid;
const messageId = json.api_result_data.api_request_messageid;
const message = unresolved[messageId];
debug("Got result for ", messageId);
resolve(messageId);
if (message.callback) {
message.callback(null, body.api_result_data.api_result_data, body);
} else {
message.resolve(body.api_result_data.api_result_data);
}
message.resolve(json.api_result_data.api_result_data);
if (Object.keys(unresolved).length > 0) {
getResult();
}
return;
}
}
);
});
});
};
};

Expand Down

0 comments on commit d569e9e

Please sign in to comment.