From c2458dee78d2844cccb37bd79607af61dd191229 Mon Sep 17 00:00:00 2001 From: Austin Schaffer Date: Thu, 7 Nov 2019 17:07:22 -0500 Subject: [PATCH] [DEV-5312] Convert most parts of RatingService Publish to TS --- conf/vanilla-server.json | 32 ++-- src/server/request/UserSession.d.ts | 27 +++ src/server/service/RatingServicePublish.js | 166 +++++++----------- src/server/service/RatingServicePublish.ts | 191 +++++++++++++++++++++ test/bucket/delta.ts | 90 ++++++++++ test/server/request/DataProcessorTest.ts | 2 +- 6 files changed, 394 insertions(+), 114 deletions(-) create mode 100644 src/server/service/RatingServicePublish.ts diff --git a/conf/vanilla-server.json b/conf/vanilla-server.json index da222bb..4f79ad6 100644 --- a/conf/vanilla-server.json +++ b/conf/vanilla-server.json @@ -51,16 +51,28 @@ "domain": "" }, "postRatePublish": { - "protocol": "amqp", - "hostname": "localhost", - "port": 5672, - "username": "", - "password": "", - "locale": "en_US", - "frameMax": 0, - "heartbeat": 0, - "vhost": "/", - "queueName": "postrate" + "protocol": "amqp", + "hostname": "localhost", + "port": 5672, + "username": "", + "password": "", + "locale": "en_US", + "frameMax": 0, + "heartbeat": 0, + "vhost": "/", + "exchange": "postrate" + }, + "deltaPublish": { + "protocol": "amqp", + "hostname": "localhost", + "port": 5672, + "username": "", + "password": "", + "locale": "en_US", + "frameMax": 0, + "heartbeat": 0, + "vhost": "/", + "exchange": "quoteupdate" } }, "c1export": { diff --git a/src/server/request/UserSession.d.ts b/src/server/request/UserSession.d.ts index 01937d7..f7d08ae 100644 --- a/src/server/request/UserSession.d.ts +++ b/src/server/request/UserSession.d.ts @@ -20,6 +20,9 @@ */ +import { PositiveInteger } from "../../numeric"; + + /** * Session management */ @@ -31,4 +34,28 @@ export declare class UserSession * @return true if internal user, otherwise false */ isInternal(): boolean; + + + /** + * Gets the agent id, if available + * + * @return agent id or undefined if unavailable + */ + agentId(): PositiveInteger | undefined + + + /** + * Gets the broker entity id, if available + * + * @return agent entity id or undefined if unavailable + */ + agentEntityId(): PositiveInteger | undefined + + + /** + * Gets the agent name, if available + * + * @return agent name or undefined if unavailable + */ + agentName(): string | undefined } diff --git a/src/server/service/RatingServicePublish.js b/src/server/service/RatingServicePublish.js index ce843f0..2ce2d98 100644 --- a/src/server/service/RatingServicePublish.js +++ b/src/server/service/RatingServicePublish.js @@ -1,3 +1,4 @@ +"use strict"; /** * Publishes message to queue after rating * @@ -18,13 +19,22 @@ * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ - -'use strict'; - -const { Interface, Trait } = require( 'easejs' ); -const { RatingService } = require( './RatingService' ); - - +var __extends = (this && this.__extends) || (function () { + var extendStatics = function (d, b) { + extendStatics = Object.setPrototypeOf || + ({ __proto__: [] } instanceof Array && function (d, b) { d.__proto__ = b; }) || + function (d, b) { for (var p in b) if (b.hasOwnProperty(p)) d[p] = b[p]; }; + return extendStatics(d, b); + }; + return function (d, b) { + extendStatics(d, b); + function __() { this.constructor = d; } + d.prototype = b === null ? Object.create(b) : (__.prototype = b.prototype, new __()); + }; +})(); +Object.defineProperty(exports, "__esModule", { value: true }); +var RatingService_1 = require("./RatingService"); +var amqplib_1 = require("amqplib"); /** * Publish message to a queue after rating * @@ -50,35 +60,11 @@ const { RatingService } = require( './RatingService' ); * * See the body of `#_sendMessage' for their values. */ -module.exports = Trait( 'RatingServicePublish' ) - .implement( Interface( { 'postProcessRaterData': [] } ) ) - .extend( -{ - /** - * AMQP library (amqplib API) - * - * @type {amqplib} - */ - 'private _amqp': null, - - /** - * AMQP configuration - * - * This should be the configuration expected by amqplib's #connect. It - * should additionally contain a `queueName' field. - * - * @type {Object} - */ - 'private _conf': {}, - - /** - * Logger - * - * @type {DebugLog} - */ - 'private _log': null, - - +var RatingServicePublish = /** @class */ (function (_super) { + __extends(RatingServicePublish, _super); + function RatingServicePublish() { + return _super !== null && _super.apply(this, arguments) || this; + } /** * Initialize trait * @@ -86,14 +72,13 @@ module.exports = Trait( 'RatingServicePublish' ) * @param {Object} conf AMQP configuration * @param {DebugLog} logger logger instance */ - __mixin( amqp, conf, logger ) - { - this._amqp = amqp; - this._conf = conf; - this._log = logger; - }, - - + RatingServicePublish.prototype.__mixin = function ( + // constructor( + // private readonly _amqp: Connection, + _conf) { + // super(); + this._conf = _conf; + }; /** * Publish quote message to exchange post-rating * @@ -105,44 +90,25 @@ module.exports = Trait( 'RatingServicePublish' ) * * @return {undefined} */ - 'abstract override postProcessRaterData'( - request, data, actions, program, quote - ) - { + RatingServicePublish.prototype.postProcessRaterData = function (request, data, actions, program, quote) { + var _this = this; // check both as we transition from one to the other - const exchange = this._conf.exchange || this._conf.queueName; - - this._amqp.connect( this._conf ) - .then( conn => - { - setTimeout( () => conn.close(), 10000 ); - return conn.createChannel(); - } ) - .then( ch => { - ch.assertExchange( exchange, 'fanout', { durable: true } ); - - return this._sendMessage( - ch, - exchange, - request.getSession(), - quote - ); - } ) - .then( () => this._log.log( - this._log.PRIORITY_INFO, - "Published quote " + quote.getId() + - " to post-rate exchange '" + exchange + "'" - ) ) - .catch( e => this._log.log( - this._log.PRIORITY_ERROR, - "Post-rate exchange publish failure for quote " + - quote.getId() + ": " + e.message - ) ); - - this.__super( request, data, actions, program, quote ); - }, - - + var exchange = this._conf.exchange; + amqplib_1.connect(this._conf) + .then(function (conn) { + setTimeout(function () { return conn.close(); }, 10000); + return conn.createChannel(); + }) + .then(function (ch) { + ch.assertExchange(exchange, 'fanout', { durable: true }); + return _this._sendMessage(ch, exchange, request.getSession(), quote); + }) + .then(function () { return _this._logger.log(_this._logger.PRIORITY_INFO, "Published quote " + quote.getId() + + " to post-rate exchange '" + exchange + "'"); }) + .catch(function (e) { return _this._logger.log(_this._logger.PRIORITY_ERROR, "Post-rate exchange publish failure for quote " + + quote.getId() + ": " + e.message); }); + _super.prototype.postProcessRaterData.call(this, request, data, actions, program, quote); + }; /** * Send message to exchange * @@ -151,31 +117,25 @@ module.exports = Trait( 'RatingServicePublish' ) * @param {UserSession} session user session * @param {Quote} quote rated quote * - * @return {Promise} whether publish was successful + * @return whether publish was successful */ - 'private _sendMessage'( channel, exchange, session, quote ) - { - const headers = { + RatingServicePublish.prototype._sendMessage = function (channel, exchange, session, quote) { + var headers = { version: 1, created: Date.now(), }; - - const data = new Buffer( JSON.stringify( { - quote_id: quote.getId(), - program_id: quote.getProgramId(), - agent_id: session.agentId(), - entity_id: session.agentEntityId(), + var data = new Buffer(JSON.stringify({ + quote_id: quote.getId(), + program_id: quote.getProgramId(), + agent_id: session.agentId(), + entity_id: session.agentEntityId(), entity_name: session.agentName(), - } ) ); - + })); // we don't use a routing key; fanout exchange - const routing_key = ''; - - return channel.publish( - exchange, - routing_key, - data, - { headers: headers } - ); - }, -} ); + var routing_key = ''; + return channel.publish(exchange, routing_key, data, { headers: headers }); + }; + return RatingServicePublish; +}(RatingService_1.RatingService)); +exports.RatingServicePublish = RatingServicePublish; +//# sourceMappingURL=data:application/json;base64,eyJ2ZXJzaW9uIjozLCJmaWxlIjoiUmF0aW5nU2VydmljZVB1Ymxpc2guanMiLCJzb3VyY2VSb290IjoiIiwic291cmNlcyI6WyJSYXRpbmdTZXJ2aWNlUHVibGlzaC50cyJdLCJuYW1lcyI6W10sIm1hcHBpbmdzIjoiO0FBQUE7Ozs7Ozs7Ozs7Ozs7Ozs7Ozs7R0FtQkc7Ozs7Ozs7Ozs7Ozs7OztBQUdILGlEQUFnRDtBQVFoRCxtQ0FJaUI7QUFRakI7Ozs7Ozs7Ozs7Ozs7Ozs7Ozs7Ozs7OztHQXdCRztBQUNIO0lBQTBDLHdDQUFhO0lBQXZEOztJQTJIQSxDQUFDO0lBOUdHOzs7Ozs7T0FNRztJQUNILHNDQUFPLEdBQVA7SUFDQSxlQUFlO0lBQ1gsd0NBQXdDO0lBQ3hDLEtBQW1CO1FBRW5CLFdBQVc7UUFDWCxJQUFJLENBQUMsS0FBSyxHQUFHLEtBQUssQ0FBQztJQUN2QixDQUFDO0lBR0Q7Ozs7Ozs7Ozs7TUFVRTtJQUNPLG1EQUFvQixHQUE5QixVQUNLLE9BQW9CLEVBQ3BCLElBQW1CLEVBQ25CLE9BQXNCLEVBQ3RCLE9BQWdCLEVBQ2hCLEtBQXdCO1FBTDdCLGlCQXVDRTtRQS9CRyxvREFBb0Q7UUFDcEQsSUFBTSxRQUFRLEdBQUcsSUFBSSxDQUFDLEtBQUssQ0FBQyxRQUFRLENBQUM7UUFFckMsaUJBQVcsQ0FBRSxJQUFJLENBQUMsS0FBSyxDQUFFO2FBQ3BCLElBQUksQ0FBRSxVQUFBLElBQUk7WUFFUCxVQUFVLENBQUUsY0FBTSxPQUFBLElBQUksQ0FBQyxLQUFLLEVBQUUsRUFBWixDQUFZLEVBQUUsS0FBSyxDQUFFLENBQUM7WUFDeEMsT0FBTyxJQUFJLENBQUMsYUFBYSxFQUFFLENBQUM7UUFDaEMsQ0FBQyxDQUFFO2FBQ0YsSUFBSSxDQUFFLFVBQUEsRUFBRTtZQUNMLEVBQUUsQ0FBQyxjQUFjLENBQUUsUUFBUSxFQUFFLFFBQVEsRUFBRSxFQUFFLE9BQU8sRUFBRSxJQUFJLEVBQUUsQ0FBRSxDQUFDO1lBRTNELE9BQU8sS0FBSSxDQUFDLFlBQVksQ0FDcEIsRUFBRSxFQUNGLFFBQVEsRUFDUixPQUFPLENBQUMsVUFBVSxFQUFFLEVBQ3BCLEtBQUssQ0FDUixDQUFDO1FBQ04sQ0FBQyxDQUFFO2FBQ0YsSUFBSSxDQUFFLGNBQU0sT0FBQSxLQUFJLENBQUMsT0FBTyxDQUFDLEdBQUcsQ0FDekIsS0FBSSxDQUFDLE9BQU8sQ0FBQyxhQUFhLEVBQzFCLGtCQUFrQixHQUFHLEtBQUssQ0FBQyxLQUFLLEVBQUU7WUFDOUIsMEJBQTBCLEdBQUcsUUFBUSxHQUFHLEdBQUcsQ0FDbEQsRUFKWSxDQUlaLENBQUU7YUFDRixLQUFLLENBQUUsVUFBQSxDQUFDLElBQUksT0FBQSxLQUFJLENBQUMsT0FBTyxDQUFDLEdBQUcsQ0FDekIsS0FBSSxDQUFDLE9BQU8sQ0FBQyxjQUFjLEVBQzNCLCtDQUErQztZQUMzQyxLQUFLLENBQUMsS0FBSyxFQUFFLEdBQUcsSUFBSSxHQUFHLENBQUMsQ0FBQyxPQUFPLENBQ3ZDLEVBSlksQ0FJWixDQUFFLENBQUM7UUFFUixpQkFBTSxvQkFBb0IsWUFBRSxPQUFPLEVBQUUsSUFBSSxFQUFFLE9BQU8sRUFBRSxPQUFPLEVBQUUsS0FBSyxDQUFFLENBQUM7SUFDekUsQ0FBQztJQUdEOzs7Ozs7Ozs7T0FTRztJQUNILDJDQUFZLEdBQVosVUFDSSxPQUFpQixFQUNqQixRQUFnQixFQUNoQixPQUFxQixFQUNyQixLQUF5QjtRQUd6QixJQUFNLE9BQU8sR0FBRztZQUNaLE9BQU8sRUFBRSxDQUFDO1lBQ1YsT0FBTyxFQUFFLElBQUksQ0FBQyxHQUFHLEVBQUU7U0FDdEIsQ0FBQztRQUVGLElBQU0sSUFBSSxHQUFHLElBQUksTUFBTSxDQUFFLElBQUksQ0FBQyxTQUFTLENBQUU7WUFDckMsUUFBUSxFQUFLLEtBQUssQ0FBQyxLQUFLLEVBQUU7WUFDMUIsVUFBVSxFQUFHLEtBQUssQ0FBQyxZQUFZLEVBQUU7WUFDakMsUUFBUSxFQUFLLE9BQU8sQ0FBQyxPQUFPLEVBQUU7WUFDOUIsU0FBUyxFQUFJLE9BQU8sQ0FBQyxhQUFhLEVBQUU7WUFDcEMsV0FBVyxFQUFFLE9BQU8sQ0FBQyxTQUFTLEVBQUU7U0FDbkMsQ0FBRSxDQUFFLENBQUM7UUFFTiw4Q0FBOEM7UUFDOUMsSUFBTSxXQUFXLEdBQUcsRUFBRSxDQUFDO1FBRXZCLE9BQU8sT0FBTyxDQUFDLE9BQU8sQ0FDbEIsUUFBUSxFQUNSLFdBQVcsRUFDWCxJQUFJLEVBQ0osRUFBRSxPQUFPLEVBQUUsT0FBTyxFQUFFLENBQ3ZCLENBQUM7SUFDTixDQUFDO0lBQ0wsMkJBQUM7QUFBRCxDQUFDLEFBM0hELENBQTBDLDZCQUFhLEdBMkh0RDtBQTNIWSxvREFBb0IifQ== \ No newline at end of file diff --git a/src/server/service/RatingServicePublish.ts b/src/server/service/RatingServicePublish.ts new file mode 100644 index 0000000..1966759 --- /dev/null +++ b/src/server/service/RatingServicePublish.ts @@ -0,0 +1,191 @@ +/** + * Publishes message to queue after rating + * + * Copyright (C) 2010-2019 R-T Specialty, LLC. + * + * This file is part of liza. + * + * liza is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + + +import { RatingService } from "./RatingService"; +import { UserRequest } from "../request/UserRequest"; +import { RateResult } from "../rater/Rater"; +import { ClientActions } from "../../client/action/ClientAction"; +import { Program } from "../../program/Program"; +import { ServerSideQuote } from "../quote/ServerSideQuote"; +import { UserSession } from "../request/UserSession"; + +import { + connect as amqpConnect, + Options, + Channel, +} from 'amqplib'; + + +export interface AmqpConfig extends Options.Connect { + /** The name of a queue or exchange to publish to */ + exchange: string; +} + +/** + * Publish message to a queue after rating + * + * This is an initial proof-of-concept implementation. In particular, we + * have the following considerations: + * + * - A fresh connection is made for each message until we can ensure that + * we can auto-reconnect on failure; + * - This trait is not yet tested; + * - It does not use an exchange; + * - It does a poor job checking for and reporting errors. + * + * The message consists of a `version' header that is set to 1. Future + * changes to the message format will bump this version. There is also a + * `created' header holding a Unix timestamp of the moment that the message + * was created. + * + * Version 1 of the body consists of four fields: + * - quote_id + * - agent_id + * - entity_id + * - entity_name + * + * See the body of `#_sendMessage' for their values. + */ +export class RatingServicePublish extends RatingService +{ + /** + * AMQP configuration + * + * This should be the configuration expected by amqplib's #connect. It + * should additionally contain a `queueName' field. + * + * @type {Object} + */ + private _conf: AmqpConfig; + + + /** + * Initialize trait + * + * @param {amqplib} AMQP library + * @param {Object} conf AMQP configuration + * @param {DebugLog} logger logger instance + */ + __mixin( + // constructor( + // private readonly _amqp: Connection, + _conf: AmqpConfig, + ) { + // super(); + this._conf = _conf; + } + + + /** + * Publish quote message to exchange post-rating + * + * @param {UserRequest} request user request + * @param {Object} data rating data returned + * @param {Array} actions actions to send to client + * @param {Program} program program used to perform rating + * @param {Quote} quote quote used for rating + * + * @return {undefined} + */ + protected postProcessRaterData( + request: UserRequest, + data: RateResult, + actions: ClientActions, + program: Program, + quote: ServerSideQuote, + ): void + { + // check both as we transition from one to the other + const exchange = this._conf.exchange; + + amqpConnect( this._conf ) + .then( conn => + { + setTimeout( () => conn.close(), 10000 ); + return conn.createChannel(); + } ) + .then( ch => { + ch.assertExchange( exchange, 'fanout', { durable: true } ); + + return this._sendMessage( + ch, + exchange, + request.getSession(), + quote + ); + } ) + .then( () => this._logger.log( + this._logger.PRIORITY_INFO, + "Published quote " + quote.getId() + + " to post-rate exchange '" + exchange + "'" + ) ) + .catch( e => this._logger.log( + this._logger.PRIORITY_ERROR, + "Post-rate exchange publish failure for quote " + + quote.getId() + ": " + e.message + ) ); + + super.postProcessRaterData( request, data, actions, program, quote ); + } + + + /** + * Send message to exchange + * + * @param {Channel} channel AMQP channel + * @param {string} exchange exchange name + * @param {UserSession} session user session + * @param {Quote} quote rated quote + * + * @return whether publish was successful + */ + _sendMessage( + channel: Channel, + exchange: string, + session: UserSession, + quote: ServerSideQuote + ): boolean + { + const headers = { + version: 1, + created: Date.now(), + }; + + const data = new Buffer( JSON.stringify( { + quote_id: quote.getId(), + program_id: quote.getProgramId(), + agent_id: session.agentId(), + entity_id: session.agentEntityId(), + entity_name: session.agentName(), + } ) ); + + // we don't use a routing key; fanout exchange + const routing_key = ''; + + return channel.publish( + exchange, + routing_key, + data, + { headers: headers } + ); + } +} diff --git a/test/bucket/delta.ts b/test/bucket/delta.ts index ba1d192..a729f29 100644 --- a/test/bucket/delta.ts +++ b/test/bucket/delta.ts @@ -89,6 +89,96 @@ describe( 'Delta', () => dest_data: { foo: [ '' ] }, expected: {}, }, + { + label: "", + src_data: { foo: [ 'bar', 'baz' ] }, + dest_data: { foo: [ 'change', 'baz' ] }, + data: { foo: [ 'change', 'baz' ] }, + diff: { foo: [ 'change' ] }, + }, + { + label: "", + src_data: { foo: [ 'bar', 'baz' ] }, + dest_data: { foo: [ 'bar', 'change' ] }, + data: { foo: [ 'bar', 'change' ] }, + diff: { foo: [ , 'change' ] }, + }, + { + label: "", + src_data: { foo: [ 'bar', 'baz' ] }, + dest_data: { foo: [ undefined, 'change' ] }, + data: { foo: [ 'bar', 'change' ] }, + diff: { foo: [ , 'change' ] }, + }, + { + label: "", + src_data: { foo: [ 'bar', 'baz' ] }, + dest_data: { foo: [ undefined, 'baz' ] }, + data: { foo: [ 'bar', 'baz' ] }, + diff: {}, + }, + { + label: "", + src_data: { foo: [ 'bar', 'baz' ] }, + dest_data: { foo: [ 'bar', undefined ] }, + data: { foo: [ 'bar', 'baz' ] }, + diff: {}, + }, + { + label: "", + src_data: { foo: [ 'bar', 'baz' ] }, + dest_data: { foo: [ 'bar', null ] }, + data: { foo: [ 'bar' ] }, + diff: { foo: [ , null ] }, + }, + { + label: "", + src_data: { foo: [ 'bar', 'baz' ] }, + dest_data: { foo: [ 'bar', 'baz', null ] }, + data: { foo: [ 'bar', 'baz' ] }, + diff: {}, + }, + { + label: "", + src_data: { foo: [ 'bar', 'baz' ] }, + dest_data: { foo: [ 'bar', 'baz', 'quux' ] }, + data: { foo: [ 'bar', 'baz', 'quux' ] }, + diff: { foo: [ , , 'quux' ]}, + }, + { + label: "", + src_data: { foo: [ 'bar', 'baz' ] }, + dest_data: { foo: [] }, + data: { foo: [ 'bar', 'baz' ] }, + diff: {}, + }, + + // null not at end of set means unchanged + { + label: "", + src_data: { foo: [ 'bar', 'baz', 'quux' ] }, + dest_data: { foo: [ null, null, 'quux' ] }, + data: { foo: [ 'bar', 'baz', 'quux' ] }, + diff: {}, + }, + // but the last one is + { + label: "", + src_data: { foo: [ 'bar', 'baz', 'quux' ] }, + dest_data: { foo: [ null, 'baz', null ] }, + data: { foo: [ 'bar', 'baz' ] }, + diff: { foo: [ , , null ] }, + }, + // given a string of nulls, only the last one is terminating; the + // rest are interpreted as undefined (because JSON serializes + // undefined values to `null' -_-) + { + label: "", + src_data: { foo: [ 'bar', 'baz', 'quux' ] }, + dest_data: { foo: [ null, null, null ] }, + data: { foo: [ 'bar', 'baz' ] }, + diff: { foo: [ , , null ] }, + }, ] ).forEach( ( { label, src_data, dest_data, expected } ) => { it( label, () => diff --git a/test/server/request/DataProcessorTest.ts b/test/server/request/DataProcessorTest.ts index d0e52d8..a3a4d0e 100644 --- a/test/server/request/DataProcessorTest.ts +++ b/test/server/request/DataProcessorTest.ts @@ -503,7 +503,7 @@ function createStubs( function createStubUserRequest( internal: boolean ) { - return { + return { getSession: () => ( { isInternal: () => internal } )