qpid-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Gary Ogden <gog...@gmail.com>
Subject Re: persistent messages
Date Tue, 15 May 2012 20:05:45 GMT
So I did some more reading and I now understand why the messages weren't
being retained. So I need to create a queue, and have that queue created in
the virtualhosts file so that the broker starts it up. But even after doing
that, I still have the issue where the messages are getting dropped until
the consumer starts.

Here's my VH file:

   <virtualhost>
   <name>blah</name>
   <blah>
   <queue>
      <name>blahqueue</name>
      <blahqueue>
        <exchange>amq.direct</exchange>
        <maximumQueueDepth>4235264</maximumQueueDepth>  <!-- 4Mb -->
        <maximumMessageSize>2117632</maximumMessageSize> <!-- 2Mb -->
        <maximumMessageAge>600000</maximumMessageAge>  <!-- 10 mins -->
      </blahqueue>
   </queue>
   </blah>
  </virtualhost>

Here's my properties file:

java.naming.factory.initial =
org.apache.qpid.jndi.PropertiesFileInitialContextFactory
connectionfactory.qpidConnectionfactory = amqp://guest:guest@clientid
/blah?brokerlist='tcp://localhost:5672'
destination.queueExchange = amq.direct

Here's a simple method to produce:
private void startProducing() {
        Context context = null;
        Connection connection = null;
        try {
            Properties properties = new Properties();

properties.load(this.getClass().getResourceAsStream("blah.properties"));
            context = new InitialContext(properties);

            ConnectionFactory connectionFactory = (ConnectionFactory)
context.lookup("qpidConnectionfactory");
            connection = connectionFactory.createConnection();
            connection.start();

            Session session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
            Destination destination = (AMQAnyDestination)
context.lookup("queueExchange");

            MessageProducer messageProducer =
session.createProducer(destination);
            int count = 0;
                while (true) {
                    DataInputStream dis = new DataInputStream(new
FileInputStream("test.xml"));
                    try {
                        File testFile = new File("test.xml");
                        long len = new File("test.xml").length();
                        System.out.println("test.xml length = " + len);
                        byte[] bytes = new byte[(int) len];
                        dis.readFully(bytes);
                        for (int i = 0; i <= 100; i++) {
                            count++;
                            System.out.println(" [x] Sending test.xml. Time
= " + System.currentTimeMillis());
                            BytesMessage message =
session.createBytesMessage();
                            message.writeBytes(bytes);
                            messageProducer.send(message);
                            System.out.println(" [x] Sent test.xml + " +
count + ". Time = " + System.currentTimeMillis());
                        }
                    } finally {
                        dis.close();
                    }
                    Thread.sleep(4000);
                }
            } catch (Exception e) {
                System.out.println("Exception caught: " + e.getMessage() +
e.getStackTrace().toString());
                try {
                    if (context != null) {
                        context.close();
                    }
                    if (connection != null) {
                        connection.close();
                    }
                } catch (Exception ex) {
                    System.out.println("Exception caught closing
connection: " + e.getMessage());
                }

            }
        }

What am I missing? It still waits for the consumer and no messages get
queued.

Mime
  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message