| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| package flex.messaging.services.messaging.adapters; |
| |
| import java.util.Collection; |
| import java.util.Enumeration; |
| import java.util.HashMap; |
| import java.util.Hashtable; |
| import java.util.Iterator; |
| import java.util.LinkedList; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.concurrent.ConcurrentHashMap; |
| |
| import javax.jms.JMSException; |
| import javax.naming.Context; |
| |
| import flex.management.runtime.messaging.services.messaging.adapters.JMSAdapterControl; |
| import flex.messaging.Destination; |
| import flex.messaging.MessageClient; |
| import flex.messaging.MessageClientListener; |
| import flex.messaging.MessageDestination; |
| import flex.messaging.MessageException; |
| import flex.messaging.config.ConfigMap; |
| import flex.messaging.config.ConfigurationException; |
| import flex.messaging.log.Log; |
| import flex.messaging.log.LogCategories; |
| import flex.messaging.messages.CommandMessage; |
| import flex.messaging.messages.ErrorMessage; |
| import flex.messaging.messages.Message; |
| import flex.messaging.messages.MessagePerformanceInfo; |
| import flex.messaging.messages.MessagePerformanceUtils; |
| import flex.messaging.services.MessageService; |
| import flex.messaging.services.messaging.adapters.JMSSettings.DeliverySettings; |
| |
| /** |
| * This adapter for the MessageService integrates Flex messaging |
| * with Java Message Service destinations. |
| */ |
| public class JMSAdapter extends MessagingAdapter implements JMSConfigConstants, JMSExceptionListener, JMSMessageListener, MessageClientListener { |
| public static final String LOG_CATEGORY = LogCategories.SERVICE_MESSAGE_JMS; |
| private static final String DURABLE_SUBSCRIBER_NAME_PREFIX = "FlexClient_"; |
| |
| // Note that clientId is kept track as Object (instead of String) in all of |
| // these data structures because in clustering, clientId is not a String, |
| // it's an instance of org.jgroups.stack.IpAddress instead. |
| private Map<JMSConsumer, Object> consumerToClientId; |
| private Map<Object, MessageClient> messageClients; |
| private LinkedList<JMSProducer> topicProducers; |
| private Map<Object, JMSConsumer> topicConsumers; |
| private LinkedList<JMSProducer> queueProducers; |
| private Map<Object, JMSConsumer> queueConsumers; |
| |
| // JMSAdapter properties |
| private JMSSettings settings; |
| private JMSAdapterControl controller; |
| |
| //-------------------------------------------------------------------------- |
| // |
| // Constructor |
| // |
| //-------------------------------------------------------------------------- |
| |
| /** |
| * Constructs an unmanaged <code>JMSAdapter</code> instance. |
| */ |
| public JMSAdapter() { |
| this(false); |
| } |
| |
| /** |
| * Constructs a <code>JMSAdapter</code> instance. |
| * |
| * @param enableManagement <code>true</code> if the <code>JMSAdapter</code> |
| * has a corresponding MBean control for management; otherwise <code>false</code>. |
| */ |
| public JMSAdapter(boolean enableManagement) { |
| super(enableManagement); |
| consumerToClientId = new ConcurrentHashMap<JMSConsumer, Object>(); |
| messageClients = new ConcurrentHashMap<Object, MessageClient>(); |
| topicProducers = new LinkedList<JMSProducer>(); |
| topicConsumers = new ConcurrentHashMap<Object, JMSConsumer>(); |
| queueProducers = new LinkedList<JMSProducer>(); |
| queueConsumers = new ConcurrentHashMap<Object, JMSConsumer>(); |
| settings = new JMSSettings(); |
| } |
| |
| //-------------------------------------------------------------------------- |
| // |
| // Initialize, validate, start, and stop methods. |
| // |
| //-------------------------------------------------------------------------- |
| |
| /** |
| * Initializes the <code>JMSAdapter</code> with the properties. |
| * |
| * @param id The id of the <code>JMSAdapter</code>. |
| * @param properties Properties for the <code>JMSAdapter</code>. |
| */ |
| @Override |
| public void initialize(String id, ConfigMap properties) { |
| super.initialize(id, properties); |
| |
| if (properties == null || properties.size() == 0) |
| return; |
| |
| // JMS specific properties |
| jms(properties); |
| } |
| |
| /** |
| * Verifies that the <code>JMSAdapter</code> is in valid state before |
| * it is started. |
| */ |
| @Override |
| protected void validate() { |
| if (isValid()) |
| return; |
| |
| super.validate(); |
| |
| if (settings.getConnectionFactory() == null) { |
| // JMS connection factory of message destinations with JMS Adapters must be specified. |
| ConfigurationException ce = new ConfigurationException(); |
| ce.setMessage(MISSING_CONNECTION_FACTORY); |
| throw ce; |
| } |
| |
| if (settings.getDestinationJNDIName() == null) { |
| // JNDI names for message destinations with JMS Adapters must be specified. |
| ConfigurationException ce = new ConfigurationException(); |
| ce.setMessage(JMSConfigConstants.MISSING_DESTINATION_JNDI_NAME); |
| throw ce; |
| } |
| |
| if (settings.getMessageType() == null) { |
| // Unsupported JMS Message Type ''{0}''. Valid values are javax.jms.TextMessage, javax.jms.ObjectMessage, and javax.jms.MapMessage. |
| ConfigurationException ce = new ConfigurationException(); |
| ce.setMessage(INVALID_JMS_MESSAGE_TYPE, new Object[]{null}); |
| throw ce; |
| } |
| } |
| |
| /** |
| * Starts the adapter. |
| */ |
| @Override |
| public void start() { |
| if (isStarted()) |
| return; |
| |
| super.start(); |
| |
| // Add JMS adapter as a MessageClient created listener so that its |
| // JMS consumers can be associated with their message clients. |
| MessageClient.addMessageClientCreatedListener(this); |
| } |
| |
| /** |
| * Stops the adapter. |
| */ |
| @Override |
| public void stop() { |
| if (!isStarted()) |
| return; |
| |
| super.stop(); |
| |
| stopConsumers(topicConsumers.values()); |
| stopConsumers(queueConsumers.values()); |
| } |
| |
| //-------------------------------------------------------------------------- |
| // |
| // Public methods |
| // |
| //-------------------------------------------------------------------------- |
| |
| /** |
| * Casts the <code>Destination</code> into <code>MessageDestination</code> |
| * and calls super.setDestination. |
| * |
| * @param destination The destination of the adapter. |
| */ |
| @Override |
| public void setDestination(Destination destination) { |
| MessageDestination dest = (MessageDestination) destination; |
| super.setDestination(dest); |
| } |
| |
| /** |
| * Gets the <code>JMSSettings</code> of the <code>JMSAdapter</code>. |
| * |
| * @return <code>JMSSettings</code> of the <code>JMSAdapter</code>. |
| */ |
| public JMSSettings getJMSSettings() { |
| return settings; |
| } |
| |
| /** |
| * Sets the <code>JMSSettings</code> of the <code>JMSAdapter</code>. |
| * |
| * @param jmsSettings <code>JMSSettings</code> of the <code>JMSAdapter</code>. |
| */ |
| public void setJMSSettings(JMSSettings jmsSettings) { |
| this.settings = jmsSettings; |
| } |
| |
| /** |
| * Returns the count of queue consumers managed by this adapter. |
| * |
| * @return The count of queue consumers managed by this adapter. |
| */ |
| public int getQueueConsumerCount() { |
| return queueConsumers.size(); |
| } |
| |
| /** |
| * Returns the ids of all queue consumers. |
| * |
| * @return The ids of all queue consumers. |
| */ |
| public String[] getQueueConsumerIds() { |
| Set<Object> consumerIds = queueConsumers.keySet(); |
| if (consumerIds != null) { |
| String[] ids = new String[consumerIds.size()]; |
| return consumerIds.toArray(ids); |
| } |
| return new String[0]; |
| } |
| |
| /** |
| * Returns the count of topic consumers currently managed by this adapter. |
| * |
| * @return The count of topic consumers currently managed by this adapter. |
| */ |
| public int getTopicConsumerCount() { |
| return topicConsumers.size(); |
| } |
| |
| /** |
| * Returns the ids of all topic consumers. |
| * |
| * @return The ids of all topic consumers. |
| */ |
| public String[] getTopicConsumerIds() { |
| Set<Object> consumerIds = topicConsumers.keySet(); |
| if (consumerIds != null) { |
| String[] ids = new String[consumerIds.size()]; |
| return consumerIds.toArray(ids); |
| } |
| return new String[0]; |
| } |
| |
| /** |
| * Returns the count of topic producers currently managed by this adapter. |
| * |
| * @return The count of topic producers currently managed by this adapter. |
| */ |
| public int getTopicProducerCount() { |
| return topicProducers.size(); |
| } |
| |
| /** |
| * Returns the count of queue producers currently managed by this adapter. |
| * |
| * @return The count of queue producers currently managed by this adapter. |
| */ |
| public int getQueueProducerCount() { |
| return queueProducers.size(); |
| } |
| |
| /** |
| * Implements JMSExceptionListener. |
| * When a JMSConsumer receives a JMS exception from its underlying JMS |
| * connection, it dispatches a JMS exception event to pass the exception |
| * to the JMS adapter. |
| * |
| * @param evt The <code>JMSExceptionEvent</code>. |
| */ |
| public void exceptionThrown(JMSExceptionEvent evt) { |
| JMSConsumer consumer = (JMSConsumer) evt.getSource(); |
| JMSException jmsEx = evt.getJMSException(); |
| |
| // Client is unsubscribed because its corresponding JMS consumer for JMS destination ''{0}'' encountered an error during message delivery: {1} |
| MessageException messageEx = new MessageException(); |
| messageEx.setMessage(JMSConfigConstants.CLIENT_UNSUBSCRIBE_DUE_TO_MESSAGE_DELIVERY_ERROR, new Object[]{consumer.getDestinationJndiName(), jmsEx.getMessage()}); |
| removeConsumer(consumer, true, true, messageEx.createErrorMessage()); |
| } |
| |
| /** |
| * JMS adapter handles its subscriptions so this returns <code>true</code>. |
| * |
| * @return <code>true</code>. |
| */ |
| @Override |
| public boolean handlesSubscriptions() { |
| return true; |
| } |
| |
| /** |
| * Publish a message to this adapter's JMS destination. |
| * |
| * @param message The Flex message to publish. |
| * @return The body of the acknowledge message which is null is this case. |
| */ |
| @SuppressWarnings("unchecked") |
| @Override |
| public Object invoke(Message message) { |
| JMSProducer producer = null; |
| |
| // named Flex message props become JMS headers |
| Map msgProps = message.getHeaders(); |
| msgProps.put(JMSConfigConstants.TIME_TO_LIVE, new Long(message.getTimeToLive())); |
| |
| if (settings.getDestinationType().equals(TOPIC)) { |
| synchronized (topicProducers) { |
| if (topicProducers.size() < settings.getMaxProducers()) { |
| producer = new JMSTopicProducer(); |
| try { |
| producer.initialize(settings); |
| producer.start(); |
| } catch (Exception e) { |
| throw constructMessageException(e); |
| } |
| } else { |
| producer = topicProducers.removeFirst(); |
| } |
| |
| topicProducers.addLast(producer); |
| } |
| } else if (settings.getDestinationType().equals(QUEUE)) { |
| synchronized (queueProducers) { |
| if (queueProducers.size() < settings.getMaxProducers()) { |
| producer = new JMSQueueProducer(); |
| try { |
| producer.initialize(settings); |
| producer.start(); |
| } catch (Exception e) { |
| throw constructMessageException(e); |
| } |
| } else { |
| producer = queueProducers.removeFirst(); |
| } |
| |
| queueProducers.addLast(producer); |
| } |
| } |
| |
| try { |
| if (producer != null) |
| producer.sendMessage(message); |
| } catch (JMSException jmsEx) { |
| // At this point we give up on this producer, so we just |
| // stop and remove it from the pool. |
| if (settings.getDestinationType().equals(TOPIC)) { |
| synchronized (topicProducers) { |
| if (producer != null) { |
| producer.stop(); |
| topicProducers.remove(producer); |
| } |
| } |
| } else if (settings.getDestinationType().equals(QUEUE)) { |
| synchronized (queueProducers) { |
| if (producer != null) { |
| producer.stop(); |
| queueProducers.remove(producer); |
| } |
| } |
| } |
| |
| throw constructMessageException(jmsEx); |
| } |
| |
| return null; |
| } |
| |
| /** |
| * Handle a CommandMessage sent by this adapter's service. |
| * |
| * @param commandMessage The command message to manage. |
| * @return The result of manage which is null in this case. |
| */ |
| @Override |
| public Object manage(CommandMessage commandMessage) { |
| JMSConsumer consumer = null; |
| Object clientId = commandMessage.getClientId(); |
| |
| if (commandMessage.getOperation() == CommandMessage.SUBSCRIBE_OPERATION) { |
| // Keep track of the selector expression. |
| Object selectorExpression = commandMessage.getHeaders().get(CommandMessage.SELECTOR_HEADER); |
| |
| // Create a JMSConsumer for this destination and associate it with the client id |
| if (settings.getDestinationType().equals(TOPIC)) { |
| MessageClient existingMessageClient = null; |
| // This could happen when client disconnects without unsubscribing first. |
| if (topicConsumers.containsKey(clientId)) { |
| removeConsumer(clientId, true /*unsubscribe*/, false /*invalidate*/, null); |
| existingMessageClient = messageClients.get(clientId); |
| } |
| // Create the consumer. |
| consumer = new JMSTopicConsumer(); |
| consumer.initialize(settings); |
| if (selectorExpression != null) |
| consumer.setSelectorExpression((String) selectorExpression); |
| // Need to build a subscription name, in case durable subscriptions are used. |
| ((JMSTopicConsumer) consumer).setDurableSubscriptionName(buildSubscriptionName(clientId)); |
| consumer.setMessageReceiver(buildMessageReceiver(consumer)); |
| |
| // Add JMSAdapter as JMS exception and message listener. |
| consumer.addJMSExceptionListener(this); |
| consumer.addJMSMessageListener(this); |
| topicConsumers.put(clientId, consumer); |
| consumerToClientId.put(consumer, clientId); |
| |
| // Means client was disconnected without unsubscribing, hence no |
| // new message client will be created. Make sure the old one is |
| // wired up with the new JMS consumer properly. |
| if (existingMessageClient != null) |
| messageClientCreated(existingMessageClient); |
| } else if (settings.getDestinationType().equals(QUEUE)) { |
| MessageClient existingMessageClient = null; |
| if (queueConsumers.containsKey(clientId)) { |
| removeConsumer(clientId, true /*unsubscribe*/, false /*invalidate*/, null); |
| existingMessageClient = messageClients.get(clientId); |
| } |
| // Create the consumer. |
| consumer = new JMSQueueConsumer(); |
| consumer.initialize(settings); |
| if (selectorExpression != null) |
| consumer.setSelectorExpression((String) selectorExpression); |
| consumer.setMessageReceiver(buildMessageReceiver(consumer)); |
| |
| // Add JMSAdapter as JMS exception and message listener. |
| consumer.addJMSExceptionListener(this); |
| consumer.addJMSMessageListener(this); |
| queueConsumers.put(clientId, consumer); |
| consumerToClientId.put(consumer, clientId); |
| |
| // Means client was disconnected without unsubscribing, hence no |
| // new message client will be created. Make sure the old one is |
| // wired up with the new JMS consumer properly. |
| if (existingMessageClient != null) |
| messageClientCreated(existingMessageClient); |
| } |
| } else if (commandMessage.getOperation() == CommandMessage.UNSUBSCRIBE_OPERATION) { |
| // Determines if the durable subscription should be unsubscribed |
| // when the JMS consumer is removed. |
| boolean unsubscribe = true; |
| |
| boolean preserveDurable = false; |
| if (commandMessage.getHeader(CommandMessage.PRESERVE_DURABLE_HEADER) != null) |
| preserveDurable = ((Boolean) (commandMessage.getHeader(CommandMessage.PRESERVE_DURABLE_HEADER))).booleanValue(); |
| |
| // Don't destroy a durable subscription if the MessageClient's session has been invalidated. |
| // or this is a JMS durable connection that has requested to be undestroyed |
| if (commandMessage.getHeader(CommandMessage.SUBSCRIPTION_INVALIDATED_HEADER) != null |
| && ((Boolean) commandMessage.getHeader(CommandMessage.SUBSCRIPTION_INVALIDATED_HEADER)).booleanValue() |
| || preserveDurable) |
| unsubscribe = false; |
| |
| removeConsumer(clientId, unsubscribe, false, null); |
| } |
| |
| // CommandMessage.POLL_OPERATION handling is left to the Endpoint |
| // hence not handled by this adapter. |
| |
| return null; |
| } |
| |
| /** |
| * Implements MessageClientListener. |
| * When a MessageClient is created, JMSAdapter looks up its JMSConsumers |
| * and if there is a JMSConsumer that has the same clientId as MessageClient, |
| * it adds the MessageClient to its list clients. This helps in keeping both |
| * sides of the bridge (MessageClient and JMSConsumer) notified when there's |
| * a failure on either side of the bridge. |
| * |
| * @param messageClient The newly created MessageClient. |
| */ |
| public void messageClientCreated(MessageClient messageClient) { |
| Object clientId = messageClient.getClientId(); |
| JMSConsumer consumer = null; |
| if (topicConsumers.containsKey(clientId)) |
| consumer = topicConsumers.get(clientId); |
| else if (queueConsumers.containsKey(clientId)) |
| consumer = queueConsumers.get(clientId); |
| |
| // If there is a JMSConsumer created for the same clientId, register |
| // the MessageClient with JMSAdapter and start the consumer. |
| if (consumer != null) { |
| messageClients.put(clientId, messageClient); |
| try { |
| consumer.start(); |
| // Add JMS adapter as a client destroyed listener, so client |
| // invalidation (eg. due to session timeout) can be handled properly. |
| messageClient.addMessageClientDestroyedListener(this); |
| } catch (MessageException messageEx) { |
| removeConsumer(consumer, true, true, messageEx.createErrorMessage()); |
| } catch (Exception ex) { |
| removeConsumer(consumer, true, true, constructMessageException(ex).createErrorMessage()); |
| } |
| } |
| } |
| |
| /** |
| * Implements MessageClientListener. |
| * When a MessageClient is destroyed (usually due to session timeout), its |
| * corresponding JMS consumer is removed. Note that this might have already |
| * happened if the client first unsubscribes and in that case, this is a no-op. |
| * |
| * @param messageClient The MessageClient that was destroyed. |
| */ |
| public void messageClientDestroyed(MessageClient messageClient) { |
| Object clientId = messageClient.getClientId(); |
| removeConsumer(clientId); |
| messageClients.remove(clientId); |
| } |
| |
| /** |
| * Implements JMSMessageListener. |
| * When a JMSConsumer receives a JMS message, it dispatched a JMS message |
| * event to pass the message to the JMS adapter. |
| * |
| * @param evt The <code>JMSMessageEvent</code>. |
| */ |
| public void messageReceived(JMSMessageEvent evt) { |
| JMSConsumer consumer = (JMSConsumer) evt.getSource(); |
| javax.jms.Message jmsMessage = evt.getJMSMessage(); |
| |
| flex.messaging.messages.AsyncMessage flexMessage = convertToFlexMessage(jmsMessage, consumer); |
| if (flexMessage != null) { |
| MessagePerformanceUtils.markServerPostAdapterExternalTime(flexMessage); |
| ((MessageService) getDestination().getService()).serviceMessageFromAdapter(flexMessage, false); |
| } |
| } |
| |
| /** |
| * Removes (unsubscribes) the specified consumer. By default, it removes |
| * the durable subscription and pushes a generic error message to the client |
| * before MessageClient invalidation. |
| * |
| * @param clientId The identifier for the consumer to remove. |
| */ |
| public void removeConsumer(Object clientId) { |
| // Client is unsubscribed because its corresponding JMS consumer has been removed from the JMS adapter. |
| MessageException messageEx = new MessageException(); |
| messageEx.setMessage(JMSConfigConstants.CLIENT_UNSUBSCRIBE_DUE_TO_CONSUMER_REMOVAL); |
| removeConsumer(clientId, true, true, messageEx.createErrorMessage()); |
| } |
| |
| //-------------------------------------------------------------------------- |
| // |
| // Protected and Private methods |
| // |
| //-------------------------------------------------------------------------- |
| |
| /** |
| * Removes (unsubscribes) the JMSConsumer associated with the clientId. |
| * |
| * @param clientId The clientId associated with the JMSConsumer to remove. |
| * @param unsubscribe Whether to unsubscribe the durable subscription or not. |
| * @param invalidate Whether to invalidate the MessageClient or not. |
| * @param invalidateMessage A message to push to the client before consumer |
| * is removed and its MessageClient is invalidated. If the message is null, |
| * MessageClient is invalidated silently. |
| */ |
| protected void removeConsumer(Object clientId, boolean unsubscribe, boolean invalidate, ErrorMessage invalidateMessage) { |
| JMSConsumer consumer = null; |
| if (topicConsumers.containsKey(clientId)) |
| consumer = topicConsumers.get(clientId); |
| else if (queueConsumers.containsKey(clientId)) |
| consumer = queueConsumers.get(clientId); |
| |
| removeConsumer(consumer, unsubscribe, invalidate, invalidateMessage); |
| } |
| |
| /** |
| * Removes (unsubscribes) the specified consumer. |
| * |
| * @param consumer The JMSConsumer instance to remove. |
| * @param unsubscribe Whether to unsubscribe the durable subscription or not. |
| * @param invalidate Whether to invalidate the MessageClient or not. |
| * @param invalidateMessage A message to push to the client before consumer |
| * is removed and its MessageClient is invalidated. If the message is null, |
| * MessageClient is invalidated silently. |
| */ |
| protected void removeConsumer(JMSConsumer consumer, boolean unsubscribe, boolean invalidate, ErrorMessage invalidateMessage) { |
| if (consumer == null) |
| return; |
| |
| Object clientId = consumerToClientId.get(consumer); |
| if (clientId == null) |
| return; |
| |
| if (Log.isInfo()) { |
| String logMessage = "JMS consumer for JMS destination '" + consumer.getDestinationJndiName() |
| + "' is being removed from the JMS adapter"; |
| |
| if (invalidateMessage != null) |
| logMessage += " due to the following error: " + invalidateMessage.faultString; |
| |
| Log.getLogger(JMSAdapter.LOG_CATEGORY).info(logMessage); |
| } |
| |
| consumer.removeJMSExceptionListener(this); |
| consumer.removeJMSMessageListener(this); |
| consumer.stop(unsubscribe); |
| if (invalidate) |
| invalidateMessageClient(consumer, invalidateMessage); |
| if (consumer instanceof JMSTopicConsumer) |
| topicConsumers.remove(clientId); |
| else // assuming JMSQueueConsumer. |
| queueConsumers.remove(clientId); |
| consumerToClientId.remove(consumer); |
| // Message client will be removed in messageClientDestroyed. |
| } |
| |
| /** |
| * Invoked automatically to allow the <code>JMSAdapter</code> to setup its corresponding |
| * MBean control. |
| * |
| * @param destination The <code>Destination</code> that manages this <code>JMSAdapter</code>. |
| */ |
| @Override |
| protected void setupAdapterControl(Destination destination) { |
| controller = new JMSAdapterControl(this, destination.getControl()); |
| controller.register(); |
| setControl(controller); |
| } |
| |
| /** |
| * Builds a MessageReceiver for JMSConsumer from DeliverySettings. |
| * |
| * @param consumer The <code>JMSConsumer</code>. |
| * @return MessageReceiver configured for JMSConsumer per DeliverySettings. |
| */ |
| private MessageReceiver buildMessageReceiver(JMSConsumer consumer) { |
| DeliverySettings deliverySettings = settings.getDeliverySettings(); |
| if (deliverySettings.getMode().equals(JMSConfigConstants.ASYNC)) |
| return new AsyncMessageReceiver(consumer); |
| SyncMessageReceiver syncMessageReceiver = new SyncMessageReceiver(consumer); |
| syncMessageReceiver.setSyncReceiveIntervalMillis(deliverySettings.getSyncReceiveIntervalMillis()); |
| syncMessageReceiver.setSyncReceiveWaitMillis(deliverySettings.getSyncReceiveWaitMillis()); |
| return syncMessageReceiver; |
| } |
| |
| /** |
| * Prefixes a clientId with DURABLE_SUBSCRIBER_NAME_PREFIX to build a |
| * subscription name to be used in JMSConsumers with durable connections. |
| */ |
| private String buildSubscriptionName(Object clientId) { |
| return DURABLE_SUBSCRIBER_NAME_PREFIX + clientId.toString(); |
| } |
| |
| /** |
| * Construct a MessageException for the JMS invocation |
| * |
| * @param e the Exception caught in the JMS invocation |
| * @return MessageException encapsulates the JMS Exception message |
| */ |
| private MessageException constructMessageException(Exception e) { |
| MessageException messageEx = new MessageException(); |
| messageEx.setMessage(JMSINVOCATION_EXCEPTION, new Object[]{e.getMessage()}); |
| return messageEx; |
| } |
| |
| /** |
| * Convert from a <code>javax.jms.Message</code> type to the <code>flex.messaging.messages.AsyncMessage</code> type. |
| * Supported types are <code>javax.jms.TextMessage</code>, <code>javax.jms.ObjectMessage</code>, |
| * and <code>javax.jms.MapMessage</code>. |
| */ |
| private flex.messaging.messages.AsyncMessage convertToFlexMessage(javax.jms.Message jmsMessage, JMSConsumer consumer) { |
| flex.messaging.messages.AsyncMessage flexMessage = null; |
| flexMessage = new flex.messaging.messages.AsyncMessage(); |
| |
| Object clientId = consumerToClientId.get(consumer); |
| if (clientId == null) { |
| if (Log.isWarn()) |
| Log.getLogger(LOG_CATEGORY).warn("JMSAdapter encountered a null clientId during JMS to Flex message conversion"); |
| |
| return null; |
| } |
| flexMessage.setClientId(clientId); |
| |
| |
| flexMessage.setDestination(getDestination().getId()); |
| |
| // Set JMSMessageID header as Flex messageId property. |
| try { |
| flexMessage.setMessageId(jmsMessage.getJMSMessageID()); |
| } catch (JMSException jmsEx) { |
| if (Log.isWarn()) |
| Log.getLogger(LOG_CATEGORY).warn("JMSAdapter encountered an error while retrieving JMS message id during JMS to Flex message conversion: " + jmsEx.getMessage()); |
| } |
| |
| // Set JMSTimestamp header as Flex timestamp property. |
| try { |
| flexMessage.setTimestamp(jmsMessage.getJMSTimestamp()); |
| } catch (JMSException jmsEx) { |
| if (Log.isWarn()) |
| Log.getLogger(LOG_CATEGORY).warn("JMSAdapter encountered an error while retrieving JMS timestamp during JMS to Flex message conversion: " + jmsEx.getMessage()); |
| } |
| |
| // Set JMS headers and Flex headers. |
| if (settings.isPreserveJMSHeaders()) { |
| // Set standard JMS headers except JMSMessageId and JMSTimestamp, |
| // as they are already set on the Flex message directly. |
| try { |
| flexMessage.setHeader(JMS_CORRELATION_ID, jmsMessage.getJMSCorrelationID()); |
| flexMessage.setHeader(JMS_DELIVERY_MODE, Integer.toString(jmsMessage.getJMSDeliveryMode())); |
| flexMessage.setHeader(JMS_DESTINATION, jmsMessage.getJMSDestination().toString()); |
| flexMessage.setHeader(JMS_EXPIRATION, Long.toString(jmsMessage.getJMSExpiration())); |
| flexMessage.setHeader(JMS_PRIORITY, Integer.toString(jmsMessage.getJMSPriority())); |
| flexMessage.setHeader(JMS_REDELIVERED, Boolean.toString(jmsMessage.getJMSRedelivered())); |
| flexMessage.setHeader(JMS_REPLY_TO, jmsMessage.getJMSReplyTo()); |
| flexMessage.setHeader(JMS_TYPE, jmsMessage.getJMSType()); |
| } catch (JMSException jmsEx) { |
| // These should not cause errors to be pushed to Flash clients |
| if (Log.isWarn()) |
| Log.getLogger(LOG_CATEGORY).warn("JMSAdapter encountered an error while retrieving JMS headers during JMS to Flex conversion: " + jmsEx.getMessage()); |
| |
| } |
| } |
| |
| // Set JMS properties as Flex headers. |
| |
| // While iterating through JMS message properties, build a message |
| // performance object to send back to the client with the message |
| // properties starting with with MPI_HEADER_IN (if any). |
| MessagePerformanceInfo mpi = null; |
| |
| try { |
| for (Enumeration propEnum = jmsMessage.getPropertyNames(); propEnum.hasMoreElements(); ) { |
| String propName = (String) propEnum.nextElement(); |
| try { |
| Object propValue = jmsMessage.getObjectProperty(propName); |
| if (propName.startsWith(MessagePerformanceUtils.MPI_HEADER_IN)) { |
| if (mpi == null) |
| mpi = new MessagePerformanceInfo(); |
| propName = propName.substring(MessagePerformanceUtils.MPI_HEADER_IN.length()); |
| java.lang.reflect.Field field; |
| try { |
| field = mpi.getClass().getField(propName); |
| field.set(mpi, propValue); |
| } catch (Exception ignore) { |
| // Simply don't set the property if the value cannot be retrieved. |
| } |
| } else { |
| flexMessage.setHeader(propName, propValue); |
| } |
| } catch (JMSException jmsEx) { |
| if (Log.isWarn()) |
| Log.getLogger(LOG_CATEGORY).warn("JMSAdapter encountered an error while retrieving JMS properties during JMS to Flex conversion: " + jmsEx.getMessage()); |
| } |
| } |
| |
| if (mpi != null) |
| flexMessage.setHeader(MessagePerformanceUtils.MPI_HEADER_IN, mpi); |
| } catch (JMSException jmsEx) { |
| if (Log.isWarn()) |
| Log.getLogger(LOG_CATEGORY).warn("JMSAdapter encountered an error while retrieving JMS properties during JMS to Flex conversion: " + jmsEx.getMessage()); |
| } |
| |
| // Finally, set the JMS message body of the Flex message body. |
| try { |
| if (jmsMessage instanceof javax.jms.TextMessage) { |
| javax.jms.TextMessage textMessage = (javax.jms.TextMessage) jmsMessage; |
| flexMessage.setBody(textMessage.getText()); |
| } else if (jmsMessage instanceof javax.jms.ObjectMessage) { |
| javax.jms.ObjectMessage objMessage = (javax.jms.ObjectMessage) jmsMessage; |
| flexMessage.setBody(objMessage.getObject()); |
| } else if (jmsMessage instanceof javax.jms.MapMessage) { |
| javax.jms.MapMessage mapMessage = (javax.jms.MapMessage) jmsMessage; |
| @SuppressWarnings("unchecked") |
| Enumeration names = mapMessage.getMapNames(); |
| Map<String, Object> body = new HashMap<String, Object>(); |
| while (names.hasMoreElements()) { |
| String name = (String) names.nextElement(); |
| body.put(name, mapMessage.getObject(name)); |
| } |
| flexMessage.setBody(body); |
| } |
| } catch (JMSException jmsEx) { |
| if (Log.isWarn()) |
| Log.getLogger(LOG_CATEGORY).warn("JMSAdapter encountered an error while retrieving JMS message body during JMS to Flex conversion: " + jmsEx.getMessage()); |
| } |
| |
| return flexMessage; |
| } |
| |
| /** |
| * Invalidates the MessageClient associated with the JMSConsumer with the |
| * supplied error message. |
| * |
| * @param consumer The JMSConsumer whose MessageClient will be invalidated. |
| * @param message The error message to push out before invalidating the |
| * MessageClient. If the message is null, MessageClient is invalidated |
| * silently. |
| */ |
| private void invalidateMessageClient(JMSConsumer consumer, flex.messaging.messages.Message message) { |
| Object clientId = consumerToClientId.get(consumer); |
| if (clientId != null && messageClients.containsKey(clientId)) { |
| MessageClient messageClient = messageClients.get(clientId); |
| |
| if (Log.isInfo()) |
| Log.getLogger(JMSAdapter.LOG_CATEGORY).info("The corresponding MessageClient for JMS consumer for JMS destination '" |
| + consumer.getDestinationJndiName() + "' is being invalidated"); |
| |
| messageClient.invalidate(message); |
| } |
| } |
| |
| /** |
| * Handle JMS specific configuration. |
| */ |
| private void jms(ConfigMap properties) { |
| ConfigMap jms = properties.getPropertyAsMap(JMS, null); |
| if (jms != null) { |
| String destType = jms.getPropertyAsString(DESTINATION_TYPE, defaultDestinationType); |
| settings.setDestinationType(destType); |
| |
| String msgType = jms.getPropertyAsString(MESSAGE_TYPE, null); |
| settings.setMessageType(msgType); |
| |
| String factory = jms.getPropertyAsString(CONNECTION_FACTORY, null); |
| settings.setConnectionFactory(factory); |
| |
| ConfigMap connectionCredentials = jms.getPropertyAsMap(CONNECTION_CREDENTIALS, null); |
| if (connectionCredentials != null) { |
| String username = connectionCredentials.getPropertyAsString(USERNAME, null); |
| settings.setConnectionUsername(username); |
| String password = connectionCredentials.getPropertyAsString(PASSWORD, null); |
| settings.setConnectionPassword(password); |
| } |
| |
| ConfigMap deliverySettings = jms.getPropertyAsMap(DELIVERY_SETTINGS, null); |
| if (deliverySettings != null) { |
| // Get the default delivery settings. |
| DeliverySettings ds = settings.getDeliverySettings(); |
| |
| String mode = deliverySettings.getPropertyAsString(MODE, JMSConfigConstants.defaultMode); |
| ds.setMode(mode); |
| |
| long receiveIntervalMillis = deliverySettings.getPropertyAsLong(SYNC_RECEIVE_INTERVAL_MILLIS, defaultSyncReceiveIntervalMillis); |
| ds.setSyncReceiveIntervalMillis(receiveIntervalMillis); |
| |
| long receiveWaitMillis = deliverySettings.getPropertyAsLong(SYNC_RECEIVE_WAIT_MILLIS, defaultSyncReceiveWaitMillis); |
| ds.setSyncReceiveWaitMillis(receiveWaitMillis); |
| } |
| |
| String destJNDI = jms.getPropertyAsString(DESTINATION_JNDI_NAME, null); |
| settings.setDestinationJNDIName(destJNDI); |
| |
| String dest = jms.getPropertyAsString(DESTINATION_NAME, null); |
| if (dest != null && Log.isWarn()) |
| Log.getLogger(LOG_CATEGORY).warn("The <destination-name> configuration option is deprecated and non-functional. Please remove this from your configuration file."); |
| |
| boolean durable = getDestination() instanceof MessageDestination ? |
| ((MessageDestination) getDestination()).getServerSettings().isDurable() : false; |
| settings.setDurableConsumers(durable); |
| |
| String deliveryMode = jms.getPropertyAsString(DELIVERY_MODE, null); |
| settings.setDeliveryMode(deliveryMode); |
| |
| boolean preserveJMSHeaders = jms.getPropertyAsBoolean(PRESERVE_JMS_HEADERS, settings.isPreserveJMSHeaders()); |
| settings.setPreserveJMSHeaders(preserveJMSHeaders); |
| |
| String defPriority = jms.getPropertyAsString(MESSAGE_PRIORITY, null); |
| if (defPriority != null && !defPriority.equalsIgnoreCase(DEFAULT_PRIORITY)) { |
| int priority = jms.getPropertyAsInt(MESSAGE_PRIORITY, settings.getMessagePriority()); |
| settings.setMessagePriority(priority); |
| } |
| |
| String ackMode = jms.getPropertyAsString(ACKNOWLEDGE_MODE, defaultAcknowledgeMode); |
| settings.setAcknowledgeMode(ackMode); |
| |
| boolean transMode = jms.getPropertyAsBoolean(TRANSACTION_MODE, false); |
| if (transMode && Log.isWarn()) |
| Log.getLogger(LOG_CATEGORY).warn("The <transacted-sessions> configuration option is deprecated and non-functional. Please remove this from your configuration file."); |
| |
| int maxProducers = jms.getPropertyAsInt(MAX_PRODUCERS, defaultMaxProducers); |
| settings.setMaxProducers(maxProducers); |
| |
| // Retrieve any JNDI initial context environment properties. |
| ConfigMap env = jms.getPropertyAsMap(INITIAL_CONTEXT_ENVIRONMENT, null); |
| if (env != null) { |
| List props = env.getPropertyAsList(PROPERTY, null); |
| if (props != null) { |
| Class contextClass = Context.class; |
| Hashtable envProps = new Hashtable(); |
| for (Iterator iter = props.iterator(); iter.hasNext(); ) { |
| Object prop = iter.next(); |
| if (prop instanceof ConfigMap) { |
| ConfigMap pair = (ConfigMap) prop; |
| String name = pair.getProperty(NAME); |
| String value = pair.getProperty(VALUE); |
| if (name == null || value == null) { |
| // A <property> element for the <initial-context-environment> settings for the ''{0}'' destination does not specify both <name> and <value> subelements. |
| MessageException messageEx = new MessageException(); |
| messageEx.setMessage(MISSING_NAME_OR_VALUE, new Object[]{getDestination().getId()}); |
| throw messageEx; |
| } |
| // If the name is a Context field, use the |
| // constant value rather than this literal name. |
| if (name.startsWith("Context.")) { |
| String fieldName = name.substring(name.indexOf('.') + 1); |
| java.lang.reflect.Field field = null; |
| try { |
| field = contextClass.getDeclaredField(fieldName); |
| } catch (NoSuchFieldException nsfe) { |
| // A <property> element for the <initial-context-environment> settings for the ''{0}'' destination specifies an invalid javax.naming.Context field for its <name>: {1} |
| MessageException messageEx = new MessageException(); |
| messageEx.setMessage(INVALID_CONTEXT_NAME, new Object[]{getDestination().getId(), fieldName}); |
| throw messageEx; |
| } |
| String fieldValue = null; |
| try { |
| fieldValue = (String) field.get(null); |
| } catch (IllegalAccessException iae) { |
| // A <property> element for the <initial-context-environment> settings for the ''{0}'' destination specifies an inaccessible javax.naming.Context field for its <name>: {1} |
| MessageException messageEx = new MessageException(); |
| messageEx.setMessage(INACCESIBLE_CONTEXT_NAME, new Object[]{getDestination().getId(), fieldName}); |
| throw messageEx; |
| } |
| envProps.put(fieldValue, value); |
| } else { |
| envProps.put(name, value); |
| } |
| } else { |
| // A <property> element for the <initial-context-environment> settings for the ''{0}'' destination does not specify both <name> and <value> subelements. |
| MessageException messageEx = new MessageException(); |
| messageEx.setMessage(MISSING_NAME_OR_VALUE, new Object[]{getDestination().getId()}); |
| throw messageEx; |
| } |
| } |
| settings.setInitialContextEnvironment(envProps); |
| } else { |
| // The <initial-context-environment> settings for the ''{0}'' destination does not include any <property> subelements. |
| MessageException messageEx = new MessageException(); |
| messageEx.setMessage(MISSING_PROPERTY_SUBELEMENT, new Object[]{getDestination().getId()}); |
| throw messageEx; |
| } |
| } |
| } |
| } |
| |
| private void stopConsumers(Collection<JMSConsumer> consumers) { |
| Iterator<JMSConsumer> itr = consumers.iterator(); |
| while (itr.hasNext()) { |
| JMSConsumer consumer = itr.next(); |
| // Client is unsubscribed because its corresponding JMS consumer for JMS destination ''{0}'' has been stopped. |
| MessageException me = new MessageException(); |
| me.setMessage(JMSConfigConstants.CLIENT_UNSUBSCRIBE_DUE_TO_CONSUMER_STOP, new Object[]{consumer.getDestinationJndiName()}); |
| consumer.stop(true); |
| invalidateMessageClient(consumer, me.createErrorMessage()); |
| } |
| } |
| } |
| |
