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
019package org.apache.oozie.coord.input.dependency;
020
021import java.io.UnsupportedEncodingException;
022
023import org.apache.commons.lang.StringUtils;
024import org.apache.oozie.StringBlob;
025import org.apache.oozie.util.WritableUtils;
026import org.apache.oozie.util.XLog;
027
028public class CoordInputDependencyFactory {
029
030    // We need to choose magic number which is not allowed for file/dir.
031    // Magic number is ::$
032    private static final byte[] MAGIC_NUMBER = new byte[] { 58, 58, 36 };
033    public static final String CHAR_ENCODING = "ISO-8859-1";
034
035    public static XLog LOG = XLog.getLog(CoordInputDependencyFactory.class);
036
037    /**
038     * Create the pull dependencies.
039     *
040     * @param isInputLogicSpecified to check if input logic is enable
041     * @return the pull dependencies
042     */
043    public static CoordInputDependency createPullInputDependencies(boolean isInputLogicSpecified) {
044        if (!isInputLogicSpecified) {
045            return new CoordOldInputDependency();
046        }
047        else {
048            return new CoordPullInputDependency();
049        }
050    }
051
052    /**
053     * Create the push dependencies.
054     *
055     * @param isInputLogicSpecified to check if input logic is enable
056     * @return the push dependencies
057     */
058    public static CoordInputDependency createPushInputDependencies(boolean isInputLogicSpecified) {
059        if (!isInputLogicSpecified) {
060            return new CoordOldInputDependency();
061        }
062        else {
063            return new CoordPushInputDependency();
064        }
065    }
066
067    /**
068     * Gets the pull input dependencies.
069     *
070     * @param missingDependencies the missing dependencies
071     * @return the pull input dependencies
072     */
073    public static CoordInputDependency getPullInputDependencies(StringBlob missingDependencies) {
074        if (missingDependencies == null) {
075            return new CoordOldInputDependency();
076        }
077        return getPullInputDependencies(missingDependencies.getString());
078    }
079
080    public static CoordInputDependency getPullInputDependencies(String dependencies) {
081
082        if (StringUtils.isEmpty(dependencies) || !hasInputLogic(dependencies)) {
083            return new CoordOldInputDependency(dependencies);
084        }
085        else
086            try {
087                return WritableUtils.fromByteArray(getDependenciesWithoutMagicNumber(dependencies).getBytes(CHAR_ENCODING),
088                        CoordPullInputDependency.class);
089            }
090            catch (UnsupportedEncodingException e) {
091                throw new RuntimeException(e);
092            }
093    }
094
095    /**
096     * Gets the push input dependencies.
097     *
098     * @param pushMissingDependencies the push missing dependencies
099     * @return the push input dependencies
100     */
101    public static CoordInputDependency getPushInputDependencies(StringBlob pushMissingDependencies) {
102
103        if (pushMissingDependencies == null) {
104            return new CoordOldInputDependency();
105        }
106        return getPushInputDependencies(pushMissingDependencies.getString());
107
108    }
109
110    public static CoordInputDependency getPushInputDependencies(String dependencies) {
111
112        if (StringUtils.isEmpty(dependencies) || !hasInputLogic(dependencies)) {
113            return new CoordOldInputDependency(dependencies);
114        }
115
116        else {
117            try {
118                return WritableUtils.fromByteArray(getDependenciesWithoutMagicNumber(dependencies).getBytes(CHAR_ENCODING),
119                        CoordPushInputDependency.class);
120            }
121            catch (UnsupportedEncodingException e) {
122                throw new RuntimeException(e);
123            }
124
125        }
126    }
127
128    /**
129     * Checks if input logic is enable.
130     *
131     * @param dependencies the dependencies
132     * @return true, if is input logic enable
133     */
134    private static boolean hasInputLogic(String dependencies) {
135        return dependencies.startsWith(getMagicNumber());
136    }
137
138    /**
139     * Gets the magic number.
140     *
141     * @return the magic number
142     */
143    public static String getMagicNumber() {
144        try {
145            return new String(MAGIC_NUMBER, CHAR_ENCODING);
146        }
147        catch (UnsupportedEncodingException e) {
148            throw new RuntimeException(e.getMessage());
149        }
150    }
151
152    /**
153     * Gets the dependencies without magic number.
154     *
155     * @param dependencies the dependencies
156     * @return the dependencies without magic number
157     */
158    public static String getDependenciesWithoutMagicNumber(String dependencies) {
159        return dependencies.substring(getMagicNumber().length());
160    }
161
162}