[
https://issues.apache.org/jira/browse/QPID-1642?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=12676300#action_12676300
]
Arnaud Simon commented on QPID-1642:
------------------------------------
no, the problem is that we need to know whether messages have been sent to the
client after messageFlow returns.
I believe that a way to achieve that is to do something like
_0_10session.getQpidSession().messageFlow(getConsumerTagString(),
MessageCreditUnit.MESSAGE, 1);
_0_10session.getQpidSession().messageFlush();
_0_10session.getQpidSession().sync();
According to messageFlush definition:
"Forces the sender to exhaust his credit supply. The sender's credit will
always be zero when this command completes.
The command completes when immediately available message data has been
transferred, or when the credit supply
is exhausted."
Once the sync ope returns potentially available messages will have been
transfered and ready to consume form our local queue synchronousQueue
> JMS ReceiveNowait does not return a message even if the queue is not empty
> --------------------------------------------------------------------------
>
> Key: QPID-1642
> URL: https://issues.apache.org/jira/browse/QPID-1642
> Project: Qpid
> Issue Type: Bug
> Components: Java Client
> Affects Versions: M4
> Reporter: Arnaud Simon
> Assignee: Rafael H. Schloming
> Fix For: M5
>
>
> Description:
> The first invocation of receiveNoWait does not return a message even if the
> queue is full.
> How to replicate:
> 1) create a queue
> 2) send a message
> 3) try to consume this message using receiveNoWait
> ==> we get null even if there is a message seating in the queue
> Solution:
> When a JMS consumer first invokes receiveNowait then
> BasicMessageConsumer_0_10 request for more credits and then poll the
> _synchronousQueue. This does not leave enough time for the message to be
> enqueued.
> One solution would be to query the queue size before allowing the credits:
> Something like that would work:
> public Object getMessageFromQueue(long l) throws InterruptedException
> {
> long size = 0;
> if(l < 0)
> {
> size = _0_10session.requestQueueDepth(getDestination()) ;
> }
> if (isStrated() && ! getSession().prefetch() &&
> _synchronousQueue.isEmpty())
> {
> _0_10session.getQpidSession().messageFlow(getConsumerTagString(),
>
> MessageCreditUnit.MESSAGE, 1);
> }
> if (! getSession().prefetch())
> {
> _syncReceive.set(true);
> }
> Object o;// = super.getMessageFromQueue(l);
> if (l > 0)
> {
> o = _synchronousQueue.poll(l, TimeUnit.MILLISECONDS);
> }
> else if (l < 0)
> {
> if(size <= 0 )
> {
> o = _synchronousQueue.poll();
> }
> else
> {
> o = _synchronousQueue.take();
> }
> }
> else
> {
> o = _synchronousQueue.take();
> }
> if (! getSession().prefetch())
> {
> _syncReceive.set(false);
> }
> return o;
> }
--
This message is automatically generated by JIRA.
-
You can reply to this email to add a comment to the issue online.
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:[email protected]
