title: ActiveMQ and embedded broker
body: <p>EDIT: Rephrased the question:</p>
  
  <p>I want to use ActiveMQ as a messenger service between my server and client applications.  </p>
  
  <p>I am trying to set up an embedded broker (i.e. not a separate process) within the server to handle the produced messages for my clients to consume.  This queue is persisted.</p>
  
  <p>The broker initialisation as follows:</p>
  
  <pre><code>BrokerService broker = new BrokerService();
  KahaPersistenceAdapter adaptor = new KahaPersistenceAdapter();
  adaptor.setDirectory(new File("activemq"));
  broker.setPersistenceAdapter(adaptor);
  broker.setUseJmx(true);
  broker.addConnector("tcp://localhost:61616");
  broker.start();
  </code></pre>
  
  <p>After tinkering, I ended up with the server part being:</p>
  
  <pre><code>public static class HelloWorldProducer implements Runnable {
      public void run() {
          try {
              ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost"); // apparently the vm part is all i need
              Connection connection = connectionFactory.createConnection(); 
              connection.start();
              Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
              Destination destination = session.createQueue("TEST.FOO");
              MessageProducer producer = session.createProducer(destination);
              producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
              String text = "Hello world! From: " + Thread.currentThread().getName() + " : " + this.hashCode();
              TextMessage message = session.createTextMessage(text);
              System.out.println("Sent message: "+ message.hashCode() + " : " + Thread.currentThread().getName());
              producer.send(message);
              session.close();
              connection.close();
          }
          catch (Exception e) {
              System.out.println("Caught: " + e);
              e.printStackTrace();
          }
      }
  }
  </code></pre>
  
  <p>The client is very similar and looks like this:</p>
  
  <pre><code>public static class HelloWorldConsumer implements Runnable, ExceptionListener {
      public void run() {
          try {
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost");
              Connection connection = connectionFactory.createConnection(); // exception happens here...
              connection.start();
              connection.setExceptionListener(this);
              Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
              Destination destination = session.createQueue("TEST.FOO");
              MessageConsumer consumer = session.createConsumer(destination);
              Message message = consumer.receive(1000);
              if (message instanceof TextMessage) {
                  TextMessage textMessage = (TextMessage) message;
                  String text = textMessage.getText();
                  System.out.println("*****Received: " + text);
              } else {
                  System.out.println("*****Received obj: " + message);
              }
              consumer.close();
              session.close();
              connection.close();
          } catch (Exception e) {
              System.out.println("Caught: " + e);
              e.printStackTrace();
          }
      }
  </code></pre>
  
  <p>The main method simply starts each of these in a thread to start producing/receiving messages.</p>
  
  <p>...but I am running into the following with the start of each thread:</p>
  
  <pre><code>2013-01-24 07:54:31,271 INFO  [org.apache.activemq.broker.BrokerService] Using Persistence Adapter: AMQPersistenceAdapter(activemq-data/localhost)
  2013-01-24 07:54:31,281 INFO  [org.apache.activemq.store.amq.AMQPersistenceAdapter] AMQStore starting using directory: activemq-data/localhost
  2013-01-24 07:54:31,302 INFO  [org.apache.activemq.kaha.impl.KahaStore] Kaha Store using data directory activemq-data/localhost/kr-store/state
  2013-01-24 07:54:31,339 INFO  [org.apache.activemq.store.amq.AMQPersistenceAdapter] Active data files: []
  2013-01-24 07:54:31,445 DEBUG [org.apache.activemq.broker.jmx.ManagementContext] Probably not using JRE 1.4: mx4j.tools.naming.NamingService
  2013-01-24 07:54:31,450 DEBUG [org.apache.activemq.broker.jmx.ManagementContext] Failed to create local registry
      java.rmi.server.ExportException: internal error: ObjID already in use
      at sun.rmi.transport.ObjectTable.putTarget(ObjectTable.java:186)
      at sun.rmi.transport.Transport.exportObject(Transport.java:92)
      at sun.rmi.transport.tcp.TCPTransport.exportObject(TCPTransport.java:247)
      at sun.rmi.transport.tcp.TCPEndpoint.exportObject(TCPEndpoint.java:411)
      at sun.rmi.transport.LiveRef.exportObject(LiveRef.java:147)
          &lt;snip....&gt;
  </code></pre>
  
  <p>It seems like the messages are produced and consumed successfully (the other issues I previously posted about was resolved), but the above exception is worrying me.  </p>
  
  <p>EDIT: During broker shutdown, I am now also greeted by the following:</p>
  
  <pre><code>2013-01-25 08:40:17,486 DEBUG [org.apache.activemq.transport.failover.FailoverTransport] Transport failed with the following exception:
      java.io.EOFException
      at java.io.DataInputStream.readInt(DataInputStream.java:392)
      at org.apache.activemq.openwire.OpenWireFormat.unmarshal(OpenWireFormat.java:269)
      at org.apache.activemq.transport.tcp.TcpTransport.readCommand(TcpTransport.java:210)
      at org.apache.activemq.transport.tcp.TcpTransport.doRun(TcpTransport.java:202)
      at org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:185)
      at java.lang.Thread.run(Thread.java:722)
  </code></pre>