0

Hi I am testing spring integration project using an embedded broker Qpid. But the problem is that HOW CAN i make queues and exchanges in qpid. I thought that rabbit-config.xml would make the queues and exchanges in qpid broker but to no avail. My flow is create queues and exchanges in qpid broker pass messages to them and inbound amqp adapters bounded to these queues would get messages and i can proceed with the test

Error : Queue: 'push.customer.arkona.controller.search' not found on VirtualHost 'default'.

    qpid-config.json:

    {   "name": "EmbeddedBroker",   "modelVersion": "2.0",   "storeVersion" : 1,   "authenticationproviders" : [ {
        "name" : "noPassword",
        "type" : "Anonymous",
        "secureOnlyMechanisms": []
            },
        {
          "name" : "passwordFile",
          "type" : "PlainPasswordFile",
          "path" : "${qpid.home_dir}${file.separator}src${file.separator}main${file.separator}resources${file.separator}password.properties",
          "secureOnlyMechanisms": []
        }    ],   "ports" : [
        {
          "name": "AMQP",
          "port": "${qpid.amqp_port}",
          "authenticationProvider": "passwordFile",
          "protocols": [
            "AMQP_0_10",
            "AMQP_0_8",
            "AMQP_0_9",
            "AMQP_0_9_1"
          ]
        }],
        "virtualhostnodes" : [ {
        "name" : "default",
        "type" : "JSON",
        "defaultVirtualHostNode" : "true",
        "virtualHostInitialConfiguration" : "${qpid.initial_config_virtualhost_config}",
         "storeType" : "DERBY"   
} 
] 
}

password.properties has

guest:guest

I have created a separate profile for running my tests. this is the rabbitmq configuration. apart from this i have a rabbit-context xml file where all the queues, exchanges are defined.

@Configuration
@Profile("qpid")
public class QpidConfig {

    String amqpPort = "5672";

    //String qpidHomeDir = "complete";
    String configFileName = "src/main/resources/qpid-config.json";

    @Bean
    BrokerOptions brokerOptions() {

        File tmpFolder= Files.createTempDir();

        //small hack, because userDir is not same when running Application and ApplicationTest
        //it leads to some issue locating the files after, so hacking it here
        String userDir=System.getProperty("user.dir").toString();

        File file = new File(userDir);
        String homePath = file.getAbsolutePath();

        BrokerOptions brokerOptions=new BrokerOptions();

        brokerOptions.setConfigProperty("qpid.work_dir", tmpFolder.getAbsolutePath());
        brokerOptions.setConfigProperty("qpid.amqp_port",amqpPort);
        brokerOptions.setConfigProperty("qpid.home_dir", homePath);
        brokerOptions.setInitialConfigurationLocation(homePath + "/"+configFileName);

        return brokerOptions;
    }

    @SuppressWarnings("rawtypes")
    @Bean
    Broker broker() throws Exception {

            org.apache.qpid.server.Broker broker = new org.apache.qpid.server.Broker();
        broker.startup(brokerOptions());
        return (Broker) broker;
    }

    private ConnectionFactory connectionFactory() {
        ConnectionFactory factory = new ConnectionFactory();

        factory.setUsername("admin");
        factory.setPassword("admin");
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        return factory;
    }

    @Bean(name ="rabbitConnectionFactory")
    public CachingConnectionFactory rabbitConnectionFactory(){
        return new CachingConnectionFactory(connectionFactory());
    }

    @Bean(name="rabbitTemplate")
    public RabbitTemplate rabbitTemplate(){
        return new RabbitTemplate(rabbitConnectionFactory());
    }

    @Bean(name ="arkonaHeaderMapper")
    public DefaultAmqpHeaderMapper syncerHeaderMapper() {
        DefaultAmqpHeaderMapper amqpHeaderMapper = DefaultAmqpHeaderMapper.inboundMapper();
        amqpHeaderMapper.setRequestHeaderNames("*");
        amqpHeaderMapper.setReplyHeaderNames("*");
        return amqpHeaderMapper;
    }


}

EDIT

MY rabbit-context.xml 

    <?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:rabbit="http://www.springframework.org/schema/rabbit"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
    http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">

    <rabbit:queue name="pull.appt.arkona.scheduler.adapter" />
    <rabbit:queue name="pull.appt.arkona.adapter.processor" />

    <rabbit:queue name="pull.customer.arkona.to.lookup" />
    <rabbit:queue name="pull.customer.arkona.lookup.processor" />

    <rabbit:queue name="pull.customer.arkona.scheduler.adapter" />

    <rabbit:queue name="pull.ro.arkona.to.lookup" />
    <rabbit:queue name="pull.ro.arkona.adapter.processor" />

    <rabbit:queue name="pull.ro.arkona.scheduler.adapter" />

    <rabbit:queue name="pull.closed.arkona.scheduler.adapter" />
    <rabbit:queue name="pull.parts.arkona.scheduler.adapter" />

    <rabbit:queue name="pull.closed.arkona.adapter.processor" />
    <rabbit:queue name="pull.parts.arkona.adapter.processor" />

    <rabbit:queue name="pull.vehicle.arkona.to.lookup" />
    <rabbit:queue name="pull.vehicle.arkona.lookup.processor" />

    <rabbit:direct-exchange name="dms.arkona.exchange" durable="true">
        <rabbit:bindings>
            <rabbit:binding queue="pull.appt.arkona.scheduler.adapter" key="pull.appt.arkona.scheduler.adapter.key"></rabbit:binding>
            <rabbit:binding queue="pull.appt.arkona.adapter.processor" key="pull.appt.arkona.adapter.processor.key"></rabbit:binding>
            <rabbit:binding queue="pull.customer.arkona.to.lookup" key="pull.customer.arkona.to.lookup.key"></rabbit:binding>
            <rabbit:binding queue="pull.customer.arkona.lookup.processor" key="pull.customer.arkona.lookup.processor.key"></rabbit:binding>
            <rabbit:binding queue="pull.customer.arkona.scheduler.adapter" key="pull.customer.arkona.scheduler.adapter.key"></rabbit:binding>
            <rabbit:binding queue="pull.ro.arkona.to.lookup" key="pull.ro.arkona.to.lookup.key"></rabbit:binding>
            <rabbit:binding queue="pull.ro.arkona.adapter.processor" key="pull.ro.arkona.adapter.processor.key"></rabbit:binding>
            <rabbit:binding queue="pull.ro.arkona.scheduler.adapter" key="pull.ro.arkona.scheduler.adapter.key"></rabbit:binding>
            <rabbit:binding queue="pull.vehicle.arkona.to.lookup" key="pull.vehicle.arkona.to.lookup.key"></rabbit:binding>
            <rabbit:binding queue="pull.vehicle.arkona.lookup.processor" key="pull.vehicle.arkona.lookup.processor.key"></rabbit:binding>
            <rabbit:binding queue="pull.closed.arkona.scheduler.adapter" key="pull.closed.arkona.scheduler.adapter.key"></rabbit:binding>
            <rabbit:binding queue="pull.closed.arkona.adapter.processor" key="pull.closed.arkona.adapter.processor.key"></rabbit:binding>
            <rabbit:binding queue="pull.parts.arkona.scheduler.adapter" key="pull.parts.arkona.scheduler.adapter.key"></rabbit:binding>
            <rabbit:binding queue="pull.parts.arkona.adapter.processor" key="pull.parts.arkona.adapter.processor.key"></rabbit:binding>
        </rabbit:bindings>
    </rabbit:direct-exchange>
</beans>
vipulk10
  • 117
  • 3
  • 15

1 Answers1

1

Do you have a RabbitAdmin in your application context? (It detects the queues/exchanges/bindings and declares them when the connection is established).

I just tested the Spring Integration AMQP Sample with QPID 6.1.2 and it created everything ok...

<!-- Infrastructure -->

<rabbit:connection-factory id="connectionFactory" host="xx.xx.xx.xx" virtual-host="default" />

<rabbit:template id="amqpTemplate" connection-factory="connectionFactory" />

<rabbit:admin connection-factory="connectionFactory" />

<rabbit:queue name="si.test.queue" />

<rabbit:direct-exchange name="si.test.exchange">
    <rabbit:bindings>
        <rabbit:binding queue="si.test.queue" key="si.test.binding" />
    </rabbit:bindings>
</rabbit:direct-exchange>

EDIT

Boot app works fine for me too...

@SpringBootApplication
public class So50364236Application {

    public static void main(String[] args) {
        SpringApplication.run(So50364236Application.class, args);
    }

    @Bean
    public ApplicationRunner runner(RabbitTemplate template) {
        return args -> template.convertAndSend("so50364236", "foo");
    }

    @Bean
    public Queue queue() {
        return new Queue("so50364236");
    }

    @RabbitListener(queues = "so50364236")
    public void listen(String in) {
        System.out.println(in);
    }

}

and

spring.rabbitmq.addresses=xx.x.x.x
spring.rabbitmq.virtual-host=default

and

2018-05-16 13:17:25.013  INFO 34714 --- [           main] com.example.So50364236Application        : Started So50364236Application in 1.151 seconds (JVM running for 1.579)
foo

And I see the queue on the broker's admin page.

EDIT2

Here's another boot app where the queue is declared in an XML file; using QPID 6.1.6 embedded...

qpid-config.json

{
    "name": "EmbeddedBroker",
    "modelVersion": "2.0",
    "storeVersion": 1,
    "authenticationproviders": [
        {
            "name": "noPassword",
            "type": "Anonymous",
            "secureOnlyMechanisms": []
        },
        {
            "name": "passwordFile",
            "type": "PlainPasswordFile",
            "path": "${qpid.home_dir}${file.separator}etc${file.separator}passwd",
            "secureOnlyMechanisms": []
        }
    ],
    "ports": [
        {
            "name": "AMQP",
            "port": "${qpid.amqp_port}",
            "authenticationProvider": "passwordFile",
            "protocols": [
                "AMQP_0_10",
                "AMQP_0_8",
                "AMQP_0_9",
                "AMQP_0_9_1"
            ]
        }
    ],
    "virtualhostnodes": [
        {
            "name": "default",
            "type": "JSON",
            "defaultVirtualHostNode": "true",
            "virtualHostInitialConfiguration": "${qpid.initial_config_virtualhost_config}",
            "storeType": "DERBY"
        }
    ]
}

config.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:rabbit="http://www.springframework.org/schema/rabbit"
    xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">

    <rabbit:queue name="so50364236b" />

</beans>

application.properties

spring.rabbitmq.addresses=localhost:8888

Boot app

@SpringBootApplication
@ImportResource("config.xml")
public class So50364236Application {

    public static void main(String[] args) {
        new SpringApplicationBuilder(So50364236Application.class)
            .web(WebApplicationType.NONE)
            .run(args);
    }

    @Bean
    public ApplicationRunner runner(RabbitTemplate template) {
        return args -> template.convertAndSend("so50364236b", "foo");
    }

    @Bean
    BrokerOptions brokerOptions() throws Exception {

        Path tmpFolder = Files.createTempDirectory("qpidWork");
        Path homeFolder = Files.createTempDirectory("qpidHome");
        File etc = new File(homeFolder.toFile(), "etc");
        etc.mkdir();
        FileOutputStream fos = new FileOutputStream(new File(etc, "passwd"));
        fos.write("guest:guest\n".getBytes());
        fos.close();

        BrokerOptions brokerOptions = new BrokerOptions();

        brokerOptions.setConfigProperty("qpid.work_dir", tmpFolder.toAbsolutePath().toString());
        brokerOptions.setConfigProperty("qpid.amqp_port", "8888");
        brokerOptions.setConfigProperty("qpid.home_dir", homeFolder.toAbsolutePath().toString());
        Resource config = new ClassPathResource("qpid-config.json");
        brokerOptions.setInitialConfigurationLocation(config.getFile().getAbsolutePath());

        return brokerOptions;
    }

    @Bean
    Broker broker() throws Exception {
        org.apache.qpid.server.Broker broker = new org.apache.qpid.server.Broker();
        broker.startup(brokerOptions());
        return broker;
    }

    @RabbitListener(queues = "so50364236b")
    public void listen(String in) {
        System.out.println(in);
    }

}

and

[Broker] BRK-1004 : Qpid Broker Ready
received: foo

Perhaps you are doing something that causes boot's Admin to be not declared. It's not clear why you are adding your own connection factory and template; have you tried adding your own RabbitAdmin too?

Gary Russell
  • 166,535
  • 14
  • 146
  • 179
  • ok you mean to say that I need to make another rabbit-context.xml for rabbit admin? since I have already rabbit-context – vipulk10 May 16 '18 at 16:53
  • I m using spring-boot and thought it automatically creates rabbit admin bean with these bindings – vipulk10 May 16 '18 at 16:55
  • A boot app works fine for me too. See the edit to my answer. Edit your question to show your Boot code and config. – Gary Russell May 16 '18 at 17:31
  • whenever i run my test the log says it loading bean definition from rabbit-context ***main] o.s.b.f.xml.XmlBeanDefinitionReader : Loading XML bean definitions from class path resource [rabbitmq-context.xml]*** – vipulk10 May 17 '18 at 04:36
  • still getting error ***reply-code=404, reply-text=Queue: 'push.customer.arkona.controller.add' not found on VirtualHost 'default'.*** – vipulk10 May 17 '18 at 06:02
  • You're not showing all your configuration - I just updated my test to use an embedded broker similar to yours (6.1.6); and everything still works just fine. I edited my answer with the new version. – Gary Russell May 17 '18 at 15:03
  • Also you suggest not create rabbit template ,and connection factory but what values do I need to put in spring.rabbitmq.host, username and password. also why do you have declared spring.rabbitmq.addresses=localhost:8888 – vipulk10 May 17 '18 at 15:16
  • You need to show how you are bootstrapping your application (config etc). I am using boot's connection factory and I start the broker on 8888. Just put the desired properties in application.properties and boot will auto configure it for you. – Gary Russell May 17 '18 at 15:27
  • `push.customer.arkona.controller.search` is not in your xml file. They are all `pull...`. – Gary Russell May 17 '18 at 15:31
  • but can you help me in one case .as soon as i send a message my test gets executed but my consumer which i want to test is on another thread . so how can i make my consumer listen as soon as i send a message – vipulk10 May 17 '18 at 16:44
  • It's not clear what you mean - I suggest you start a new question and show your test case. – Gary Russell May 17 '18 at 17:38