Hi, There is a test case available at:
https://github.com/advanced-365/activemq-artemis-amqp2core-test Cheers, Andy -----Original Message----- From: Andy Redhead [mailto:andy.redh...@oneadvanced.com] Sent: 02 September 2016 18:02 To: users@activemq.apache.org Subject: RE: Artemis Core consumer hangs when reading message sent by AMQP producer Thanks both. I may have time over the weekend to convert the various bits I've got into a self-contained test case in a public github repo. -----Original Message----- From: John D. Ament [mailto:johndam...@apache.org] Sent: 02 September 2016 17:52 To: users@activemq.apache.org Subject: Re: Artemis Core consumer hangs when reading message sent by AMQP producer Cool, thanks. I won't be able to recreate a test today. But I have a suspicion that the message length is being lost. Will follow up. And like Justin, I would expect this to work as well. John On Fri, Sep 2, 2016 at 12:48 PM Andy Redhead <andy.redh...@oneadvanced.com> wrote: > Hmm, something is eating my attachments, both classes inline below: > > > NativeProcessMsgReader> > > > > package com...c2java; > > import java.util.concurrent.ArrayBlockingQueue; > import java.util.concurrent.BlockingQueue; > import java.util.concurrent.ThreadPoolExecutor; > import java.util.concurrent.TimeUnit; > > import javax.annotation.PostConstruct; import > javax.annotation.PreDestroy; > > import org.apache.activemq.artemis.api.core.ActiveMQException; > import org.apache.activemq.artemis.api.core.TransportConfiguration; > import org.apache.activemq.artemis.api.core.client.ActiveMQClient; > import org.apache.activemq.artemis.api.core.client.ClientConsumer; > import org.apache.activemq.artemis.api.core.client.ClientMessage; > import org.apache.activemq.artemis.api.core.client.ClientSession; > import > org.apache.activemq.artemis.api.core.client.ClientSessionFactory; > import org.apache.activemq.artemis.api.core.client.ServerLocator; > import > org.apache.activemq.artemis.core.remoting.impl.invm.InVMConnectorFacto > ry; > import org.slf4j.Logger; > import org.slf4j.LoggerFactory; > > import com...NativeProcessCmd; > > /** > * Read messages from Artemis using server API. > * > * Uses a single thread (with a persistent ClientSession) to read from > the shared queue used by all > * native processes to send outbound messages. > * > * Messages are handed on to a fixed sized executor for processing and > sending results to browser. > * > * @author andyredhead > * > */ > public class NativeProcessMsgReader extends Thread { > > final static Logger logger = > LoggerFactory.getLogger(NativeProcessMsgReader.class); > > int workerThreadPoolSize = 1; > int executorQueueLength = 500; > > String incomingMsgFromNativeProcQueueName; > > BlockingQueue<Runnable> nativeProcCmdQueue; > ThreadPoolExecutor nativeProcCmdExec; > > ClientSessionFactory factory; > ClientSession session; > ClientConsumer consumer; > > boolean shutdown = false; > > public NativeProcessMsgReader(int workerThreadPoolSize, int > executorQueueLength, > String incomingMsgFromNativeProcQueueName) { > this.workerThreadPoolSize = workerThreadPoolSize; > this.incomingMsgFromNativeProcQueueName = > incomingMsgFromNativeProcQueueName; > } > > @PostConstruct > public void startUp() throws Exception { > logger.info("startUp - start, thread pool size: {}, queue length: > {}, queue name: {}", > workerThreadPoolSize, workerThreadPoolSize, > incomingMsgFromNativeProcQueueName); > > try { > > // processing work taken from queue > nativeProcCmdQueue = new > ArrayBlockingQueue<Runnable>(executorQueueLength); > nativeProcCmdExec = new ThreadPoolExecutor(workerThreadPoolSize, > workerThreadPoolSize, 5, > TimeUnit.MINUTES, nativeProcCmdQueue); > > > > this.start(); > > } catch (Exception asyncQueueSetupErr) { > logger.error("ctor - problem setting up queue consumer: " + > asyncQueueSetupErr.getMessage(), > asyncQueueSetupErr); > throw asyncQueueSetupErr; > } > > logger.info("startUp - complete."); > } > > @PreDestroy > public void shutDown() { > > logger.info("shutDown - commencing shut down"); > > // stop new work coming in > shutdown = true; > > // close down the native prod cmd executor > logger.info("shutDown - requesting executor shuts down..."); > nativeProcCmdExec.shutdownNow(); > logger.info("shutDown - requested executor shuts down."); > } > > @Override > public void run() { > > // reading from queue > ServerLocator locator = ActiveMQClient.createServerLocatorWithoutHA( > new > TransportConfiguration(InVMConnectorFactory.class.getName())); > > try { > factory = locator.createSessionFactory(); > session = factory.createSession(); > consumer = > session.createConsumer(incomingMsgFromNativeProcQueueName); > session.start(); > } catch (Exception e) { > logger.warn("run - problem setting up native proc to java queue > consumer: {}", e.getMessage(), > e); > } > > while (!shutdown) { > try { > > logger.debug("run - in while loop, waiting for message"); > > /* break every half second to check shutdown status */ > ClientMessage msgReceived = consumer.receive(500); > > if (msgReceived != null) { > > logger.debug("run - received non-null message"); > logger.debug("run - non-null message, body length: {}", > msgReceived.getBodySize()); > > String receivedMsg = > msgReceived.getBodyBuffer().readString(); > > logger.debug("run - message text: {}", receivedMsg); > > if ((receivedMsg != null) && (receivedMsg.length() > 0)) { > > logger.debug("run - in while loop, got message"); > > NativeProcessCmd nativeProcCmd = new > NativeProcessCmd(receivedMsg); > nativeProcCmdExec.submit(nativeProcCmd); > logger.debug("run - in while loop, cmd pushed to > executor"); > > } else { > logger.debug("run - null or zero length message read from > native process cmd queue"); > } > } else { > logger.debug("run - no message during last read period"); > } > > } catch (ActiveMQException e) { > logger.warn( > "run - problem reading message from native process command > queue: " + e.getMessage(), > e); > } > > } // end not shutdown while loop > > > // been told to shut down so close Artemis client objects > > try { > > consumer.close(); > session.stop(); > factory.close(); > > logger.info("run - Artemis client objects shut down."); > > } catch (Exception e) { > logger.warn("run - problem closing Artemis client objects during > shutdown: " + e.getMessage(), > e); > } > > } > > } > > > NumberGuessMain> > > package com...numberguess; > > import java.io.IOException; > import java.util.HashMap; > > import org.apache.qpid.proton.amqp.messaging.AmqpValue; > import org.apache.qpid.proton.message.Message; > import org.apache.qpid.proton.messenger.Messenger; > > import com.google.gson.Gson; > > public class NumberGuessMain { > > private int answer = -1; > > public static final int EXIT_CODE_WRONG_CMD_LINE_ARGS = 1000; > > public static final int EXIT_CODE_BROWSER_CONTROL_QUEUE_SETUP_ERR = > 1001; > > public static final int EXIT_CODE_BROWSER_RESPONSE_QUEUE_SETUP_ERR = > 1002; > > private String taskInstanceId; > > private String browserControlQueueUrl; > > private String browserResponseQueueUrl; > > Messenger sendBrowserControlMessenger; > > Messenger receiveBrowserResponseMessenger; > > Gson gson = new Gson(); > > > public static void main(String[] args) { > if ((args != null) && (args.length > 2)) { > NumberGuessMain me = new NumberGuessMain(args[0], args[1], args[2]); > me.run(); > me.shutdown(); > } else { > > System.err.println("task instance id, browser control and > response queues not specified" > + " (expects taskInstanceId browserControlQueueAmqpUrl > browserResponseQueueAmqpUrl)"); > System.exit(EXIT_CODE_WRONG_CMD_LINE_ARGS); > } > > > > } > > public NumberGuessMain(String taskInstanceId, String > browserControlQueueUrl, > String browserResponseQueueUrl) { > > this.taskInstanceId = taskInstanceId; > this.browserControlQueueUrl = browserControlQueueUrl; > this.browserResponseQueueUrl = browserResponseQueueUrl; > > System.out.println("Task Instance ID: " + this.taskInstanceId); > > // browser control queue > > System.out.println("Browser Control Queue: " + > this.browserControlQueueUrl); > > sendBrowserControlMessenger = new Messenger.Factory().create(); > try { > sendBrowserControlMessenger.start(); > } catch (IOException e) { > System.err.println("Problem starting browser control AMQP messenger: > " + e.getMessage()); > e.printStackTrace(); > System.exit(EXIT_CODE_BROWSER_CONTROL_QUEUE_SETUP_ERR); > } > > > // browser response queue > > System.out.println("Browser Response Queue: " + > this.browserResponseQueueUrl); > > receiveBrowserResponseMessenger = new Messenger.Factory().create(); > try { > > receiveBrowserResponseMessenger.start(); > > receiveBrowserResponseMessenger.subscribe(browserResponseQueueUrl); > > } catch (IOException e) { > System.err.println("Problem starting receive from browser AMQP > messenger: " + e.getMessage()); > e.printStackTrace(); > System.exit(EXIT_CODE_BROWSER_RESPONSE_QUEUE_SETUP_ERR); > } > > } > > protected void run() { > > System.out.println("Hello from numberguess"); > > OutboundNativeProcessMessage showFirstPanelMsg = > new OutboundNativeProcessMessage(taskInstanceId, > "ShowTitlePanel", null); > > send(showFirstPanelMsg); > } > > protected void send(OutboundNativeProcessMessage outMsg) { > Message msg = new Message.Factory().create(); > msg.setAddress(browserControlQueueUrl); > String msgJson = gson.toJson(outMsg); > msg.setBody(new AmqpValue(msgJson)); > sendBrowserControlMessenger.put(msg); > sendBrowserControlMessenger.send(); > System.out.println("Sent browser control msg: " + msgJson); > } > > protected HashMap sendReceive(OutboundNativeProcessMessage outMsg) { > HashMap receivedData = null; > > send(outMsg); > > System.out.println("Waiting for response message..."); > receiveBrowserResponseMessenger.recv(); > Message msg = receiveBrowserResponseMessenger.get(); > System.out.println("Got response message."); > > // @ToDo process the message here... > > return receivedData; > } > > protected void shutdown() { > if (sendBrowserControlMessenger != null) { > sendBrowserControlMessenger.stop(); > } > if (receiveBrowserResponseMessenger != null) { > try { > // receiveBrowserResponseMessenger. > receiveBrowserResponseMessenger.stop(); > } catch (Exception e) { > System.err.println("Problem closing browser response queue > messenger: " + e.getMessage()); > e.printStackTrace(); > } > } > } > > } > > -----Original Message----- > From: Andy Redhead [mailto:andy.redh...@oneadvanced.com] > Sent: 02 September 2016 17:45 > To: users@activemq.apache.org > Subject: RE: Artemis Core consumer hangs when reading message sent by > AMQP producer > > Hi, > > Sorry, I could have sworn I attached the code in my original email - > trying again... > > Cheers, Andy > > -----Original Message----- > From: John D. Ament [mailto:johndam...@apache.org] > Sent: 02 September 2016 17:42 > To: users@activemq.apache.org > Subject: Re: Artemis Core consumer hangs when reading message sent by > AMQP producer > > Andy, > > Are you able to provide your client code? NativeProcessMsgReader > isn't Artemis's so its not clear what else you're doing to connect, > subscribe, etc. > > John > > On Fri, Sep 2, 2016 at 12:38 PM Andy Redhead > <andy.redh...@oneadvanced.com > > > wrote: > > > Hi, > > > > > > > > Adding to the information below, if I change the junit test case > > that tries sending a message through artemis into “NativeProcessMsgReader” > > to use AMQP (rather than the core api), I get the stack trace: > > > > > > > > Exception in thread "Thread-1" *java.lang.IndexOutOfBoundsException*: > > readerIndex(25) + length(276496418) exceeds writerIndex(294): > > DuplicatedByteBuf(ridx: 25, widx: 294, cap: 512, unwrapped: > > UnpooledHeapByteBuf(ridx: 17, widx: 294, cap: 512)) > > > > at io.netty.buffer.AbstractByteBuf.checkReadableBytes( > > *AbstractByteBuf.java:1165*) > > > > at io.netty.buffer.AbstractByteBuf.readBytes( > > *AbstractByteBuf.java:675*) > > > > at io.netty.buffer.AbstractByteBuf.readBytes( > > *AbstractByteBuf.java:683*) > > > > at io.netty.buffer.WrappedByteBuf.readBytes( > > *WrappedByteBuf.java:511*) > > > > at > > org.apache.activemq.artemis.core.buffers.impl.ChannelBufferWrapper.r > > ea > > dSimpleStringInternal( > > *ChannelBufferWrapper.java:90*) > > > > at > > org.apache.activemq.artemis.core.buffers.impl.ChannelBufferWrapper.r > > ea > > dStringInternal( > > *ChannelBufferWrapper.java:113*) > > > > at > > org.apache.activemq.artemis.core.buffers.impl.ChannelBufferWrapper.r > > ea > > dString( > > *ChannelBufferWrapper.java:96*) > > > > at > > com.advanced365.transoft.renault.poc.decforms.asyncmsg.c2java.Native > > Pr > > ocessMsgReader.run( > > *NativeProcessMsgReader.java:132*) > > > > > > > > It is interesting to note that “NativeProcessMsgReader.java:132” > > links to this line: > > > > > > > > String receivedMsg = > > msgReceived.getBodyBuffer().readString(); > > > > > > > > Which is the line I thought was causing trouble when the code ran in > > Tomcat… > > > > > > > > > > > > Next, I modified “NativeProcessMsgReader” to use Qpid AMQP client to > > read messages from Artemis (with the test case still sending > > messages using > > AMQP) – this worked perfectly. > > > > > > > > So it seems that: > > > > · if the producer and consumer use the same protocol, messages flow > > through ok. > > > > · If the producer users AMQP API and the consumer uses Artemis Core > > API then there is a problem > > > > > > > > For now I’ll stick with using AMQP at both ends. > > > > > > > > I’m still curious to know if it’s reasonable to expect using an AMQP > > producer and an Artemis Core consumer to work? > > > > > > > > Cheers, Andy > > > > > > > > *From:* Andy Redhead [mailto:andy.redh...@oneadvanced.com] > > *Sent:* 01 September 2016 23:03 > > *To:* users@activemq.apache.org > > *Subject:* Artemis Core consumer hangs when reading message sent by > > AMQP producer > > > > > > > > Hi, > > > > > > > > I’m running Artemis 1.3.0 embedded inside a spring app running in Tomcat. > > > > > > > > I have a remote message producer (NumberGuessMain.java) that uses > > the Apache Qpid Proton library to push messages onto an Artemis > > queue with the > > URL: > > > > > > > > amqp://localhost:5672/native-2-java > > > > > > > > The messages created by the producer are JSON strings. > > > > > > > > I have a single threaded, singleton consumer that uses the native > > Artemis API (NativeProcessMsgReader.java) running inside the same > > web app as Artemis, reading from the queue: > > > > > > > > native-2-java > > > > > > > > While there are no messages to read, the consumer happily loops > > through the while loop in the “run” method. > > > > > > > > As soon as the consumer tries to read the body of the first message, > > it hangs on the line: > > > > > > > > String receivedMsg = > > msgReceived.getBodyBuffer().readString(); > > > > > > > > The last lines in the log file are: > > > > > > > > 2016-09-01T22:27:26,270 15137 [Thread-6] DEBUG > > c.a.t.r.p.d.a.c.NativeProcessMsgReader - run - no message during > > last read period > > > > 2016-09-01T22:27:26,270 15137 [Thread-6] DEBUG > > c.a.t.r.p.d.a.c.NativeProcessMsgReader - run - in while loop, > > waiting for message > > > > 2016-09-01T22:27:26,772 15639 [Thread-6] DEBUG > > c.a.t.r.p.d.a.c.NativeProcessMsgReader - run - no message during > > last read period > > > > 2016-09-01T22:27:26,772 15639 [Thread-6] DEBUG > > c.a.t.r.p.d.a.c.NativeProcessMsgReader - run - in while loop, > > waiting for message > > > > 2016-09-01T22:27:26,912 15779 [Thread-6] DEBUG > > c.a.t.r.p.d.a.c.NativeProcessMsgReader - run - received non-null > > message > > > > 2016-09-01T22:27:26,912 15779 [Thread-6] DEBUG > > c.a.t.r.p.d.a.c.NativeProcessMsgReader - run - non-null message, > > body > > length: 307 > > > > > > > > To me this looks like the message has reached Artemis and is made > > available to the consumer but something goes wrong when the consumer > > tries to access the body of the message. > > > > > > > > I’m new to Artemis and AMQP so it’s quite possible I’m doing > > something stupid… > > > > > > > > Is the basic assumption that it’s ok to send a message using AMQP > > and receive it using Artemis core API valid? > > > > > > > > Is there something obvious in the code that is causing this problem? > > > > > > > > Any pointers gratefully received. > > > > > > > > Cheers, Andy > > > > > > > > [image: cid:image012.png@01D17AF7.D972DF70] > > <http://www.oneadvanced.com/> > > > > *Andy Redhead* > > Principal Consultant > Solutions > Advanced > > *________________________* > > > > *Advanced* > > 230 City Road, London, EC1V 2TT > > t: 020 7880 8888 > m: 0781 392 5246 > > > > www.oneadvanced.com > > > > [image: cid:image018.png@01D17AF7.D972DF70] > > <http://www.linkedin.com/company/2426258>[image: > > cid:image019.png@01D17AF7.D972DF70] > > <https://twitter.com/Going_Advanced> > > > > *>* > > * A Sunday Times Top Track 250 Company 2015 **>** Ranked in UK's 50 > > fastest growing technology companies 2014* > > > > > > > > ***** Email confidentiality ***** > > > > This message is private and confidential. If you have received this > > message in error, please notify us and remove it from your system. > > The dissemination, copying or distribution of this message, or > > related files, by anyone other than the intended recipient is > > strictly > prohibited. > > > > > > > > Any views or opinions expressed are solely those of the author and > > do not necessarily represent those of Advanced 365 Limited. > > > > > > > > ***** Email monitoring ***** > > > > Advanced 365 Limited may monitor email traffic data and also the > > content of email for the purposes of security and staff training. > > > > > > > > ***** Email security ***** > > > > In keeping with good computing practice, the recipient of this email > > should ensure that it is virus-free. Advanced 365 Limited does not > > accept responsibility for any virus that may be transferred by way > > of > this email. > > > > > > > > Email may be susceptible to data corruption, interception and/or > > unauthorised amendment. Advanced 365 Limited does not accept > > liability for any such corruption, interception or amendment or any > > consequences > thereof. > > > > > > > > This email has been scanned for viruses by the Symantec Email > > Security.cloud service. > > > > > > > > Advanced 365 Limited, part of the Advanced Computer Software Group > > > > Registered office: Ditton Park, Riding Court Road, Datchet, > > Berkshire, > > SL3 9LL, UK > > > > Registered in England under number 2124540 > > > > > > ------------------------------ > > > > > > Please consider the environment: Think before you print! > > > > ***** Email confidentiality ***** > > > > This message is private and confidential. If you have received this > > message in error, please notify us and remove it from your system. > > The dissemination, copying or distribution of this message, or > > related files, by anyone other than the intended recipient is > > strictly > prohibited. > > > > > > > > Any views or opinions expressed are solely those of the author and > > do not necessarily represent those of Advanced 365 Limited. > > > > > > > > ***** Email monitoring ***** > > > > Advanced 365 Limited may monitor email traffic data and also the > > content of email for the purposes of security and staff training. > > > > > > > > ***** Email security ***** > > > > In keeping with good computing practice, the recipient of this email > > should ensure that it is virus-free. Advanced 365 Limited does not > > accept responsibility for any virus that may be transferred by way > > of > this email. > > > > > > > > Email may be susceptible to data corruption, interception and/or > > unauthorised amendment. Advanced 365 Limited does not accept > > liability for any such corruption, interception or amendment or any > > consequences > thereof. > > > > > > > > This email has been scanned for viruses by the Symantec Email > > Security.cloud service. > > > > > > > > Advanced 365 Limited, part of the Advanced Computer Software Group > > > > Registered office: Ditton Park, Riding Court Road, Datchet, > > Berkshire, > > SL3 9LL, UK > > > > Registered in England under number 2124540 > > > ***** Email confidentiality ***** > > This message is private and confidential. If you have received this > message in error, please notify us and remove it from your system. The > dissemination, copying or distribution of this message, or related > files, by anyone other than the intended recipient is strictly prohibited. > > > > Any views or opinions expressed are solely those of the author and do > not necessarily represent those of Advanced 365 Limited. > > > > ***** Email monitoring ***** > > Advanced 365 Limited may monitor email traffic data and also the > content of email for the purposes of security and staff training. > > > > ***** Email security ***** > > In keeping with good computing practice, the recipient of this email > should ensure that it is virus-free. Advanced 365 Limited does not > accept responsibility for any virus that may be transferred by way of this > email. > > > > Email may be susceptible to data corruption, interception and/or > unauthorised amendment. Advanced 365 Limited does not accept liability > for any such corruption, interception or amendment or any consequences > thereof. > > > > This email has been scanned for viruses by the Symantec Email > Security.cloud service. > > > > Advanced 365 Limited, part of the Advanced Computer Software Group > > Registered office: Ditton Park, Riding Court Road, Datchet, Berkshire, > SL3 9LL, UK > > Registered in England under number 2124540 > ***** Email confidentiality ***** > > This message is private and confidential. If you have received this > message in error, please notify us and remove it from your system. The > dissemination, copying or distribution of this message, or related > files, by anyone other than the intended recipient is strictly prohibited. > > > > Any views or opinions expressed are solely those of the author and do > not necessarily represent those of Advanced 365 Limited. > > > > ***** Email monitoring ***** > > Advanced 365 Limited may monitor email traffic data and also the > content of email for the purposes of security and staff training. > > > > ***** Email security ***** > > In keeping with good computing practice, the recipient of this email > should ensure that it is virus-free. Advanced 365 Limited does not > accept responsibility for any virus that may be transferred by way of this > email. > > > > Email may be susceptible to data corruption, interception and/or > unauthorised amendment. Advanced 365 Limited does not accept liability > for any such corruption, interception or amendment or any consequences > thereof. > > > > This email has been scanned for viruses by the Symantec Email > Security.cloud service. > > > > Advanced 365 Limited, part of the Advanced Computer Software Group > > Registered office: Ditton Park, Riding Court Road, Datchet, Berkshire, > SL3 9LL, UK > > Registered in England under number 2124540 > ***** Email confidentiality ***** This message is private and confidential. If you have received this message in error, please notify us and remove it from your system. The dissemination, copying or distribution of this message, or related files, by anyone other than the intended recipient is strictly prohibited. Any views or opinions expressed are solely those of the author and do not necessarily represent those of Advanced 365 Limited. ***** Email monitoring ***** Advanced 365 Limited may monitor email traffic data and also the content of email for the purposes of security and staff training. ***** Email security ***** In keeping with good computing practice, the recipient of this email should ensure that it is virus-free. Advanced 365 Limited does not accept responsibility for any virus that may be transferred by way of this email. Email may be susceptible to data corruption, interception and/or unauthorised amendment. Advanced 365 Limited does not accept liability for any such corruption, interception or amendment or any consequences thereof. This email has been scanned for viruses by the Symantec Email Security.cloud service. Advanced 365 Limited, part of the Advanced Computer Software Group Registered office: Ditton Park, Riding Court Road, Datchet, Berkshire, SL3 9LL, UK Registered in England under number 2124540 ***** Email confidentiality ***** This message is private and confidential. If you have received this message in error, please notify us and remove it from your system. The dissemination, copying or distribution of this message, or related files, by anyone other than the intended recipient is strictly prohibited. Any views or opinions expressed are solely those of the author and do not necessarily represent those of Advanced 365 Limited. ***** Email monitoring ***** Advanced 365 Limited may monitor email traffic data and also the content of email for the purposes of security and staff training. ***** Email security ***** In keeping with good computing practice, the recipient of this email should ensure that it is virus-free. Advanced 365 Limited does not accept responsibility for any virus that may be transferred by way of this email. Email may be susceptible to data corruption, interception and/or unauthorised amendment. Advanced 365 Limited does not accept liability for any such corruption, interception or amendment or any consequences thereof. This email has been scanned for viruses by the Symantec Email Security.cloud service. Advanced 365 Limited, part of the Advanced Computer Software Group Registered office: Ditton Park, Riding Court Road, Datchet, Berkshire, SL3 9LL, UK Registered in England under number 2124540
