nope/test/testAmqp.ts

import 'reflect-metadata';

import { AmqpLayer, } from "../lib/communication/amqpLayer";
import { generateBenchmarkFunction } from "../modules/funcs/generateBenchmarkFunction";
import { nopeDispatcher } from "../lib/dispatcher/nopeDispatcher";
import { promisify } from 'util'
import { getLogger } from "../lib/logger/getLogger";
import { nopeObservable } from "../lib/observables/nopeObservable";

const client1 = new AmqpLayer('localhost');
const client2 = new AmqpLayer('localhost');
const client3 = new AmqpLayer('localhost');


const sleep = promisify(setTimeout)
const logger = getLogger('info','DISPATCHER')


function genFunc(name, delay) {
    const logger = getLogger('debug',name)
    return (idx: number) => {
        return new Promise((resolve, reject) => {
            try {
                logger.info('started ' + idx.toString());
                setTimeout(() => {
                    logger.info('finished ' + idx.toString());
                    resolve();
                }, delay);
            } catch(e){
                reject(e)
            }
            
        })
    }
}


const main = async () => {

    let max = 10000;

    await sleep(1000);    
    const dispatcher1 = new nopeDispatcher({ communicator: client1, logger }, () => new nopeObservable());
    const dispatcher2 = new nopeDispatcher({ communicator: client2, logger }, () => new nopeObservable());

    const benchmark = generateBenchmarkFunction(1000, 'Executed')
    // Register Functions
    dispatcher1.registerFunction(genFunc('dispatcher-1-slow', 2000), {
        id: 'func'
    });
    dispatcher2.registerFunction(genFunc('dispatcher-2-fast', 500), {
        id: 'func'
    });
    dispatcher1.registerFunction(async () => benchmark(), {
        id: 'benchmark'
    })
    await sleep(500);
    const dispatcher3 = new nopeDispatcher({ communicator: client3, logger }, () => new nopeObservable());
    await sleep(500);
    let c = 0
    // Call Tasks Parallel!
    while (c < 10) {
        dispatcher3.performCall('func', [c++]).catch(e => console.error(e)); 
    }

    // Benchmark:
    c = 0

    // Call Tasks Parallel!
    while (c < max) {
        await dispatcher3.performCall('benchmark', [], {
            paramsHasNoCallback: true,
        });

        c++;
    }

    await sleep(500);
    const obs01 = new nopeObservable<number>();
    dispatcher1.registerObservable(obs01, {
        mode: 'publish',
        topic: 'test.var'
    });

    const obs02 = new nopeObservable<number>();
    dispatcher2.registerObservable(obs02, {
        mode: 'subscribe',
        topic: 'test.var'
    });

    obs02.subscribe(benchmark);
    obs01.setContent(1200);

    c = 0;
    while (c < max) {
        obs01.setContent(c++);
        await sleep(0)
    }


    await client1.on('subscribe','test_01', (...args) => {
        console.log('c1',...args);
    })

    await client2.on('subscribe','test_02', benchmark)

    c = 0;
    while(c < max){
        await client1.emit('publish','test_02', c++);
    }


}

main();