jmstemplate.java

Java/J2EE application framework based on [Expert One-on-One J2EE Design and Development] by Rod John

		return deliveryMode;
	}

	/**
	 * Set the priority of a message when sending. Since a default value may be defined
	 * administratively, it is only used when isExplicitQosEnabled equals true.
	 * @see #isExplicitQosEnabled
	 */
	public void setPriority(int priority) {
		this.priority = priority;
	}

	/**
	 * Return the priority of a message when sending.
	 */
	public int getPriority() {
		return priority;
	}

	/**
	 * Set the time-to-live of the message when sending. Since a default value may be
	 * defined administratively, it is only used when isExplicitQosEnabled equals true.
	 * @param timeToLive the message's lifetime (in milliseconds)
	 * @see #isExplicitQosEnabled
	 */
	public void setTimeToLive(long timeToLive) {
		this.timeToLive = timeToLive;
	}

	/**
	 * Return the time-to-live of the message when sending.
	 */
	public long getTimeToLive() {
		return timeToLive;
	}


	/**
	 * Make sure the connection factory has been set.
	 */
	public void afterPropertiesSet() {
		if (this.connectionFactory == null) {
			throw new IllegalArgumentException("connectionFactory is required");
		}
	}


	/**
	 * Create a JMS Connection via this template's ConnectionFactory.
	 * <p>This implementation uses JMS 1.1 API.
	 * @return the new JMS Connection
	 * @throws JMSException if thrown by JMS API methods
	 */
	protected Connection createConnection() throws JMSException {
		return getConnectionFactory().createConnection();
	}

	/**
	 * Create a JMS Session for the given Connection.
	 * <p>This implementation uses JMS 1.1 API.
	 * @param con the JMS Connection to create a Session for
	 * @return the new JMS Session
	 * @throws JMSException if thrown by JMS API methods
	 */
	protected Session createSession(Connection con) throws JMSException {
		return con.createSession(isSessionTransacted(), getSessionAcknowledgeMode());
	}

	/**
	 * Create a JMS MessageProducer for the given Session and Destination.
	 * <p>This implementation uses JMS 1.1 API.
	 * @param session the JMS Session to create a MessageProducer for
	 * @param destination the JMS Destination to create a MessageProducer for
	 * @return the new JMS MessageProducer
	 * @throws JMSException if thrown by JMS API methods
	 */
	protected MessageProducer createProducer(Session session, Destination destination) throws JMSException {
		return session.createProducer(destination);
	}

	/**
	 * Create a JMS MessageConsumer for the given Session and Destination.
	 * <p>This implementation uses JMS 1.1 API.
	 * @param session the JMS Session to create a MessageConsumer for
	 * @param destination the JMS Destination to create a MessageConsumer for
	 * @return the new JMS MessageConsumer
	 * @throws JMSException if thrown by JMS API methods
	 */
	protected MessageConsumer createConsumer(Session session, Destination destination) throws JMSException {
		return session.createConsumer(destination);
	}


	/**
	 * Resolve the given destination name into a JMS Destination,
	 * via this template's DestinationResolver.
	 * @param session the current JMS Session
	 * @param destinationName the name of the destination
	 * @return the located Destination
	 * @throws JMSException if resolution failed
	 * @see #setDestinationResolver
	 */
	protected Destination resolveDestinationName(Session session, String destinationName) throws JMSException {
		return getDestinationResolver().resolveDestinationName(session, destinationName, isPubSubDomain());
	}

	/**
	 * Convert the specified checked {@link javax.jms.JMSException JMSException} to
	 * a Spring runtime {@link org.springframework.jms.JmsException JmsException}
	 * equivalent.
	 * <p>Default implementation delegates to JmsUtils.
	 * @param ex the original checked JMSException to convert
	 * @return the Spring runtime JmsException wrapping <code>ex</code>
	 * @see org.springframework.jms.support.JmsUtils#convertJmsAccessException
	 */
	protected JmsException convertJmsAccessException(JMSException ex) {
		return JmsUtils.convertJmsAccessException(ex);
	}


	/**
	 * Execute the action specified by the given action object within a
	 * JMS Session. Generalized version of execute(SessionCallback),
	 * allowing to start the JMS Connection on the fly.
	 * <p>Use execute(SessionCallback) for the general case. Starting
	 * the JMS Connection is just necessary for receiving messages,
	 * which is preferably achieve through the <code>receive</code> methods.
	 * @param action callback object that exposes the session
	 * @return the result object from working with the session
	 * @throws JmsException if there is any problem
	 * @see #execute(SessionCallback)
	 * @see #receive
	 */
	public Object execute(SessionCallback action, boolean startConnection) throws JmsException {
		Connection con = null;
		Session session = null;
		try {
			Connection conToUse = null;
			Session sessionToUse = null;
			ConnectionHolder conHolder =
					(ConnectionHolder) TransactionSynchronizationManager.getResource(getConnectionFactory());
			if (conHolder != null) {
				conToUse = conHolder.getConnection();
				if (startConnection) {
					conToUse.start();
				}
				sessionToUse = conHolder.getSession();
			}
			else {
				con = createConnection();
				if (startConnection) {
					con.start();
				}
				session = createSession(con);
				conToUse = con;
				sessionToUse = session;
			}
			if (logger.isDebugEnabled()) {
				logger.debug("Executing callback on JMS session [" + sessionToUse + "] from connection [" + conToUse + "]");
			}
			return action.doInJms(sessionToUse);
		}
		catch (JMSException ex) {
			throw convertJmsAccessException(ex);
		}
		finally {
			JmsUtils.closeSession(session);
			JmsUtils.closeConnection(con);
		}
	}

	public Object execute(SessionCallback action) throws JmsException {
		return execute(action, false);
	}

	public Object execute(final ProducerCallback action) throws JmsException {
		return execute(new SessionCallback() {
			public Object doInJms(Session session) throws JMSException {
				MessageProducer producer = createProducer(session, null);
				return action.doInJms(session, producer);
			}
		});
	}


	public void send(MessageCreator messageCreator) throws JmsException {
		if (getDefaultDestination() == null) {
			throw new IllegalStateException("No defaultDestination specified. Check configuration of JmsTemplate.");
		}
		send(getDefaultDestination(), messageCreator);
	}

	public void send(final Destination destination, final MessageCreator messageCreator) throws JmsException {
		execute(new SessionCallback() {
			public Object doInJms(Session session) throws JMSException {
				doSend(session, destination, messageCreator);
				return null;
			}
		});
	}

	public void send(final String destinationName, final MessageCreator messageCreator) throws JmsException {
		execute(new SessionCallback() {
			public Object doInJms(Session session) throws JMSException {
				Destination destination = resolveDestinationName(session, destinationName);
				doSend(session, destination, messageCreator);
				return null;
			}
		});
	}

	protected void doSend(Session session, Destination destination, MessageCreator messageCreator)
			throws JMSException {
		MessageProducer producer = createProducer(session, destination);
		Message message = messageCreator.createMessage(session);
		if (logger.isDebugEnabled()) {
			logger.debug("Sending created message [" + message + "]");
		}
		doSend(producer, message);
		if (session.getTransacted() && !TransactionSynchronizationManager.hasResource(getConnectionFactory())) {
			// transacted session created by this template -> commit
			session.commit();
		}
	}

	protected void doSend(MessageProducer producer, Message message) throws JMSException {
		if (isExplicitQosEnabled()) {
			producer.send(message, getDeliveryMode(), getPriority(), getTimeToLive());
		}
		else {
			producer.send(message);
		}
	}


	public void convertAndSend(Object message) throws JmsException {
		if (getDefaultDestination() == null) {
			throw new IllegalStateException("No defaultDestination specified. Check configuration of JmsTemplate.");
		}
		convertAndSend(getDefaultDestination(), message);
	}

	public void convertAndSend(Destination destination, final Object message) throws JmsException {
		if (getMessageConverter() == null) {
			throw new IllegalStateException("No MessageConverter registered. Check configuration of JmsTemplate.");
		}
		send(destination, new MessageCreator() {
			public Message createMessage(Session session) throws JMSException {
				return getMessageConverter().toMessage(message, session);
			}
		});
	}

	public void convertAndSend(String destinationName, final Object message) throws JmsException {
		if (getMessageConverter() == null) {
			throw new IllegalStateException("No MessageConverter registered. Check configuration of JmsTemplate.");
		}
		send(destinationName, new MessageCreator() {
			public Message createMessage(Session session) throws JMSException {
				return getMessageConverter().toMessage(message, session);
			}
		});
	}

	public void convertAndSend(Object message, MessagePostProcessor postProcessor) throws JmsException {
		if (getDefaultDestination() == null) {
			throw new IllegalStateException("No defaultDestination specified. Check configuration of JmsTemplate.");
		}
		convertAndSend(getDefaultDestination(), message, postProcessor);
	}

	public void convertAndSend(Destination destination, final Object message,
	                           final MessagePostProcessor postProcessor) throws JmsException {
		if (getMessageConverter() == null) {
			throw new IllegalStateException("No MessageConverter registered. Check configuration of JmsTemplate.");
		}
		send(destination, new MessageCreator() {
			public Message createMessage(Session session) throws JMSException {
				Message m = getMessageConverter().toMessage(message, session);
				return postProcessor.postProcessMessage(m);
			}
		});
	}

	public void convertAndSend(String destinationName, final Object message, final MessagePostProcessor postProcessor)
	    throws JmsException {
		if (getMessageConverter() == null) {
			throw new IllegalStateException("No MessageConverter registered. Check configuration of JmsTemplate.");
		}
		send(destinationName, new MessageCreator() {
			public Message createMessage(Session session) throws JMSException {
				Message m = getMessageConverter().toMessage(message, session);
				return postProcessor.postProcessMessage(m);
			}
		});
	}


	public Message receive() throws JmsException {
		if (getDefaultDestination() == null) {
			throw new IllegalStateException("No defaultDestination specified. Check configuration of JmsTemplate.");
		}
		return receive(getDefaultDestination());
	}

	public Message receive(final Destination destination) throws JmsException {
		return (Message) execute(new SessionCallback() {
			public Object doInJms(Session session) throws JMSException {
				return doReceive(session, destination);
			}
		}, true);
	}

	public Message receive(final String destinationName) throws JmsException {
		return (Message) execute(new SessionCallback() {
			public Object doInJms(Session session) throws JMSException {
				Destination destination = resolveDestinationName(session, destinationName);
				return doReceive(session, destination);
			}
		}, true);
	}

	protected Message doReceive(Session session, Destination destination) throws JMSException {
		MessageConsumer consumer = createConsumer(session, destination);
		try {
			// use transaction timeout if available
			long timeout = getReceiveTimeout();
			ConnectionHolder conHolder =
					(ConnectionHolder) TransactionSynchronizationManager.getResource(getConnectionFactory());
			if (conHolder != null && conHolder.hasTimeout()) {
				timeout = conHolder.getTimeToLiveInMillis();
			}
			Message message = (timeout >= 0) ?
					consumer.receive(timeout) : consumer.receive();
			if (session.getTransacted()) {
				if (conHolder == null) {
					// transacted session created by this template -> commit
					session.commit();
				}
			}
			else if (message != null && isClientAcknowledge(session)) {
				message.acknowledge();
			}
			return message;
		}
		finally {
			JmsUtils.closeMessageConsumer(consumer);
		}
	}

	protected boolean isClientAcknowledge(Session session) throws JMSException {
		return (session.getAcknowledgeMode() == Session.CLIENT_ACKNOWLEDGE);
	}


	public Object receiveAndConvert() throws JmsException {
		if (getMessageConverter() == null) {
			throw new IllegalStateException("No MessageConverter registered. Check configuration of JmsTemplate.");
		}
		return doConvertFromMessage(receive());
	}

	public Object receiveAndConvert(Destination destination) throws JmsException {
		if (getMessageConverter() == null) {
			throw new IllegalStateException("No MessageConverter registered. Check configuration of JmsTemplate.");
		}
		return doConvertFromMessage(receive(destination));
	}

	public Object receiveAndConvert(String destinationName) throws JmsException {
		if (getMessageConverter() == null) {
			throw new IllegalStateException("No MessageConverter registered. Check configuration of JmsTemplate.");
		}
		return doConvertFromMessage(receive(destinationName));
	}

	protected Object doConvertFromMessage(Message message) {
		try {
			return getMessageConverter().fromMessage(message);
		}
		catch (JMSException ex) {
			throw convertJmsAccessException(ex);
		}
	}

}