001 /**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements. See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership. The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License. You may obtain a copy of the License at
009 *
010 * http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018 package org.apache.oozie.client.event.jms;
019
020 import java.io.IOException;
021 import java.io.InputStream;
022 import java.util.Properties;
023
024 import javax.jms.JMSException;
025 import javax.jms.Message;
026 import javax.jms.TextMessage;
027
028 import org.apache.oozie.client.event.message.EventMessage;
029
030 /**
031 * Client utility to convert JMS message to EventMessage object
032 */
033 public class JMSMessagingUtils {
034 private static final String DESERIALIZER_PROP = "oozie.msg.deserializer.";
035 private static MessageDeserializer deserializer;
036 private static Properties jmsDeserializerInfo;
037 private static final String CLIENT_PROPERTIES = "oozie_client.properties";
038
039 static {
040 InputStream is = JMSMessagingUtils.class.getClassLoader().getResourceAsStream(CLIENT_PROPERTIES);
041 if (is == null) {
042 System.out.println("Using default JSON Deserializer");
043 deserializer = new JSONMessageDeserializer();
044 }
045 else {
046 jmsDeserializerInfo = new Properties();
047 try {
048 jmsDeserializerInfo.load(is);
049 is.close();
050 }
051 catch (IOException ioe) {
052 throw new RuntimeException("I/O error occured for " + CLIENT_PROPERTIES, ioe);
053 }
054 }
055
056 }
057
058 /**
059 * Constructs the EventMessage object from JMS message
060 *
061 * @param <T>
062 * @param msg the JMS message
063 * @return the EventMessage
064 * @throws IOException
065 * @throws JMSException
066 */
067 public static <T extends EventMessage> T getEventMessage(Message msg) throws IOException, JMSException {
068 if (msg == null) {
069 throw new IllegalArgumentException("Could not extract EventMessage as JMS message is null");
070 }
071 if (deserializer == null) {
072 String msgFormat = msg.getStringProperty(JMSHeaderConstants.MESSAGE_FORMAT);
073 deserializer = getDeserializer(msgFormat);
074 }
075 return deserializer.getEventMessage(msg);
076 }
077
078 private static MessageDeserializer getDeserializer(String msgFormat) throws IOException {
079 String deserializerString = (String) jmsDeserializerInfo.get(DESERIALIZER_PROP + msgFormat);
080 if (deserializerString == null) {
081 return new JSONMessageDeserializer();
082 }
083 else {
084 try {
085 MessageDeserializer msgDeserializer = (MessageDeserializer) Class.forName(deserializerString)
086 .newInstance();
087 return msgDeserializer;
088 }
089 catch (Exception cnfe) {
090 throw new IllegalArgumentException("Could not access class " + deserializerString, cnfe);
091 }
092
093 }
094 }
095 }