Example of standalone Apache Qpid (amqp) Junit Test

Does anyone have an example of using Apache Qpid within a standa开发者_运维问答lone junit test.

Ideally I want to be able to create a queue on the fly which I can put/get msgs within my test. So I'm not testing QPid within my test, I'll use integration tests for that, however be very useful to test methods handling msgs with having to mock out a load of services.

Here is the setup method I use for QPID 0.30 (I use this in a Spock test but should be portable to Java of Junit with no problems). This supports SSL connection, the HTTP management, and uses only in-memory startup. Startup time is sub-second. Configuration for QPID is awkward compared to using ActiveMQ for the same purpose, but QPID is AMQP compliant and allows for a smooth, neutral testing for AMQP clients (obviously the use of exchanges can not mimic RabbitMQs implementation, but for basic purposes it is sufficient)

First I created a minimal test-config.json which I put in the resources folder:

{
  "name": "${broker.name}",
  "modelVersion": "2.0",
  "defaultVirtualHost" : "default",
  "authenticationproviders" : [ {
    "name" : "passwordFile",
    "type" : "PlainPasswordFile",
    "path" : "${qpid.home_dir}${file.separator}etc${file.separator}passwd",
    "preferencesproviders" : [{
        "name": "fileSystemPreferences",
        "type": "FileSystemPreferences",
        "path" : "${qpid.work_dir}${file.separator}user.preferences.json"
    }]
  } ],
  "ports" : [  {
    "name" : "AMQP",
    "port" : "${qpid.amqp_port}",
    "authenticationProvider" : "passwordFile",
    "keyStore" : "default",
    "protocols": ["AMQP_0_10", "AMQP_0_8", "AMQP_0_9", "AMQP_0_9_1" ],
    "transports" : [ "SSL" ]
  }, {
    "name" : "HTTP",
    "port" : "${qpid.http_port}",
    "authenticationProvider" : "passwordFile",
    "protocols" : [ "HTTP" ]
  }],
  "virtualhostnodes" : [ {
    "name" : "default",
    "type" : "JSON",
    "virtualHostInitialConfiguration" : "{ \"type\" : \"Memory\" }"
  } ],
  "plugins" : [ {
    "type" : "MANAGEMENT-HTTP",
    "name" : "httpManagement"
  }],
  "keystores" : [ {
     "name" : "default",
        "password" : "password",
      "path": "${qpid.home_dir}${file.separator}keystore.jks"

    }]
}

I I also needed to create a keystore.jks file for localhost because the QPID broker and the RabbitMQ client do not like to communicate over an unencrypted channel. I also added a file called "passwd" in "integTest/resources/etc" that has this content:

guest:password

Here is the code from the unit test setup:

class level variables:

def tmpFolder = Files.createTempDir()
Broker broker

def amqpPort = PortFinder.findFreePort()
def httpPort = PortFinder.findFreePort()

def qpidHomeDir = 'src/integTest/resources/'
def configFileName = "/test-config.json"

code for the setup() method:

   def setup() {

    broker = new Broker();
    def brokerOptions = new BrokerOptions()

    File file = new File(qpidHomeDir)
    String homePath = file.getAbsolutePath();
    log.info(' qpid home dir=' + homePath)
    log.info(' qpid work dir=' + tmpFolder.absolutePath)

    brokerOptions.setConfigProperty('qpid.work_dir', tmpFolder.absolutePath);

    brokerOptions.setConfigProperty('qpid.amqp_port',"${amqpPort}")
    brokerOptions.setConfigProperty('qpid.http_port', "${httpPort}")
    brokerOptions.setConfigProperty('qpid.home_dir', homePath);


    brokerOptions.setInitialConfigurationLocation(homePath + configFileName)
    broker.startup(brokerOptions)
    log.info('broker started')
}

code for cleanup()

broker.shutdown()

To make an AMQP connection from a Rabbit MQ client:

        ConnectionFactory factory = new ConnectionFactory();
        factory.setUri("amqp://guest:password@localhost:${amqpPort}");
        factory.useSslProtocol()

        log.info('about to make connection')


        def connection = factory.newConnection();
        //get a channel for sending the "kickoff" message
        def channel = connection.createChannel();

The Qpid project has a number of tests that use an embedded broker for testing. Whilst we use a base case to handle startup shutdown you could do the following to simply integrate a broker within your tests:

public void setUp()
{
        int port=1;

// Config is actually a Configuaration File App Registry object, or Configuration Application Registry.

        ApplicationRegistry.initialise(config, port);

        TransportConnection.createVMBroker(port);        
}

public void test()
{...}

public void tearDown()
{
            TransportConnection.killVMBroker(port);
            ApplicationRegistry.remove(port);
}

Then for the connection you need to specify the conectionURL for the broker. i.e. borkerlist='vm://1'

My solution on qpid-broker @ 6.1.1, add below to pom.xml

<dependency>
    <groupId>org.apache.qpid</groupId>
    <artifactId>qpid-broker</artifactId>
    <version>6.1.1</version>
    <scope>test</scope>
</dependency>

qpid config file as:

{
  "name" : "${broker.name}",
  "modelVersion" : "6.1",
  "defaultVirtualHost" : "default",
  "authenticationproviders" : [ {
    "name" : "anonymous",
    "type" : "Anonymous"
  } ],
  "ports" : [ {
    "name" : "AMQP",
    "port" : "${qpid.amqp_port}",
    "authenticationProvider" : "anonymous",
    "virtualhostaliases" : [ {
      "name" : "defaultAlias",
      "type" : "defaultAlias"
    } ]
  } ],
  "virtualhostnodes" : [ {
    "name" : "default",
    "type" : "JSON",
    "defaultVirtualHostNode" : "true",
    "virtualHostInitialConfiguration" : "{ \"type\" : \"Memory\" }"
  } ]
}

code to start the qpid server

Broker broker = new Broker();
BrokerOptions brokerOptions = new BrokerOptions();
// I use fix port number
brokerOptions.setConfigProperty("qpid.amqp_port", "20179");
brokerOptions.setConfigurationStoreType("Memory");
// work_dir for qpid's log, configs, persist data
System.setProperty("qpid.work_dir", "/tmp/qpidworktmp");
// init config of qpid. Relative path for classloader resource or absolute path for non-resource
System.setProperty("qpid.initialConfigurationLocation", "qpid/qpid-config.json");
brokerOptions.setStartupLoggedToSystemOut(false);
broker.startup(brokerOptions);

code to stop qpid server

broker.shutdown();

Since I use anonymouse mode, client should do like:

SaslConfig saslConfig = new SaslConfig() {
    public SaslMechanism getSaslMechanism(String[] mechanisms) {
        return new SaslMechanism() {
            public String getName() {
                return "ANONYMOUS";
            }

            public LongString handleChallenge(LongString challenge, String username, String password) {
                return LongStringHelper.asLongString("");
            }
        };
    }
};
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setPort(20179);
factory.setSaslConfig(saslConfig);

Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

That's all. A little more on how to do it on other version.

You can download qpid-broker binary package from official site. After download and unzip, you can run it to test as server against your case. After your case connected server well, using commandline to generate or just copy the initial config file in QPID_WORK, remove useless id filed and use it for embedded server like above.

The most complicated thing is the authentication. You can choose PLAIN mode but you have to add the username and password in initial config. I choose anonymous mode which need a little code when connecting. For other authentication mode you have specify the password file or key/cert store, which I didnt try.

If it still not working, you can read the qpid-borker doc and Main class code in qpid-broker artifact which show how command line works for each settings.

The best I could figure out was:

PropertiesConfiguration properties = new PropertiesConfiguration();
properties.addProperty("virtualhosts.virtualhost.name", "test");
properties.addProperty("security.principal-databases.principal-database.name", "testPasswordFile");
properties.addProperty("security.principal-databases.principal-database.class", "org.apache.qpid.server.security.auth.database.PropertiesPrincipalDatabase");
ServerConfiguration config = new ServerConfiguration(properties);
ApplicationRegistry.initialise(new ApplicationRegistry(config) {
    @Override
    protected void createDatabaseManager(ServerConfiguration configuration) throws Exception {
        Properties users = new Properties();
        users.put("guest","guest");
        users.put("admin","admin");
        _databaseManager = new PropertiesPrincipalDatabaseManager("testPasswordFile", users);
    }
});
TransportConnection.createVMBroker(ApplicationRegistry.DEFAULT_INSTANCE);

With a URL of:

amqp://admin:admin@/test?brokerlist='vm://:1?sasl_mechs='PLAIN''

The big pain is with configuration and authorization. Milage may vary.