Merge pull request #12786 from rabbitmq/replace-java-amqp10-client-wi…

…th-javascript

Selenium suites: replace Java AMQP 1.0 client with a JavaScript one

selenium/Dockerfile

@@ -7,9 +7,6 @@ COPY package.json package.json
 
 FROM base as test
 RUN npm install
-RUN mkdir -p /code/amqp10-roundtriptest
-COPY amqp10-roundtriptest /code/amqp10-roundtriptest
-RUN mvn -f /code/amqp10-roundtriptest package
 
 ENTRYPOINT [ "npm" ]
 CMD [ "" ]

selenium/package.json

@@ -6,7 +6,6 @@
   "scripts": {
     "fakeportal": "node fakeportal/app.js",
     "fakeproxy": "node fakeportal/proxy.js",
-    "amqp10_roundtriptest": "eval $(cat $ENV_FILE ) && amqp10-roundtriptest/run",
     "test": " eval $(cat $ENV_FILE ) && mocha --recursive --trace-warnings --timeout 40000"
   },
   "keywords": [],

selenium/test/amqp.js

@@ -0,0 +1,72 @@
+var container = require('rhea')  // https://github.com/amqp/rhea
+var fs = require('fs');
+var path = require('path');
+
+function getAmqpConnectionOptions() {
+  return {
+    'host': process.env.RABBITMQ_HOSTNAME || 'rabbitmq',
+    'port': process.env.RABBITMQ_AMQP_PORT || 5672,
+    'username' : process.env.RABBITMQ_AMQP_USERNAME || 'guest',
+    'password' : process.env.RABBITMQ_AMQP_PASSWORD || 'guest',
+    'id': "selenium-connection-id",
+    'container_id': "selenium-container-id"        
+  }
+}
+function getAmqpsConnectionOptions() {
+  let options = getAmqpConnectionOptions()
+  let useMtls = process.env.AMQP_USE_MTLS || false
+  if (useMtls) {
+    options['enable_sasl_external'] = true  
+  }
+  options['transport'] = 'tls'
+  let certsLocation = getEnv("RABBITMQ_CERTS");
+  options['key'] = fs.readFileSync(path.resolve(certsLocation,'client_rabbitmq_key.pem'))
+  options['cert'] = fs.readFileSync(path.resolve(certsLocation,'client_rabbitmq_certificate.pem'))
+  options['ca'] = fs.readFileSync(path.resolve(certsLocation,'ca_rabbitmq_certificate.pem')) 
+}
+function getConnectionOptions() {
+  switch(process.env.RABBITMQ_AMQP_SCHEME || 'amqp'){
+    case 'amqp':
+      return getAmqpConnectionOptions()
+    case 'amqps':
+      return getAmqpsConnectionOptions()
+  }  
+}
+module.exports = {  
+
+  open: () => {
+    let promise = new Promise((resolve, reject) => {
+      container.on('connection_open', function(context) {
+        resolve()
+      })
+    })
+    let connection = container.connect(getConnectionOptions())
+    let receiver = connection.open_receiver({
+      source: 'examples',
+      target: 'receiver-target',
+      name: 'receiver-link'
+    })
+    let sender = connection.open_sender({
+      target: 'examples',
+      source: 'sender-source',
+      name: 'sender-link'
+    })
+    return {
+      'connection': connection,
+      'promise' : promise,
+      'receiver' : receiver,
+      'sender' : sender
+    }
+  },
+  close: (connection) => {
+    if (connection != null) {
+      connection.close()
+    }
+  },
+  once: (event, callback) => {
+    container.once(event, callback)
+  },
+  on: (event, callback) => {
+    container.on(event, callback)
+  }
+}

selenium/test/authnz-msg-protocols/amqp10.js

@@ -1,7 +1,21 @@
 const assert = require('assert')
 const { tokenFor, openIdConfiguration } = require('../utils')
 const { reset, expectUser, expectVhost, expectResource, allow, verifyAll } = require('../mock_http_backend')
-const {execSync} = require('child_process')
+const { open: openAmqp, once: onceAmqp, on: onAmqp, close: closeAmqp } = require('../amqp')
+
+var receivedAmqpMessageCount = 0
+var untilConnectionEstablished = new Promise((resolve, reject) => {
+  onAmqp('connection_open', function(context) {
+    resolve()
+  })
+})
+
+onAmqp('message', function (context) {
+    receivedAmqpMessageCount++
+})
+onceAmqp('sendable', function (context) {
+    context.sender.send({body:'first message'})    
+})
 
 const profiles = process.env.PROFILES || ""
 var backends = ""
@@ -15,10 +29,8 @@ describe('Having AMQP 1.0 protocol enabled and the following auth_backends: ' +
   let expectations = []
   let username = process.env.RABBITMQ_AMQP_USERNAME
   let password = process.env.RABBITMQ_AMQP_PASSWORD
-  let usemtls = process.env.AMQP_USE_MTLS
-  let amqpClientCommand = "npm run amqp10_roundtriptest" + 
-    (usemtls ? "" : " " + username + " " + password)
-
+  let amqp;
+
   before(function () {
     if (backends.includes("http") && username.includes("http")) {
       reset()
@@ -39,13 +51,29 @@ describe('Having AMQP 1.0 protocol enabled and the following auth_backends: ' +
     }
   })
 
-  it('can open an AMQP 1.0 connection', function () {     
-    console.log(execSync(amqpClientCommand).toString())
+  it('can open an AMQP 1.0 connection', async function () {     
+    amqp = openAmqp()
+    await untilConnectionEstablished
+    var untilMessageReceived = new Promise((resolve, reject) => {
+      onAmqp('message', function(context) {
+        resolve()
+      })
+    })
+    amqp.sender.send({body:'second message'})    
+    await untilMessageReceived
+    assert.equal(2, receivedAmqpMessageCount)
   })
 
   after(function () {
-      if ( backends.includes("http") ) {
-        verifyAll(expectations)
+    if ( backends.includes("http") ) {
+      verifyAll(expectations)
+    }
+    try {
+      if (amqp != null) {
+        closeAmqp(amqp.connection)
       }
+    } catch (error) {
+      console.error("Failed to close amqp10 connection due to " + error);      
+    }  
   })
 })