import { Channel , ConfirmChannel , ConsumeMessage } from 'amqplib' ; import chai from 'chai' ; import chaiJest from 'chai-jest' ; import { once } from 'events' ; import { defer , timeout } from 'promise-tools' ; import amqp , { AmqpConnectionManagerClass as AmqpConnectionManager } from '../src' ; import { IAmqpConnectionManager } from '../src/AmqpConnectionManager' ; chai . use ( chaiJest ) ; const { expect } = chai ; /**
 * Tests in this file assume you have a RabbitMQ instance running on localhost.
 * You can start one with:
 *
 *   docker-compose up -d
 *
 */ describe ( 'Integration tests' , ( ) => { let connection ; afterEach ( async ( ) => { await connection ?. close ( ) ; } ) ; it ( 'should connect to the broker' , async ( ) => { connection = amqp . connect ( [ 'amqp://localhost' ] ) ; await timeout ( once ( connection , 'connect' ) , 3000 ) ; } ) ; it ( 'should connect to the broker with a username and password' , async ( ) => { connection = amqp . connect ( [ 'amqp://guest:guest@localhost:5672' ] ) ; await timeout ( once ( connection , 'connect' ) , 3000 ) ; } ) ; it ( 'should connect to the broker with a string' , async ( ) => { connection = amqp . connect ( 'amqp://guest:guest@localhost:5672' ) ; await timeout ( once ( connection , 'connect' ) , 3000 ) ; } ) ; it ( 'should connect to the broker with a amqp.Connect object' , async ( ) => { connection = amqp . connect ( { protocol : 'amqp' , hostname : 'localhost' , port : 5672 , vhost : '/' , } ) ; await timeout ( once ( connection , 'connect' ) , 3000 ) ; } ) ; it ( 'should connect to the broker with an url/options object' , async ( ) => { connection = amqp . connect ( { url : 'amqp://guest:guest@localhost:5672' , } ) ; await timeout ( once ( connection , 'connect' ) , 3000 ) ; } ) ; it ( 'should connect to the broker with a string with options' , async ( ) => { connection = amqp . connect ( 'amqp://guest:guest@localhost:5672/%2F?heartbeat=10&channelMax=100' ) ; await timeout ( once ( connection , 'connect' ) , 3000 ) ; } ) ; it ( 'should throw on awaited connect with wrong password' , async ( ) => { connection = new AmqpConnectionManager ( 'amqp://guest:wrong@localhost' ) ; let err ; try { await connection . connect ( ) ; } catch ( error ) { err = error ; } expect ( err . message ) . to . contain ( 'ACCESS-REFUSED' ) ; } ) ; it ( 'send and receive messages' , async ( ) => { const queueName = 'testQueue1' ; const content = `hello world - ${ Date . now ( ) }` ; connection = amqp . connect ( [ 'amqp://localhost' ] ) ; const sendChannel = connection . createChannel ( { setup : async ( channel ) => { await channel . assertQueue ( queueName , { durable : false , autoDelete : true } ) ; } , } ) ; const rxPromise = defer < ConsumeMessage > ( ) ; const receiveWrapper = connection . createChannel ( { setup : async ( channel ) => { await channel . assertQueue ( queueName , { durable : false , autoDelete : true } ) ; await channel . consume ( queueName , ( message ) => { if ( ! message ) { } else if ( message . content . toString ( ) === content ) { rxPromise . resolve ( message ) ; } else { console . log ( `Received message ${ message ?. content . toString ( ) } !== ${ content }` ) ; } } , { noAck : true , } ) ; } , } ) ; await timeout ( once ( connection , 'connect' ) , 3000 ) ; await sendChannel . sendToQueue ( queueName , content ) ; const result = await timeout ( rxPromise . promise , 3000 ) ; expect ( result . content . toString ( ) ) . to . equal ( content ) ; await sendChannel . close ( ) ; await receiveWrapper . close ( ) ; } ) ; it ( 'send and receive messages with plain channel' , async ( ) => { const queueName = 'testQueue2' ; const content = `hello world - ${ Date . now ( ) }` ; connection = new AmqpConnectionManager ( 'amqp://localhost' ) ; const sendChannel = connection . createChannel ( { confirm : false , setup : async ( channel ) => { await channel . assertQueue ( queueName , { durable : false , autoDelete : true } ) ; } , } ) ; const receiveChannel = connection . createChannel ( { confirm : false , setup : async ( channel ) => { await channel . assertQueue ( queueName , { durable : false , autoDelete : true } ) ; } , } ) ; await connection . connect ( ) ; const rxPromise = defer < ConsumeMessage > ( ) ; await receiveChannel . consume ( queueName , ( message ) => { rxPromise . resolve ( message ) ; } ) ; await sendChannel . sendToQueue ( queueName , content ) ; const result = await timeout ( rxPromise . promise , 3000 ) ; expect ( result . content . toString ( ) ) . to . equal ( content ) ; await sendChannel . close ( ) ; await receiveChannel . close ( ) ; } ) ; it ( 'RPC' , async ( ) => { const queueName = 'testQueueRpc' ; connection = amqp . connect ( [ 'amqp://localhost' ] ) ; let rpcClientQueueName = '' ; const result = defer < string | undefined > ( ) ; const rpcClient = connection . createChannel ( { setup : async ( channel ) => { const qok = await channel . assertQueue ( '' , { exclusive : true } ) ; rpcClientQueueName = qok . queue ; await channel . consume ( rpcClientQueueName , ( message ) => { result . resolve ( message ?. content . toString ( ) ) ; } , { noAck : true } ) ; } , } ) ; const rpcServer = connection . createChannel ( { setup : async ( channel ) => { await channel . assertQueue ( queueName , { durable : false , autoDelete : true } ) ; await channel . prefetch ( 1 ) ; await channel . consume ( queueName , ( message ) => { if ( message ) { channel . sendToQueue ( message . properties . replyTo , Buffer . from ( 'world' ) , { correlationId : message . properties . correlationId , } ) ; } } , { noAck : true } ) ; } , } ) ; await timeout ( once ( connection , 'connect' ) , 3000 ) ; await timeout ( rpcClient . waitForConnect ( ) , 3000 ) ; await timeout ( rpcServer . waitForConnect ( ) , 3000 ) ; await rpcClient . sendToQueue ( queueName , 'hello' , { correlationId : 'test' , replyTo : rpcClientQueueName , messageId : 'asdkasldk' , } ) ; const reply = await result . promise ; expect ( reply ) . to . equal ( 'world' ) ; await rpcClient . close ( ) ; await rpcServer . close ( ) ; } ) ; it ( 'direct-reply-to' , async ( ) => { const rpcClientQueueName = 'amq.rabbitmq.reply-to' ; const queueName = 'testQueueRpc' ; connection = amqp . connect ( [ 'amqp://localhost' ] ) ; const result = defer < string | undefined > ( ) ; connection . on ( 'disconnect' , ( { err } ) => { if ( err ) { console . log ( err ) ; } } ) ; const rpcClient = connection . createChannel ( { setup : async ( channel ) => { await channel . consume ( rpcClientQueueName , ( message ) => { result . resolve ( message ?. content . toString ( ) ) ; } , { noAck : true } ) ; } , } ) ; const rpcServer = connection . createChannel ( { setup : async ( channel ) => { await channel . assertQueue ( queueName , { durable : false , autoDelete : true } ) ; await channel . prefetch ( 1 ) ; await channel . consume ( queueName , ( message ) => { if ( message ) { channel . sendToQueue ( message . properties . replyTo , Buffer . from ( 'world' ) , { correlationId : message . properties . correlationId , } ) ; } } , { noAck : true } ) ; } , } ) ; await timeout ( once ( connection , 'connect' ) , 3000 ) ; await timeout ( rpcServer . waitForConnect ( ) , 3000 ) ; await timeout ( rpcClient . waitForConnect ( ) , 3000 ) ; await rpcClient . sendToQueue ( queueName , 'hello' , { correlationId : 'test' , replyTo : rpcClientQueueName , messageId : 'asdkasldk' , } ) ; const reply = await result . promise ; expect ( reply ) . to . equal ( 'world' ) ; await rpcClient . close ( ) ; await rpcServer . close ( ) ; } ) ; } ) ;