singleconnectionfactory.java

来自「spring framework 2.5.4源代码」· Java 代码 · 共 505 行 · 第 1/2 页

	/**
	 * Initialize the underlying shared Connection.
	 * <p>Closes and reinitializes the Connection if an underlying
	 * Connection is present already.
	 * @throws javax.jms.JMSException if thrown by JMS API methods
	 */
	public void initConnection() throws JMSException {
		if (getTargetConnectionFactory() == null) {
			throw new IllegalStateException(
					"'targetConnectionFactory' is required for lazily initializing a Connection");
		}
		synchronized (this.connectionMonitor) {
			if (this.target != null) {
				closeConnection(this.target);
			}
			this.target = doCreateConnection();
			prepareConnection(this.target);
			if (logger.isInfoEnabled()) {
				logger.info("Established shared JMS Connection: " + this.target);
			}
			this.connection = getSharedConnectionProxy(this.target);
		}
	}

	/**
	 * Exception listener callback that renews the underlying single Connection.
	 */
	public void onException(JMSException ex) {
		resetConnection();
	}

	/**
	 * Close the underlying shared connection.
	 * The provider of this ConnectionFactory needs to care for proper shutdown.
	 * <p>As this bean implements DisposableBean, a bean factory will
	 * automatically invoke this on destruction of its cached singletons.
	 */
	public void destroy() {
		resetConnection();
	}

	/**
	 * Reset the underlying shared Connection, to be reinitialized on next access.
	 */
	public void resetConnection() {
		synchronized (this.connectionMonitor) {
			if (this.target != null) {
				closeConnection(this.target);
			}
			this.target = null;
			this.connection = null;
		}
	}

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

	/**
	 * Prepare the given Connection before it is exposed.
	 * <p>The default implementation applies ExceptionListener and client id.
	 * Can be overridden in subclasses.
	 * @param con the Connection to prepare
	 * @throws JMSException if thrown by JMS API methods
	 * @see #setExceptionListener
	 * @see #setReconnectOnException
	 */
	protected void prepareConnection(Connection con) throws JMSException {
		if (getClientId() != null) {
			con.setClientID(getClientId());
		}
		if (getExceptionListener() != null || isReconnectOnException()) {
			ExceptionListener listenerToUse = getExceptionListener();
			if (isReconnectOnException()) {
				listenerToUse = new InternalChainedExceptionListener(this, listenerToUse);
			}
			con.setExceptionListener(listenerToUse);
		}
	}

	/**
	 * Template method for obtaining a (potentially cached) Session.
	 * @param con the JMS Connection to operate on
	 * @param mode the Session acknowledgement mode
	 * (<code>Session.TRANSACTED</code> or one of the common modes)
	 * @return the Session to use, or <code>null</code> to indicate
	 * creation of a default Session
	 * @throws JMSException if thrown by the JMS API
	 */
	protected Session getSession(Connection con, Integer mode) throws JMSException {
		return null;
	}

	/**
	 * Close the given Connection.
	 * @param con the Connection to close
	 */
	protected void closeConnection(Connection con) {
		try {
			try {
				con.stop();
			}
			finally {
				con.close();
			}
		}
		catch (Throwable ex) {
			logger.warn("Could not close shared JMS Connection", ex);
		}
	}

	/**
	 * Wrap the given Connection with a proxy that delegates every method call to it
	 * but suppresses close calls. This is useful for allowing application code to
	 * handle a special framework Connection just like an ordinary Connection from a
	 * JMS ConnectionFactory.
	 * @param target the original Connection to wrap
	 * @return the wrapped Connection
	 */
	protected Connection getSharedConnectionProxy(Connection target) {
		List classes = new ArrayList(3);
		classes.add(Connection.class);
		if (target instanceof QueueConnection) {
			classes.add(QueueConnection.class);
		}
		if (target instanceof TopicConnection) {
			classes.add(TopicConnection.class);
		}
		return (Connection) Proxy.newProxyInstance(
				getClass().getClassLoader(),
				(Class[]) classes.toArray(new Class[classes.size()]),
				new SharedConnectionInvocationHandler(target));
	}


	/**
	 * Invocation handler for a cached JMS Connection proxy.
	 */
	private class SharedConnectionInvocationHandler implements InvocationHandler {

		private final Connection target;

		public SharedConnectionInvocationHandler(Connection target) {
			this.target = target;
		}

		public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
			if (method.getName().equals("equals")) {
				// Only consider equal when proxies are identical.
				return (proxy == args[0] ? Boolean.TRUE : Boolean.FALSE);
			}
			else if (method.getName().equals("hashCode")) {
				// Use hashCode of Connection proxy.
				return new Integer(System.identityHashCode(proxy));
			}
			else if (method.getName().equals("setClientID")) {
				// Handle setClientID method: throw exception if not compatible.
				String currentClientId = this.target.getClientID();
				if (currentClientId != null && currentClientId.equals(args[0])) {
					return null;
				}
				else {
					throw new javax.jms.IllegalStateException(
							"setClientID call not supported on proxy for shared Connection. " +
							"Set the 'clientId' property on the SingleConnectionFactory instead.");
				}
			}
			else if (method.getName().equals("setExceptionListener")) {
				// Handle setExceptionListener method: add to the chain.
				ExceptionListener currentExceptionListener = this.target.getExceptionListener();
				if (currentExceptionListener instanceof InternalChainedExceptionListener && args[0] != null) {
					((InternalChainedExceptionListener) currentExceptionListener).addDelegate((ExceptionListener) args[0]);
					return null;
				}
				else {
					throw new javax.jms.IllegalStateException(
							"setExceptionListener call not supported on proxy for shared Connection. " +
							"Set the 'exceptionListener' property on the SingleConnectionFactory instead. " +
							"Alternatively, activate SingleConnectionFactory's 'reconnectOnException' feature, " +
							"which will allow for registering further ExceptionListeners to the recovery chain.");
				}
			}
			else if (method.getName().equals("stop")) {
				// Handle stop method: don't pass the call on.
				return null;
			}
			else if (method.getName().equals("close")) {
				// Handle close method: don't pass the call on.
				return null;
			}
			else if (method.getName().equals("createSession") || method.getName().equals("createQueueSession") ||
					method.getName().equals("createTopicSession")) {
				boolean transacted = ((Boolean) args[0]).booleanValue();
				Integer ackMode = (Integer) args[1];
				Integer mode = (transacted ? new Integer(Session.SESSION_TRANSACTED) : ackMode);
				Session session = getSession(this.target, mode);
				if (session != null) {
					if (!method.getReturnType().isInstance(session)) {
						throw new javax.jms.IllegalStateException(
								"JMS Session does not implement specific domain: " + session);
					}
					return session;
				}
			}
			try {
				Object retVal = method.invoke(this.target, args);
				if (method.getName().equals("getExceptionListener") && retVal instanceof InternalChainedExceptionListener) {
					// Handle getExceptionListener method: hide internal chain.
					InternalChainedExceptionListener listener = (InternalChainedExceptionListener) retVal;
					return listener.getUserListener();
				}
				else {
					return retVal;
				}
			}
			catch (InvocationTargetException ex) {
				throw ex.getTargetException();
			}
		}
	}


	/**
	 * Internal chained ExceptionListener for handling the internal recovery listener
	 * in combination with a user-specified listener.
	 */
	private static class InternalChainedExceptionListener extends ChainedExceptionListener {

		private ExceptionListener userListener;

		public InternalChainedExceptionListener(ExceptionListener internalListener, ExceptionListener userListener) {
			addDelegate(internalListener);
			if (userListener != null) {
				addDelegate(userListener);
				this.userListener = userListener;
			}
		}

		public ExceptionListener getUserListener() {
			return this.userListener;
		}
	}

}