javax.jms.Message#setJMSRedelivered ( )源码实例Demo

@Override
protected Result doInvoke(Invocation invocation) throws Throwable {
	try {
		// transfer a RPC to a jms-requestor and return the call result
		QueueSession session = this.queueConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
		JmsQueueRequestor requestor = new JmsQueueRequestor(session, queue);
		Message message = session.createObjectMessage((Serializable) invocation);
		message.setJMSRedelivered(false);
		ObjectMessage result = (ObjectMessage) requestor.request(message, timeout);
		if (result == null)
			return new RpcResult(new RpcException("request is timeout in " + timeout + "ms"));
		return (Result) (result.getObject());
	} catch (JMSException e) {
		throw new RpcException(e);
	}
}
 

void send(MockJMSMessageProducer producer, Destination destination, Message message, int deliveryMode, int priority, long timeToLive, boolean disableMessageId, boolean disableTimestamp, long deliveryDelay, CompletionListener completionListener) throws JMSException {
    sendLock.lock();
    try {
        message.setJMSDeliveryMode(deliveryMode);
        message.setJMSPriority(priority);
        message.setJMSRedelivered(false);
        message.setJMSDestination(destination);

        long timeStamp = System.currentTimeMillis();
        boolean hasTTL = timeToLive > Message.DEFAULT_TIME_TO_LIVE;
        boolean hasDelay = deliveryDelay > Message.DEFAULT_DELIVERY_DELAY;

        if (!(message instanceof MockJMSMessage)) {
            throw new IllegalStateException("Mock JMS client cannot handle foreign messages");
        }

        if (!disableTimestamp) {
            message.setJMSTimestamp(timeStamp);
        } else {
            message.setJMSTimestamp(0);
        }

        if (hasTTL) {
            message.setJMSExpiration(timeStamp + timeToLive);
        } else {
            message.setJMSExpiration(0);
        }

        long messageSequence = producer.getNextMessageSequence();
        String messageId = null;
        if (!disableMessageId) {
            messageId = producer.getProducerId() + ":"+ messageSequence;
        }

        // Set the delivery time. Purposefully avoided doing this earlier so
        // that we use the 'outbound' JmsMessage object reference when
        // updating our own message instances, avoids using the interface
        // in case the JMS 1.1 Message API is actually being used due to
        // being on the classpath too.
        long deliveryTime = timeStamp;
        if (hasDelay) {
            deliveryTime = timeStamp + deliveryDelay;
        }

        message.setJMSDeliveryTime(deliveryTime);

        // Set the message ID
        message.setJMSMessageID(messageId);

        try {
            connection.onMessageSend(this, message);
        } catch (JMSException jmsEx) {
            // If the synchronous portion of the send fails the completion be
            // notified but might depending on the circumstances of the failures,
            // remove it from the queue and check if is is already completed
            // once we decide to add completion support to the mock
            throw jmsEx;
        }
    } finally {
        sendLock.unlock();
    }
}
 

private static void setJmsHeader(final MessageDescription messageDescription,
                                 final Message message)
        throws JMSException
{
    final HashMap<MessageDescription.MessageHeader, Serializable> header =
            messageDescription.getHeaders();

    if (header == null)
    {
        return;
    }

    for (Map.Entry<MessageDescription.MessageHeader, Serializable> entry : header.entrySet())
    {
        try
        {
            switch (entry.getKey())
            {
                case DESTINATION:
                    message.setJMSDestination((Destination) entry.getValue());
                    break;
                case DELIVERY_MODE:
                    message.setJMSDeliveryMode((Integer) entry.getValue());
                    break;
                case MESSAGE_ID:
                    message.setJMSMessageID((String) entry.getValue());
                    break;
                case TIMESTAMP:
                    message.setJMSTimestamp((Long) entry.getValue());
                    break;
                case CORRELATION_ID:
                    if (entry.getValue() instanceof byte[])
                    {
                        message.setJMSCorrelationIDAsBytes((byte[]) entry.getValue());
                    }
                    else
                    {
                        message.setJMSCorrelationID((String) entry.getValue());
                    }
                    break;
                case REPLY_TO:
                    throw new RuntimeException("The Test should not set the replyTo header."
                                               + " It should rather use the dedicated method");
                case REDELIVERED:
                    message.setJMSRedelivered((Boolean) entry.getValue());
                    break;
                case TYPE:
                    message.setJMSType((String) entry.getValue());
                    break;
                case EXPIRATION:
                    message.setJMSExpiration((Long) entry.getValue());
                    break;
                case PRIORITY:
                    message.setJMSPriority((Integer) entry.getValue());
                    break;
                default:
                    throw new RuntimeException(String.format("unexpected message header '%s'", entry.getKey()));
            }
        }
        catch (ClassCastException e)
        {
            throw new RuntimeException(String.format("Could not set message header '%s' to this value: %s",
                                                     entry.getKey(),
                                                     entry.getValue()), e);
        }
    }
}