FlowType MySql DataClient

DataClient.js

// @flow
// @author tyler

import mysql from 'mysql';

import type {IConfigReader} from '../ConfigReader';

export type TDataRow = *;

export type TDataClientType = 'mysql' | 'sqlite3';

export interface ITransaction {
  query (query: string): Promise<TDataRow>;
}

export interface IDataClient {
  /**
   * Query the data store for a specific result row
   * @param query {string}
   * @returns {Promise<TDataRow>}
   */
  query (query: string): Promise<TDataRow>;

  /**
   * Provides an interface for transactions
   * @summary Automatically commits or rollback based on
   * included promise result. E.G. transaction(t => return t.query('blah'))
   * @param fn
   * @returns {Promise<*>}
   */
  transaction (fn: (t: ITransaction) => Promise<*>): Promise<*>;

  /**
   * Returns the client name
   * @returns {TDataClientType}
   */
  getClient (): TDataClientType;
}

export class MySqlDataClient {
  _pool: *;

  constructor (configReader: IConfigReader) {
    const {db} = configReader.get();

    this._pool = mysql.createPool({
      connectionLimit: db.poolSize,
      host: db.host,
      port: db.port,
      database: db.database,
      password: db.password,
      user: db.user
    });
  }

  query (query: string): Promise<TDataRow> {
    return new Promise((resolve, reject) => {
      // get a connection
      this._pool.getConnection((err, connection) => {
        if (err) {
          reject(err);
          return;
        }

        connection.query(query, (err, rows) => {
          connection.release();

          if (err) {
            reject(err);
            return;
          }

          resolve(rows);
        });
      });
    });
  }

  /**
   * Provides a transaction interface
   * @param fn
   * @returns {Promise<*>}
   */
  transaction (fn: (t: ITransaction) => Promise<*>): Promise<*> {
    return new Promise((resolve, reject) => {
      // get a connection
      this._pool.getConnection((err, connection) => {
        if (err) {
          reject(err);
          return;
        }

        // begin transaction
        connection.beginTransaction(err => {
          if (err) {
            reject(err);
            return;
          }

          // create transaction interface
          const transaction: ITransaction = {
            query (query: string): Promise<TDataRow> {
              return new Promise((resolve, reject) => {
                connection.query(query, (err, rows) => {
                  if (err) {
                    reject(err);
                    return;
                  }

                  resolve(rows);
                });
              });
            }
          };

          // pass interface to caller
          fn(transaction)
            .then(result => {
              // commit transaction
              connection.commit(err => {
                if (err) {
                  throw err;
                }

                connection.release();

                resolve(result);
              });
            })
            .catch(err => {
              // rollback transaction
              connection.rollback(() => {
                connection.release();

                reject(err);
              });
            });
        });
      });
    });
  }

  /**
   * Return the client type "mysql"
   * @returns {TDataClientType}
   */
  getClient (): TDataClientType {
    return 'mysql';
  }
}

DataClient.test.js

// @flow
// @author tyler

import proxyquire from 'proxyquire';
import sinon from 'sinon';

import type {TNodeunitTest} from '../types';
import type {TConfig, IConfigReader} from '../../src/ConfigReader';
import type {ITransaction, IDataClient} from '../../src/data/DataClient';

const connectionMock = {
  release: sinon.stub(),
  query: sinon.stub().callsArgWith(1, null, []),
  beginTransaction: sinon.stub().callsArgWith(0, null),
  commit: sinon.stub().callsArgWith(0, null),
  rollback: sinon.stub().callsArgWith(0)
};

const poolMock = {
  getConnection: sinon.stub().callsArgWith(0, null, connectionMock)
};

const mysql = {
  createPool: sinon.stub().returns(poolMock)
};

const {MySqlDataClient} = proxyquire('../../src/data/DataClient', {
  mysql
});

export const DataClientTests = {
  setUp (cb: () => void) {
    const configReader: IConfigReader = {
      get (): TConfig {
       return {
         port: 3333,
         db: {
           host: 'localhost',
           port: 1234,
           user: 'tyler',
           password: 'shaddix',
           poolSize: 10,
           database: 'z'
         }
       };
      }
    };

    this.dataClient = new MySqlDataClient(configReader);

    cb();
  },

  tearDown (cb: () => void) {
    mysql.createPool.reset();
    connectionMock.release.reset();
    connectionMock.query.reset();
    connectionMock.beginTransaction.reset();
    connectionMock.commit.reset();
    connectionMock.rollback.reset();
    poolMock.getConnection.reset();

    cb();
  },

  /**
   * Test that pool is setup correctly with config reader
   * variables
   * @param test
   */
  testPoolCreation (test: TNodeunitTest) {
    test.expect(1);

    test.deepEqual(mysql.createPool.args[0][0], {
      connectionLimit: 10,
      host: 'localhost',
      port: 1234,
      database: 'z',
      password: 'shaddix',
      user: 'tyler'
    });

    test.done();
  },

  /**
   * Test that a basic query runs and is torn down
   * @param test
   */
  testBasicQuery (test: TNodeunitTest) {
    test.expect(3);

    this.dataClient.query('select * from users')
      .then(rows => {
        // confirm new connection opened
        test.ok(poolMock.getConnection.calledOnce);

        // confirm query was sent on connection
        test.ok(
          connectionMock.query.calledWith('select * from users')
        );

        // confirm connection was released
        test.ok(
          connectionMock.release.calledOnce
        );

        test.done();
      })
      .catch(err => {
        test.ifError(err);
        test.done();
      });
  },

  testBasicTransaction (test: TNodeunitTest) {
    test.expect(8);

    const result = {test: 'test'};

    this.dataClient
      .transaction(t => {
        return t
          .query('select * from users')
          .then(rows => {
            return t.query('insert id = 2 into users');
          })
          .then(rows => {
            return result;
          });
      })
      .then(data => {
        // confirm it returned inner promise value
        test.deepEqual(data, result);

        // confirm query was sent on connection
        test.ok(
          connectionMock.query.calledWith('select * from users')
        );

        // confirm query was sent on connection
        test.ok(
          connectionMock.query.calledWith('insert id = 2 into users')
        );

        // confirm connection was released
        test.ok(
          connectionMock.release.calledOnce
        );

        // confirm new connection opened
        test.ok(poolMock.getConnection.calledOnce);

        // confirm transaction was started
        test.ok(
          connectionMock.beginTransaction.calledOnce
        );

        // confirm transaction was commited
        test.ok(
          connectionMock.commit.calledOnce
        );

        // confirm transaction was not rolled back
        test.ok(
          !connectionMock.rollback.called
        );

        test.done();
      })
      .catch(err => {
        test.ifError(err);
        test.done();
      });
  },

  /**
   * Test that an error will cause a transaction to rollback
   * automatically
   * @param test
   */
  testBrokenTransaction (test: TNodeunitTest) {
    test.expect(3);

    const result = {test: 'test'};

    this.dataClient
      .transaction(t => {
        return t
          .query('select * from users')
          .then(rows => {
            throw new Error('I kill transactions');
          })
          .then(rows => {
            return result;
          });
      })
      .then(data => {
        test.done();
      })
      .catch(err => {
        // confirm transaction was rolled back
        test.ok(
          connectionMock.rollback.calledOnce
        );

        // confirm connection was released
        test.ok(
          connectionMock.release.calledOnce
        );

        // confirm transaction was not commited
        test.ok(
          !connectionMock.commit.called
        );

        test.done();
      });
  }
};