server.js

const path = require('path');
const grpc = require('@grpc/grpc-js');
const pino = require('pino');
const protoLoader = require('@grpc/proto-loader');
const opentelemetry = require('@opentelemetry/api');

const MAIN_PROTO_PATH = path.join(__dirname, './proto/demo.proto');
const HEALTH_PROTO_PATH = path.join(__dirname, './proto/grpc/health/v1/health.proto');

const PORT = process.env.PORT;

const shopProto = _loadProto(MAIN_PROTO_PATH).msdemo;
const healthProto = _loadProto(HEALTH_PROTO_PATH).grpc.health.v1;

const logger = pino({
    name: 'currencyservice-server',
    messageKey: 'message',
    changeLevelName: 'severity',
    useLevelLabels: true
});

function getRandomWaitTime(max, buckets) {
    let num = 0;
    const val = max / buckets;
    for (let i = 0; i < buckets; i++) {
        num += Math.random() * val;
    }
    return num;
}

function sleepRandom(max) {
    const rnd = getRandomWaitTime(max, 4);
    return new Promise((resolve, reject) => {
        setTimeout(resolve, rnd);
    })
}

async function mockDatabaseCall(maxTime, name, query) {
    const tracer = opentelemetry.trace.getTracer("");
    const span = tracer.startSpan(name);
    span.setAttribute("db.statement", query);
    span.setAttribute("db.name", "currency");
    await sleepRandom(maxTime);
    span.end();
}

/**
 * Helper function that loads a protobuf file.
 */
function _loadProto(path) {
    const packageDefinition = protoLoader.loadSync(
        path,
        {
            keepCase: true,
            longs: String,
            enums: String,
            defaults: true,
            oneofs: true
        }
    );
    return grpc.loadPackageDefinition(packageDefinition);
}

/**
 * Helper function that gets currency data from a stored JSON file
 * Uses public data from European Central Bank
 */
function _getCurrencyData(callback) {
    const data = require('./data/currency_conversion.json');
    callback(data);
}

/**
 * Helper function that handles decimal/fractional carrying
 */
function _carry(amount) {
    const fractionSize = Math.pow(10, 9);
    amount.nanos += (amount.units % 1) * fractionSize;
    amount.units = Math.floor(amount.units) + Math.floor(amount.nanos / fractionSize);
    amount.nanos = amount.nanos % fractionSize;
    return amount;
}

/**
 * Lists the supported currencies
 */
async function getSupportedCurrencies(call, callback) {
    //await mockDatabaseCall(100, "SELECT currency.currencies", "SELECT * FROM currencies");
    await sleepRandom(20)

    logger.info('Getting supported currencies...');
    _getCurrencyData((data) => {
        callback(null, {currency_codes: Object.keys(data)});
    });
}

/**
 * Converts between currencies
 */
async function convert(call, callback) {
    //await mockDatabaseCall(100, "SELECT currency.rates", "SELECT rate FROM rates WHERE source_rate = ? AND target_rate = ?");
    await sleepRandom(30)

    logger.info('received conversion request');
    try {
        _getCurrencyData((data) => {
            const request = call.request;

            // Convert: from_currency --> EUR
            const from = request.from;
            const euros = _carry({
                units: from.units / data[from.currency_code],
                nanos: from.nanos / data[from.currency_code]
            });

            euros.nanos = Math.round(euros.nanos);

            // Convert: EUR --> to_currency
            const result = _carry({
                units: euros.units * data[request.to_code],
                nanos: euros.nanos * data[request.to_code]
            });

            result.units = Math.floor(result.units);
            result.nanos = Math.floor(result.nanos);
            result.currency_code = request.to_code;

            logger.info(`conversion request successful`);
            callback(null, result);
        });
    } catch (err) {
        logger.error(`conversion request failed: ${err}`);
        callback(err.message);
    }
}

/**
 * Endpoint for health checks
 */
function check(call, callback) {
    callback(null, {status: 'SERVING'});
}

/**
 * Starts an RPC server that receives requests for the
 * CurrencyConverter service at the sample server port
 */
function main() {
    logger.info(`Starting gRPC server on port ${PORT}...`);
    const server = new grpc.Server();
    server.addService(shopProto.CurrencyService.service, {getSupportedCurrencies, convert});
    server.addService(healthProto.Health.service, {check});
    server.bindAsync(
        `0.0.0.0:${PORT}`,
        grpc.ServerCredentials.createInsecure(),
        function () {
            logger.info(`CurrencyService gRPC server started on port ${PORT}`);
            server.start();
        },
    );
}

main();