/*
 * 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.storm;

import java.io.Serializable;
import java.lang.reflect.Constructor;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import org.apache.storm.generated.Bolt;
import org.apache.storm.generated.ComponentCommon;
import org.apache.storm.generated.ComponentObject;
import org.apache.storm.generated.GlobalStreamId;
import org.apache.storm.generated.Grouping;
import org.apache.storm.generated.JavaObject;
import org.apache.storm.generated.JavaObjectArg;
import org.apache.storm.generated.NullStruct;
import org.apache.storm.generated.SpoutSpec;
import org.apache.storm.generated.StateSpoutSpec;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.generated.StormTopology._Fields;
import org.apache.storm.generated.StreamInfo;
import org.apache.storm.shade.org.json.simple.JSONValue;
import org.apache.storm.task.IBolt;
import org.apache.storm.topology.BoltDeclarer;
import org.apache.storm.topology.IBasicBolt;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.IRichSpout;
import org.apache.storm.topology.SpoutDeclarer;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class Thrift {
    private static Logger LOG = LoggerFactory.getLogger(Thrift.class);

    private static StormTopology._Fields[] STORM_TOPOLOGY_FIELDS = null;
    private static StormTopology._Fields[] SPOUT_FIELDS =
        { StormTopology._Fields.SPOUTS, StormTopology._Fields.STATE_SPOUTS };

    static {
        Set<_Fields> keys = StormTopology.metaDataMap.keySet();
        keys.toArray(STORM_TOPOLOGY_FIELDS = new StormTopology._Fields[keys.size()]);
    }

    public static StormTopology._Fields[] getTopologyFields() {
        return STORM_TOPOLOGY_FIELDS;
    }

    public static StormTopology._Fields[] getSpoutFields() {
        return SPOUT_FIELDS;
    }

    public static StreamInfo directOutputFields(List<String> fields) {
        return new StreamInfo(fields, true);
    }

    public static StreamInfo outputFields(List<String> fields) {
        return new StreamInfo(fields, false);
    }

    public static Grouping prepareShuffleGrouping() {
        return Grouping.shuffle(new NullStruct());
    }

    public static Grouping prepareLocalOrShuffleGrouping() {
        return Grouping.local_or_shuffle(new NullStruct());
    }

    public static Grouping prepareFieldsGrouping(List<String> fields) {
        return Grouping.fields(fields);
    }

    public static Grouping prepareGlobalGrouping() {
        return prepareFieldsGrouping(new ArrayList<String>());
    }

    public static Grouping prepareDirectGrouping() {
        return Grouping.direct(new NullStruct());
    }

    public static Grouping prepareAllGrouping() {
        return Grouping.all(new NullStruct());
    }

    public static Grouping prepareNoneGrouping() {
        return Grouping.none(new NullStruct());
    }

    public static Grouping prepareCustomStreamGrouping(Object obj) {
        return Grouping.custom_serialized(Utils.javaSerialize(obj));
    }

    public static Grouping prepareCustomJavaObjectGrouping(JavaObject obj) {
        return Grouping.custom_object(obj);
    }

    public static Object instantiateJavaObject(JavaObject obj) {

        List<JavaObjectArg> args = obj.get_args_list();
        Class[] paraTypes = new Class[args.size()];
        Object[] paraValues = new Object[args.size()];
        for (int i = 0; i < args.size(); i++) {
            JavaObjectArg arg = args.get(i);
            paraValues[i] = arg.getFieldValue();

            if (arg.getSetField().equals(JavaObjectArg._Fields.INT_ARG)) {
                paraTypes[i] = Integer.class;
            } else if (arg.getSetField().equals(JavaObjectArg._Fields.LONG_ARG)) {
                paraTypes[i] = Long.class;
            } else if (arg.getSetField().equals(JavaObjectArg._Fields.STRING_ARG)) {
                paraTypes[i] = String.class;
            } else if (arg.getSetField().equals(JavaObjectArg._Fields.BOOL_ARG)) {
                paraTypes[i] = Boolean.class;
            } else if (arg.getSetField().equals(JavaObjectArg._Fields.BINARY_ARG)) {
                paraTypes[i] = ByteBuffer.class;
            } else if (arg.getSetField().equals(JavaObjectArg._Fields.DOUBLE_ARG)) {
                paraTypes[i] = Double.class;
            } else {
                paraTypes[i] = Object.class;
            }
        }

        try {
            Class clazz = Class.forName(obj.get_full_class_name());
            Constructor cons = clazz.getConstructor(paraTypes);
            return cons.newInstance(paraValues);
        } catch (Exception e) {
            LOG.error("java object instantiation failed", e);
        }

        return null;

    }

    public static Grouping._Fields groupingType(Grouping grouping) {
        return grouping.getSetField();
    }

    public static List<String> fieldGrouping(Grouping grouping) {
        if (!Grouping._Fields.FIELDS.equals(groupingType(grouping))) {
            throw new IllegalArgumentException("Tried to get grouping fields from non fields grouping");
        }
        return grouping.get_fields();
    }

    public static boolean isGlobalGrouping(Grouping grouping) {
        if (Grouping._Fields.FIELDS.equals(groupingType(grouping))) {
            return fieldGrouping(grouping).isEmpty();
        }

        return false;
    }

    public static int getParallelismHint(ComponentCommon componentCommon) {
        if (!componentCommon.is_set_parallelism_hint()) {
            return 1;
        } else {
            return componentCommon.get_parallelism_hint();
        }
    }

    public static ComponentObject serializeComponentObject(Object obj) {
        return ComponentObject.serialized_java(Utils.javaSerialize(obj));
    }

    public static Object deserializeComponentObject(ComponentObject obj) {
        if (obj.getSetField() != ComponentObject._Fields.SERIALIZED_JAVA) {
            throw new RuntimeException("Cannot deserialize non-java-serialized object");
        }
        return Utils.javaDeserialize(obj.get_serialized_java(), Serializable.class);
    }

    public static ComponentCommon prepareComponentCommon(Map<GlobalStreamId, Grouping> inputs, Map<String,
        StreamInfo> outputs, Integer parallelismHint) {
        return prepareComponentCommon(inputs, outputs, parallelismHint, null);
    }

    public static ComponentCommon prepareComponentCommon(Map<GlobalStreamId, Grouping> inputs, Map<String, StreamInfo> outputs,
                                                         Integer parallelismHint, Map<String, Object> conf) {
        Map<GlobalStreamId, Grouping> mappedInputs = new HashMap<>();
        Map<String, StreamInfo> mappedOutputs = new HashMap<>();
        if (inputs != null && !inputs.isEmpty()) {
            mappedInputs.putAll(inputs);
        }
        if (outputs != null && !outputs.isEmpty()) {
            mappedOutputs.putAll(outputs);
        }
        ComponentCommon component = new ComponentCommon(mappedInputs, mappedOutputs);
        if (parallelismHint != null) {
            component.set_parallelism_hint(parallelismHint);
        }
        if (conf != null) {
            component.set_json_conf(JSONValue.toJSONString(conf));
        }
        return component;
    }

    public static SpoutSpec prepareSerializedSpoutDetails(IRichSpout spout, Map<String, StreamInfo> outputs) {
        return new SpoutSpec(ComponentObject.serialized_java(Utils.javaSerialize(spout)),
                prepareComponentCommon(new HashMap<>(), outputs, null, null));
    }

    public static Bolt prepareSerializedBoltDetails(Map<GlobalStreamId, Grouping> inputs, IBolt bolt, Map<String, StreamInfo> outputs,
                                                    Integer parallelismHint, Map<String, Object> conf) {
        ComponentCommon common = prepareComponentCommon(inputs, outputs, parallelismHint, conf);
        return new Bolt(ComponentObject.serialized_java(Utils.javaSerialize(bolt)), common);
    }

    public static BoltDetails prepareBoltDetails(Map<GlobalStreamId, Grouping> inputs, Object bolt) {
        return prepareBoltDetails(inputs, bolt, null, null);
    }

    public static BoltDetails prepareBoltDetails(Map<GlobalStreamId, Grouping> inputs, Object bolt,
                                                 Integer parallelismHint) {
        return prepareBoltDetails(inputs, bolt, parallelismHint, null);
    }

    public static BoltDetails prepareBoltDetails(Map<GlobalStreamId, Grouping> inputs, Object bolt,
                                                 Integer parallelismHint, Map<String, Object> conf) {
        BoltDetails details = new BoltDetails(bolt, conf, parallelismHint, inputs);
        return details;
    }

    public static SpoutDetails prepareSpoutDetails(IRichSpout spout) {
        return prepareSpoutDetails(spout, null, null);
    }

    public static SpoutDetails prepareSpoutDetails(IRichSpout spout, Integer parallelismHint) {
        return prepareSpoutDetails(spout, parallelismHint, null);
    }

    public static SpoutDetails prepareSpoutDetails(IRichSpout spout, Integer parallelismHint, Map<String, Object> conf) {
        SpoutDetails details = new SpoutDetails(spout, parallelismHint, conf);
        return details;
    }

    private static void addInputs(BoltDeclarer declarer, Map<GlobalStreamId, Grouping> inputs) {
        for (Entry<GlobalStreamId, Grouping> entry : inputs.entrySet()) {
            declarer.grouping(entry.getKey(), entry.getValue());
        }
    }

    public static StormTopology buildTopology(HashMap<String, SpoutDetails> spoutMap,
                                              HashMap<String, BoltDetails> boltMap, HashMap<String, StateSpoutSpec> stateMap) {
        return buildTopology(spoutMap, boltMap);
    }

    public static StormTopology buildTopology(Map<String, SpoutDetails> spoutMap, Map<String, BoltDetails> boltMap) {
        TopologyBuilder builder = new TopologyBuilder();
        for (Entry<String, SpoutDetails> entry : spoutMap.entrySet()) {
            String spoutId = entry.getKey();
            SpoutDetails spec = entry.getValue();
            SpoutDeclarer spoutDeclarer = builder.setSpout(spoutId, spec.getSpout(), spec.getParallelism());
            spoutDeclarer.addConfigurations(spec.getConf());
        }
        for (Entry<String, BoltDetails> entry : boltMap.entrySet()) {
            String spoutId = entry.getKey();
            BoltDetails spec = entry.getValue();
            BoltDeclarer boltDeclarer = null;
            if (spec.bolt instanceof IRichBolt) {
                boltDeclarer = builder.setBolt(spoutId, (IRichBolt) spec.getBolt(), spec.getParallelism());
            } else {
                boltDeclarer = builder.setBolt(spoutId, (IBasicBolt) spec.getBolt(), spec.getParallelism());
            }
            boltDeclarer.addConfigurations(spec.getConf());
            addInputs(boltDeclarer, spec.getInputs());
        }
        return builder.createTopology();
    }

    public static class SpoutDetails {
        private IRichSpout spout;
        private Integer parallelism;
        private Map<String, Object> conf;

        public SpoutDetails(IRichSpout spout, Integer parallelism, Map<String, Object> conf) {
            this.spout = spout;
            this.parallelism = parallelism;
            this.conf = conf;
        }

        public IRichSpout getSpout() {
            return spout;
        }

        public Integer getParallelism() {
            return parallelism;
        }

        public Map<String, Object> getConf() {
            return conf;
        }
    }

    public static class BoltDetails {
        private Object bolt;
        private Map<String, Object> conf;
        private Integer parallelism;
        private Map<GlobalStreamId, Grouping> inputs;

        public BoltDetails(Object bolt, Map<String, Object> conf, Integer parallelism,
                           Map<GlobalStreamId, Grouping> inputs) {
            this.bolt = bolt;
            this.conf = conf;
            this.parallelism = parallelism;
            this.inputs = inputs;
        }

        public Object getBolt() {
            return bolt;
        }

        public Map<String, Object> getConf() {
            return conf;
        }

        public Map<GlobalStreamId, Grouping> getInputs() {
            return inputs;
        }

        public Integer getParallelism() {
            return parallelism;
        }
    }
}