@artlife/micro/build/index.js

"use strict";
//
// Shared microservices framework.
//
var __awaiter = (this && this.__awaiter) || function (thisArg, _arguments, P, generator) {
    return new (P || (P = Promise))(function (resolve, reject) {
        function fulfilled(value) { try { step(generator.next(value)); } catch (e) { reject(e); } }
        function rejected(value) { try { step(generator["throw"](value)); } catch (e) { reject(e); } }
        function step(result) { result.done ? resolve(result.value) : new P(function (resolve) { resolve(result.value); }).then(fulfilled, rejected); }
        step((generator = generator.apply(thisArg, _arguments || [])).next());
    });
};
var __generator = (this && this.__generator) || function (thisArg, body) {
    var _ = { label: 0, sent: function() { if (t[0] & 1) throw t[1]; return t[1]; }, trys: [], ops: [] }, f, y, t, g;
    return g = { next: verb(0), "throw": verb(1), "return": verb(2) }, typeof Symbol === "function" && (g[Symbol.iterator] = function() { return this; }), g;
    function verb(n) { return function (v) { return step([n, v]); }; }
    function step(op) {
        if (f) throw new TypeError("Generator is already executing.");
        while (_) try {
            if (f = 1, y && (t = op[0] & 2 ? y["return"] : op[0] ? y["throw"] || ((t = y["return"]) && t.call(y), 0) : y.next) && !(t = t.call(y, op[1])).done) return t;
            if (y = 0, t) op = [op[0] & 2, t.value];
            switch (op[0]) {
                case 0: case 1: t = op; break;
                case 4: _.label++; return { value: op[1], done: false };
                case 5: _.label++; y = op[1]; op = [0]; continue;
                case 7: op = _.ops.pop(); _.trys.pop(); continue;
                default:
                    if (!(t = _.trys, t = t.length > 0 && t[t.length - 1]) && (op[0] === 6 || op[0] === 2)) { _ = 0; continue; }
                    if (op[0] === 3 && (!t || (op[1] > t[0] && op[1] < t[3]))) { _.label = op[1]; break; }
                    if (op[0] === 6 && _.label < t[1]) { _.label = t[1]; t = op; break; }
                    if (t && _.label < t[2]) { _.label = t[2]; _.ops.push(op); break; }
                    if (t[2]) _.ops.pop();
                    _.trys.pop(); continue;
            }
            op = body.call(thisArg, _);
        } catch (e) { op = [6, e]; y = 0; } finally { f = t = 0; }
        if (op[0] & 5) throw op[1]; return { value: op[0] ? op[1] : void 0, done: true };
    }
};
var __read = (this && this.__read) || function (o, n) {
    var m = typeof Symbol === "function" && o[Symbol.iterator];
    if (!m) return o;
    var i = m.call(o), r, ar = [], e;
    try {
        while ((n === void 0 || n-- > 0) && !(r = i.next()).done) ar.push(r.value);
    }
    catch (error) { e = { error: error }; }
    finally {
        try {
            if (r && !r.done && (m = i["return"])) m.call(i);
        }
        finally { if (e) throw e.error; }
    }
    return ar;
};
var __spread = (this && this.__spread) || function () {
    for (var ar = [], i = 0; i < arguments.length; i++) ar = ar.concat(__read(arguments[i]));
    return ar;
};
var __values = (this && this.__values) || function (o) {
    var m = typeof Symbol === "function" && o[Symbol.iterator], i = 0;
    if (m) return m.call(o);
    return {
        next: function () {
            if (o && i >= o.length) o = void 0;
            return { value: o && o[i++], done: !o };
        }
    };
};
Object.defineProperty(exports, "__esModule", { value: true });
var express = require("express");
var amqp = require("amqplib");
var axios_1 = require("axios");
var request = require("request");
var utils_1 = require("./utils");
exports.asyncHandler = utils_1.asyncHandler;
exports.retry = utils_1.retry;
exports.sleep = utils_1.sleep;
exports.verifyBodyParam = utils_1.verifyBodyParam;
exports.verifyQueryParam = utils_1.verifyQueryParam;
var morganBody = require('morgan-body');
var http = require("http");
var bodyParser = require("body-parser");
var inProduction = process.env.NODE_ENV === "production";
var enableMorgan = !inProduction || process.env.ENABLE_MORGAN === "true";
//
// Logging implementation.
//
var Log = /** @class */ (function () {
    function Log() {
    }
    /**
     * Issue a warning.
     */
    Log.prototype.warn = function () {
        var args = [];
        for (var _i = 0; _i < arguments.length; _i++) {
            args[_i] = arguments[_i];
        }
        console.warn.apply(console, __spread(args));
    };
    /**
     * Issue an information message.
     */
    Log.prototype.info = function () {
        var args = [];
        for (var _i = 0; _i < arguments.length; _i++) {
            args[_i] = arguments[_i];
        }
        console.log.apply(console, __spread(args));
    };
    /**
     * Issue a verbose message.
     */
    Log.prototype.verbose = function () {
        var args = [];
        for (var _i = 0; _i < arguments.length; _i++) {
            args[_i] = arguments[_i];
        }
        console.log.apply(console, __spread(args));
    };
    /**
     * Record an error message.
     */
    Log.prototype.error = function () {
        var args = [];
        for (var _i = 0; _i < arguments.length; _i++) {
            args[_i] = arguments[_i];
        }
        console.error.apply(console, __spread(args));
    };
    /**
     * Record an exception that was thrown
     */
    Log.prototype.exception = function (err) {
        var args = [];
        for (var _i = 1; _i < arguments.length; _i++) {
            args[_i - 1] = arguments[_i];
        }
        console.error("Exception:");
        console.error(err && err.stack || err);
        console.error.apply(console, __spread(args));
    };
    return Log;
}());
var defaultConfig = {};
//
// Used to register an event handler to be setup after messaging system has started.
//
var EventHandler = /** @class */ (function () {
    function EventHandler(eventName, eventHandlerFn) {
        this.eventName = eventName;
        this.eventHandlerFn = eventHandlerFn;
    }
    return EventHandler;
}());
//
// Class that represents a particular microservice instance.
//
var MicroService = /** @class */ (function () {
    function MicroService(config) {
        var _this = this;
        //
        // Event handlers that have been registered to be setup once
        // connection to message queue is established.
        ///
        this.registeredEventHandlers = new Set();
        /**
         * Reference to the logging interface.
         * This allows the logging from multiple microservices to be aggregated.
         */
        this.log = new Log();
        /**
         * Reference to the timer interface.
         * Allows code to be timed for performance.
         */
        this.timer = {
            start: function (timerName) {
                // Just a stub for the moment.
            },
            stop: function (timerName) {
                // Just a stub for the moment.
            },
        };
        /**
         * Reference to the metrics interface.
         * Allows a service to output metrics.
         */
        this.metrics = {
            discrete: function (name, value) {
                // Just a stub for the moment.
            },
            continuous: function (name, value) {
                // Just a stub for the moment.
            },
        };
        this.config = config || defaultConfig;
        this.expressApp = express();
        this.httpServer = new http.Server(this.expressApp);
        if (!inProduction) {
            this.expressApp.use(function (req, res, next) {
                res.header("Access-Control-Allow-Origin", "*");
                res.header("Access-Control-Allow-Headers", "Origin, X-Requested-With, Content-Type, Accept");
                next();
            });
        }
        this.expressApp.use(bodyParser.json());
        if (enableMorgan) {
            console.log("Enabling Morgan request tracing.");
            morganBody(this.expressApp, {
                noColors: true,
            });
        }
        this.expressApp.get("/is-alive", function (req, res) {
            res.json({ ok: true });
        });
        this.rest = {
            /**
             * Create a handler for incoming HTTP GET requests.
             * Implemented by Express under the hood.
             */
            get: function (route, requestHandler) {
                _this.expressApp.get(route, function (req, res) {
                    _this.verbose("Handling GET " + route);
                    requestHandler(req, res)
                        .then(function () {
                        _this.verbose("HTTP GET handler for " + route + " finished.");
                    })
                        .catch(function (err) {
                        console.error("Error from handler: HTTP GET " + route);
                        console.error(err && err.stack || err);
                        res.sendStatus(500);
                    });
                });
            },
            /**
             * Create a handler for incoming HTTP POST requests.
             * Implemented by Express under the hood.
             */
            post: function (route, requestHandler) {
                _this.expressApp.post(route, function (req, res) {
                    _this.verbose("Handling POST " + route);
                    requestHandler(req, res)
                        .then(function () {
                        _this.verbose("HTTP POST handler for " + route + " finished.");
                    })
                        .catch(function (err) {
                        console.error("Error from handler: HTTP POST " + route);
                        console.error(err && err.stack || err);
                        res.sendStatus(500);
                    });
                });
            },
            /**
             * Setup serving of static files.
             *
             * @param dirPath The path to the directory that contains static files.
             */
            static: function (dirPath) {
                console.log("Serving static files from " + dirPath);
                _this.expressApp.use(express.static(dirPath));
            },
        };
        this.request = {
            /**
             * Make a HTTP get request to another service.
             *
             * @param serviceName The name (logical or host) of the service.
             * @param route The HTTP route on the service to make the request to.
             * @param params Query parameters for the request.
             */
            get: function (serviceName, route, body) { return __awaiter(_this, void 0, void 0, function () {
                var url;
                return __generator(this, function (_a) {
                    switch (_a.label) {
                        case 0:
                            url = "http://" + serviceName + route;
                            return [4 /*yield*/, axios_1.default.get(url, body)];
                        case 1: return [2 /*return*/, _a.sent()];
                    }
                });
            }); },
            /**
             * Make a HTTP get request to another service.
             *
             * @param serviceName The name (logical or host) of the service.
             * @param route The HTTP route on the service to make the request to.
             * @param params Query parameters for the request.
             */
            post: function (serviceName, route, body) { return __awaiter(_this, void 0, void 0, function () {
                var url;
                return __generator(this, function (_a) {
                    switch (_a.label) {
                        case 0:
                            url = "http://" + serviceName + route;
                            return [4 /*yield*/, axios_1.default.post(url, body)];
                        case 1:
                            _a.sent();
                            return [2 /*return*/];
                    }
                });
            }); },
            /**
             * Forward HTTP get request to another named service.
             * The response from the forward requests is automatically piped into the passed in response.
             *
             * @param serviceName The name of the service to forward the request to.
             * @param route The HTTP GET route to forward to.
             * @param params Query parameters for the request.
             * @param res The stream to pipe response to.
             */
            forward: function (serviceName, route, req, res) {
                var url = "http://" + serviceName + route;
                req.pipe(request(url)).pipe(res);
            }
        };
    }
    //
    // Helper method for verbose logging.
    //
    MicroService.prototype.verbose = function (msg) {
        if (this.config.verbose) {
            console.log(msg);
        }
    };
    //
    // Start the Express HTTP server.
    //
    MicroService.prototype.startHttpServer = function () {
        return __awaiter(this, void 0, void 0, function () {
            var _this = this;
            return __generator(this, function (_a) {
                return [2 /*return*/, new Promise(function (resolve, reject) {
                        var host;
                        if (_this.config.host !== undefined) {
                            host = _this.config.host;
                        }
                        else {
                            host = process.env.HOST || '0.0.0.0';
                        }
                        var port;
                        if (_this.config.port !== undefined) {
                            port = _this.config.port;
                        }
                        else {
                            port = (process.env.PORT && parseInt(process.env.PORT)) || 3000;
                        }
                        _this.httpServer.listen(port, host, function (err) {
                            if (err) {
                                reject(err);
                            }
                            else {
                                _this.verbose("Running on http://" + port + ":" + host);
                                resolve();
                            }
                        });
                    })];
            });
        });
    };
    //
    // Lazily start RabbitMQ messaging.
    //
    MicroService.prototype.startMessaging = function () {
        return __awaiter(this, void 0, void 0, function () {
            var initMessaging;
            var _this = this;
            return __generator(this, function (_a) {
                switch (_a.label) {
                    case 0:
                        initMessaging = function () { return __awaiter(_this, void 0, void 0, function () {
                            var e_1, _a, messagingHost, _b, _c, _d, _e, eventHandler, e_1_1;
                            var _this = this;
                            return __generator(this, function (_f) {
                                switch (_f.label) {
                                    case 0:
                                        if (this.config.messagingHost) {
                                            messagingHost = this.config.messagingHost;
                                        }
                                        else {
                                            messagingHost = process.env.MESSAGING_HOST || "amqp://guest:guest@localhost:5672";
                                        }
                                        this.verbose("Connteting to messaging server at: " + messagingHost);
                                        _b = this;
                                        return [4 /*yield*/, utils_1.retry(function () { return __awaiter(_this, void 0, void 0, function () { return __generator(this, function (_a) {
                                                switch (_a.label) {
                                                    case 0: return [4 /*yield*/, amqp.connect(messagingHost)];
                                                    case 1: return [2 /*return*/, _a.sent()];
                                                }
                                            }); }); }, 10000, 1000)];
                                    case 1:
                                        _b.messagingConnection = _f.sent();
                                        this.messagingConnection.on("error", function (err) {
                                            console.error("Error from message system.");
                                            console.error(err && err.stack || err);
                                        });
                                        this.messagingConnection.on("close", function (err) {
                                            _this.messagingConnection = undefined;
                                            _this.messagingChannel = undefined;
                                            console.log("Lost connection to rabbit, waiting for restart.");
                                            initMessaging()
                                                .then(function () { return console.log("Restarted messaging."); })
                                                .catch(function (err) {
                                                console.error("Failed to restart messaging.");
                                                console.error(err && err.stack || err);
                                            });
                                        });
                                        _c = this;
                                        return [4 /*yield*/, this.messagingConnection.createChannel()];
                                    case 2:
                                        _c.messagingChannel = _f.sent();
                                        _f.label = 3;
                                    case 3:
                                        _f.trys.push([3, 8, 9, 10]);
                                        _d = __values(this.registeredEventHandlers.values()), _e = _d.next();
                                        _f.label = 4;
                                    case 4:
                                        if (!!_e.done) return [3 /*break*/, 7];
                                        eventHandler = _e.value;
                                        return [4 /*yield*/, this.internalOn(eventHandler)];
                                    case 5:
                                        _f.sent();
                                        _f.label = 6;
                                    case 6:
                                        _e = _d.next();
                                        return [3 /*break*/, 4];
                                    case 7: return [3 /*break*/, 10];
                                    case 8:
                                        e_1_1 = _f.sent();
                                        e_1 = { error: e_1_1 };
                                        return [3 /*break*/, 10];
                                    case 9:
                                        try {
                                            if (_e && !_e.done && (_a = _d.return)) _a.call(_d);
                                        }
                                        finally { if (e_1) throw e_1.error; }
                                        return [7 /*endfinally*/];
                                    case 10: return [2 /*return*/];
                                }
                            });
                        }); };
                        return [4 /*yield*/, initMessaging()];
                    case 1:
                        _a.sent();
                        return [2 /*return*/];
                }
            });
        });
    };
    /**
     * Returns true if the messaging system is currently available.
     */
    MicroService.prototype.isMessagingAvailable = function () {
        return !!this.messagingConnection;
    };
    //
    // Setup a RabbitMQ message handler.
    //
    MicroService.prototype.internalOn = function (eventHandler) {
        return __awaiter(this, void 0, void 0, function () {
            var eventName, queueName, messagingChannel, consumeCallback;
            var _this = this;
            return __generator(this, function (_a) {
                switch (_a.label) {
                    case 0:
                        eventName = eventHandler.eventName;
                        // http://www.squaremobius.net/amqp.node/channel_api.html#channel_assertExchange
                        return [4 /*yield*/, this.messagingChannel.assertExchange(eventName, "fanout", { durable: true })];
                    case 1:
                        // http://www.squaremobius.net/amqp.node/channel_api.html#channel_assertExchange
                        _a.sent();
                        return [4 /*yield*/, this.messagingChannel.assertQueue("", { durable: true, exclusive: true })];
                    case 2:
                        queueName = (_a.sent()).queue;
                        eventHandler.queueName = queueName;
                        console.log('binding queue', queueName, 'to', eventName);
                        this.messagingChannel.bindQueue(queueName, eventName, "");
                        messagingChannel = this.messagingChannel;
                        consumeCallback = function (msg) { return __awaiter(_this, void 0, void 0, function () {
                            var args, eventResponse;
                            return __generator(this, function (_a) {
                                switch (_a.label) {
                                    case 0:
                                        this.verbose("Handling " + eventName);
                                        args = JSON.parse(msg.content.toString());
                                        eventResponse = {
                                            ack: function () {
                                                return __awaiter(this, void 0, void 0, function () {
                                                    return __generator(this, function (_a) {
                                                        messagingChannel.ack(msg);
                                                        return [2 /*return*/];
                                                    });
                                                });
                                            }
                                        };
                                        return [4 /*yield*/, eventHandler.eventHandlerFn(args, eventResponse)];
                                    case 1:
                                        _a.sent();
                                        this.verbose(eventName + " handler done.");
                                        return [2 /*return*/];
                                }
                            });
                        }); };
                        this.verbose("Receiving events on queue " + eventName);
                        // http://www.squaremobius.net/amqp.node/channel_api.html#channel_consume
                        this.messagingChannel.consume(queueName, utils_1.asyncHandler(this, "ASYNC: " + eventName, consumeCallback), {
                            noAck: false,
                        });
                        return [2 /*return*/];
                }
            });
        });
    };
    //
    // Unwind a RabbitMQ message handler.
    //
    MicroService.prototype.internalOff = function (eventHandler) {
        this.messagingChannel.unbindQueue(eventHandler.queueName, eventHandler.eventName, "");
        delete eventHandler.queueName;
    };
    /**
     * Create an ongoing handler for a named incoming event.
     * Implemented by Rabbitmq under the hood for reliable messaging.
     *
     * @param eventName The name of the event to handle.
     * @param eventHandler Callback to be invoke when the incoming event is received.
     */
    MicroService.prototype.on = function (eventName, eventHandlerFn) {
        return __awaiter(this, void 0, void 0, function () {
            var eventHandler;
            return __generator(this, function (_a) {
                switch (_a.label) {
                    case 0:
                        eventHandler = new EventHandler(eventName, eventHandlerFn);
                        this.registeredEventHandlers.add(eventHandler);
                        if (!this.messagingConnection) return [3 /*break*/, 2];
                        //
                        // Message system already started.
                        //
                        return [4 /*yield*/, this.internalOn(eventHandler)];
                    case 1:
                        //
                        // Message system already started.
                        //
                        _a.sent();
                        _a.label = 2;
                    case 2: return [2 /*return*/, eventHandler];
                }
            });
        });
    };
    /**
     * Unregister a previously register event handler.
     *
     * @param handler The event handler to unregister.
     */
    MicroService.prototype.off = function (handler) {
        var eventHandler = handler;
        if (this.messagingConnection) {
            this.internalOff(eventHandler);
        }
        this.registeredEventHandlers.delete(eventHandler);
    };
    /**
     * Create a once-off handler for a named incoming event.
     * The event handler will only be invoke once before the event is unregistered.
     * Implemented by Rabbitmq under the hood for reliable messaging.
     *
     * @param eventName The name of the event to handle.
     * @param eventHandler Callback to be invoke when the incoming event is received.
     */
    MicroService.prototype.once = function (eventName, eventHandlerFn) {
        return __awaiter(this, void 0, void 0, function () {
            var eventHandler;
            var _this = this;
            return __generator(this, function (_a) {
                switch (_a.label) {
                    case 0: return [4 /*yield*/, this.on(eventName, function (args, res) { return __awaiter(_this, void 0, void 0, function () {
                            return __generator(this, function (_a) {
                                switch (_a.label) {
                                    case 0:
                                        this.off(eventHandler); // Unregister before we receive any more events.
                                        return [4 /*yield*/, eventHandlerFn(args, res)];
                                    case 1:
                                        _a.sent(); // Trigger user callback.
                                        return [2 /*return*/];
                                }
                            });
                        }); })];
                    case 1:
                        eventHandler = _a.sent();
                        return [2 /*return*/];
                }
            });
        });
    };
    /***
     * Wait for a single incoming event, returns the events arguments and then unregister the event handler.
     *
     * @param eventName The name of the event to handle.
     *
     * @returns A promise to resolve the incoming event's arguments.
     */
    MicroService.prototype.waitForOneEvent = function (eventName) {
        var _this = this;
        return new Promise(function (resolve) {
            _this.once(eventName, function (args, res) { return __awaiter(_this, void 0, void 0, function () {
                return __generator(this, function (_a) {
                    res.ack(); // Ack the response.
                    resolve(args); // Resolve event args through the promise.
                    return [2 /*return*/];
                });
            }); });
        });
    };
    /**
     * Emit a named outgoing event.
     * Implemented by Rabbitmq under the hood for reliable messaging.
     *
     * @param eventName The name of the event to emit.
     * @param eventArgs Event args to publish with the event and be received at the other end.
     */
    MicroService.prototype.emit = function (eventName, eventArgs) {
        return __awaiter(this, void 0, void 0, function () {
            return __generator(this, function (_a) {
                switch (_a.label) {
                    case 0:
                        if (!this.messagingConnection) {
                            throw new Error("Messaging system currently unavailable.");
                        }
                        // http://www.squaremobius.net/amqp.node/channel_api.html#channel_assertExchange
                        return [4 /*yield*/, this.messagingChannel.assertExchange(eventName, "fanout", { durable: true, })];
                    case 1:
                        // http://www.squaremobius.net/amqp.node/channel_api.html#channel_assertExchange
                        _a.sent();
                        console.log('sendMessage:'); //TODO: Logging.
                        console.log("    " + eventName);
                        console.log(eventArgs);
                        // http://www.squaremobius.net/amqp.node/channel_api.html#channel_publish
                        this.messagingChannel.publish(eventName, '', new Buffer(JSON.stringify(eventArgs)), {
                            persistent: true,
                        }); //TODO: Probably a more efficient way to do this! Maybe BSON?
                        return [2 /*return*/];
                }
            });
        });
    };
    /**
     * Starts the microservice.
     * It starts listening for incoming HTTP requests and events.
     */
    MicroService.prototype.start = function () {
        return __awaiter(this, void 0, void 0, function () {
            return __generator(this, function (_a) {
                switch (_a.label) {
                    case 0: return [4 /*yield*/, this.startHttpServer()];
                    case 1:
                        _a.sent();
                        return [4 /*yield*/, this.startMessaging()];
                    case 2:
                        _a.sent(); //TODO: Would be good to optionally enable this.
                        return [2 /*return*/];
                }
            });
        });
    };
    return MicroService;
}());
/**
 * Instantiates a microservice.
 *
 * @param [config] Optional configuration for the microservice.
 */
function micro(config) {
    return new MicroService(config);
}
exports.micro = micro;
//# sourceMappingURL=index.js.map