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

import org.apache.storm.opentsdb.OpenTsdbMetricDatapoint;
import org.apache.storm.tuple.ITuple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

/**
 * Converts {@link org.apache.storm.tuple.ITuple} to {@link OpenTsdbMetricDatapoint}
 */
public final class TupleOpenTsdbDatapointMapper implements ITupleOpenTsdbDatapointMapper {
    private static final Logger LOG = LoggerFactory.getLogger(TupleOpenTsdbDatapointMapper.class);

    /**
     * Default mapper which can be used when the tuple already contains fields mapping  metric, timestamp, tags and value.
     */
    public static final TupleOpenTsdbDatapointMapper DEFAULT_MAPPER =
            new TupleOpenTsdbDatapointMapper("metric", "timestamp", "tags", "value");

    private final String metricField;
    private final String timestampField;
    private final String valueField;
    private final String tagsField;

    public TupleOpenTsdbDatapointMapper(String metricField, String timestampField, String tagsField, String valueField) {
        this.metricField = metricField;
        this.timestampField = timestampField;
        this.tagsField = tagsField;
        this.valueField = valueField;
    }

    @Override
    public OpenTsdbMetricDatapoint getMetricPoint(ITuple tuple) {
            return new OpenTsdbMetricDatapoint(
                    tuple.getStringByField(metricField),
                    (Map<String, String>) tuple.getValueByField(tagsField),
                    tuple.getLongByField(timestampField),
                    (Number) tuple.getValueByField(valueField));
    }

    /**
     * @return metric field name in the tuple.
     */
    public String getMetricField() {
        return metricField;
    }

    /**
     * @return timestamp field name in the tuple.
     */
    public String getTimestampField() {
        return timestampField;
    }

    /**
     * @return value field name in the tuple.
     */
    public String getValueField() {
        return valueField;
    }

    /**
     * @return tags field name in the tuple
     */
    public String getTagsField() {
        return tagsField;
    }

    public boolean equals(Object o) {
        if (this == o) return true;
        if (!(o instanceof TupleOpenTsdbDatapointMapper)) return false;

        TupleOpenTsdbDatapointMapper that = (TupleOpenTsdbDatapointMapper) o;

        if (!metricField.equals(that.metricField)) return false;
        if (!timestampField.equals(that.timestampField)) return false;
        if (!valueField.equals(that.valueField)) return false;
        return tagsField.equals(that.tagsField);

    }

    @Override
    public int hashCode() {
        int result = metricField.hashCode();
        result = 31 * result + timestampField.hashCode();
        result = 31 * result + valueField.hashCode();
        result = 31 * result + tagsField.hashCode();
        return result;
    }

    @Override
    public String toString() {
        return "TupleOpenTsdbDatapointMapper{" +
                "metricField='" + metricField + '\'' +
                ", timestampField='" + timestampField + '\'' +
                ", valueField='" + valueField + '\'' +
                ", tagsField='" + tagsField + '\'' +
                '}';
    }
}
