content: /*
   * 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 org.apache.hadoop.chukwa.datacollection.adaptor.jms;
  
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  
  import javax.jms.Message;
  import javax.jms.JMSException;
  
  import java.nio.charset.Charset;
  import java.util.ArrayList;
  
  /**
   * JMSMessageTransformer that uses the properties of a JMS Message to build a
   * Chukwa record payload. The value for each property configured will be used
   * to create the record, with the delimiter value between each. The default
   * delimiter is a tab (i.e., '\t').
   * <P>
   * To configure this transformer, set the -p field of the adaptor to the
   * following (surrounded with double quotes):
   * <code>
   * &lt;propertyNames&gt; [-d &lt;delimiter&gt;] [-r &lt;requiredPropertyNames&gt;]
   * </code>
   * <ul>
   * <li><code>propertyNames</code> - Comma-separated list of JMS properties.</li>
   * <li><code>delimiter</code> - Delimiter to use, in single quotes.</li>
   * <li><code>requiredPropertyNames</code> - Comma-separated list of required
   * JMS properties. Default behavior is that all properties are required.</li>
   * </ul>
   *
   */
  public class JMSMessagePropertyTransformer implements JMSMessageTransformer {
    protected Log log = LogFactory.getLog(getClass());
  
    private static final String DEFAULT_DELIMITER = "\t";
  
    ArrayList<String> propertyNames = null;
    ArrayList<String> requiredPropertyNames = null;
    String delimiter = DEFAULT_DELIMITER;
  
    public String parseArgs(String args) {
      if (args == null || args.length() == 0) {
        log.error("propertyNames must be set for this transformer");
        return null;
      }
  
      log.info("Initializing JMSMessagePropertyTransformer: args=" + args);
  
      propertyNames = new ArrayList<String>();
      
      String[] tokens = args.split(" ");
      for (String propertyName : tokens[0].split(",")) {
        propertyNames.add(propertyName);
      }
  
      for(int i = 1; i < tokens.length; i++) {
        String token = tokens[i];
  
        if ("-d".equals(token) && i <= tokens.length - 2) {
          StringBuilder value = new StringBuilder();
          value.append(tokens[++i]);
  
          // we lost all spaces with the split, so we have to put them back, yuck.
          while (i <= tokens.length - 2 && !tokens[i + 1].startsWith("-")) {
            value.append(" ");
            value.append(tokens[++i]);
          }
  
          delimiter = trimSingleQuotes(value.toString());
        }
        else if ("-r".equals(token) && i <= tokens.length - 2) {
          // requiredPropertyNames = null means all are required.
          requiredPropertyNames = new ArrayList<String>();
  
          String[] required = tokens[++i].split(",");
          for (String r : required) {
            requiredPropertyNames.add(r);
          }
        }
      }
  
      log.info("Initialized JMSMessagePropertyTransformer: delimiter='" +
              delimiter + "', propertyNames=" + propertyNames +
              ", requiredProperties=" +
              (requiredPropertyNames == null ? "ALL" : requiredPropertyNames));
      return args;
    }
  
    /**
     * Transforms message propertes into a byte array delimtied by delimiter. If
     * all of the configured message properties are not found, returns null.
     * <P>
     * The could be enhanced to support the concept of optional/required properties.
     * @param message is data to be transported
     * @return byte array
     * @throws JMSException if problem transforming data
     */
    public byte[] transform(Message message) throws JMSException {
  
      if (propertyNames == null || propertyNames.size() == 0) {
        log.error("No message properties configured for this JMS transformer.");
        return null;
      }
  
      int valuesFound = 0;
      StringBuilder sb = new StringBuilder();
      for (String propertyName : propertyNames) {
        Object propertyValue = message.getObjectProperty(propertyName);
        String value = transformValue(propertyName, propertyValue);
  
        // is a required value not found?
        if (value == null && (requiredPropertyNames == null ||
            requiredPropertyNames.contains(propertyName))) {
          return null;
        }
  
        if (valuesFound > 0) {
          sb.append(delimiter);
        }
  
        if (value != null) {
          sb.append(value);
        }
  
        valuesFound++;
      }
  
      if (sb.length() == 0 || valuesFound != propertyNames.size()) {
        return null;
      }
  
      return sb.toString().getBytes(Charset.forName("UTF-8"));
    }
  
    /**
     * Transforms the propertyValue found into the string that should be used for
     * the message. Can handle String values and Number values. Override this method
     * to handle other Java types, or to apply other value transformation logic.
     *
     * @param propertyName The name of the JMS property
     * @param propertyValue The value of the property, which might be null.
     * @return
     */
    protected String transformValue(String propertyName, Object propertyValue) {
  
      if (propertyValue == null) {
        return null;
      }
      else if (propertyValue instanceof String) {
        return (String)propertyValue;
      }
      else if (propertyValue instanceof Number) {
        return propertyValue.toString();
      }
  
      return null;
    }
  
    private static String trimSingleQuotes(String value) {
      if (value.length() == 0) {
        return value;
      }
  
      // trim leading and trailing quotes
      if (value.charAt(0) == '\'') {
        value = value.substring(1);
      }
      if (value.length() > 0 && value.charAt(value.length() - 1) == '\'') {
        value = value.substring(0, value.length() - 1);
      }
      
      return value;
    }
  
  }
  