// Path: src/main/java/acromusashi/stream/constants/FieldName.java
// public interface FieldName
// {
//     /** Message grouping key's field name. */
//     String MESSAGE_KEY   = "messageKey";
// 
//     /** Message value's field name. */
//     String MESSAGE_VALUE = "messageValue";
// }
// 
// Path: src/main/java/acromusashi/stream/entity/StreamMessage.java
// public class StreamMessage implements Serializable
// {
//     /** serialVersionUID */
//     private static final long   serialVersionUID = -7553630425199269093L;
// 
//     /** Message Header */
//     private StreamMessageHeader header;
// 
//     /** Message Body */
//     private Object              body;
// 
//     /**
//      * Constructs instance.
//      */
//     public StreamMessage()
//     {
//         this.header = new StreamMessageHeader();
//     }
// 
//     /**
//      * Constructs instance with Header and body.
//      *
//      * @param header Message Header
//      * @param body Message Body
//      */
//     public StreamMessage(StreamMessageHeader header, Object body)
//     {
//         this.header = header;
//         this.body = body;
//     }
// 
//     /**
//      * @return the header
//      */
//     public StreamMessageHeader getHeader()
//     {
//         return this.header;
//     }
// 
//     /**
//      * @param header the header to set
//      */
//     public void setHeader(StreamMessageHeader header)
//     {
//         this.header = header;
//     }
// 
//     /**
//      * @return the body
//      */
//     public Object getBody()
//     {
//         return this.body;
//     }
// 
//     /**
//      * @param body the body to set
//      */
//     public void setBody(Object body)
//     {
//         this.body = body;
//     }
// 
//     /**
//      * Add field to message body.
//      *
//      * @param key key
//      * @param value value
//      */
//     @SuppressWarnings({"unchecked", "rawtypes"})
//     public void addField(String key, Object value)
//     {
//         if (this.body == null || Map.class.isAssignableFrom(this.body.getClass()) == false)
//         {
//             this.body = Maps.newLinkedHashMap();
//         }
// 
//         ((Map) this.body).put(key, value);
//     }
// 
//     /**
//      * Get field from message body.
//      *
//      * @param key key
//      * @return field mapped key
//      */
//     @SuppressWarnings("rawtypes")
//     public Object getField(String key)
//     {
//         if (this.body == null || Map.class.isAssignableFrom(this.body.getClass()) == false)
//         {
//             return null;
//         }
// 
//         return ((Map) this.body).get(key);
//     }
// 
//     /**
//      * {@inheritDoc}
//      */
//     @Override
//     public String toString()
//     {
//         String result = this.header.toString() + "," + this.body.toString();
//         return result;
//     }
// }


import java.util.Map;
import acromusashi.stream.constants.FieldName;
import acromusashi.stream.entity.StreamMessage;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Tuple;
/**
* Copyright (c) Acroquest Technology Co, Ltd. All Rights Reserved.
* Please read the associated COPYRIGHTS file for more details.
*
* THE SOFTWARE IS PROVIDED BY Acroquest Technolog Co., Ltd.,
* WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING
* BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
* IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDER BE LIABLE FOR ANY
* CLAIM, DAMAGES SUFFERED BY LICENSEE AS A RESULT OF USING, MODIFYING
* OR DISTRIBUTING THIS SOFTWARE OR ITS DERIVATIVES.
*/
package acromusashi.stream.bolt;



/**
 * BaseBolt, this class has the following values ​​in the field, <br>
 * and auto extract {@link acromusashi.stream.entity.StreamMessage} from {@link backtype.storm.tuple.Tuple}.
 *
 * <ol>
 * <li>Storm configuration</li>
 * <li>Topology context</li>
 * <li>SpoutOutputCollector</li>
 * </ol>
 */
public abstract class AmConfigurationBolt extends BaseRichBolt
{
    /** serialVersionUID */
    private static final long serialVersionUID = 103046340453102132L;

    /** StormConfig */
    @SuppressWarnings("rawtypes")
    private Map               stormConf;

    /** TopologyContext */
    private TopologyContext   context;

    /** OutputCollector */
    private OutputCollector   collector;

    /** Executing Tuple */
    private Tuple             executingTuple;

    /**
     * {@inheritDoc}
     */
    @SuppressWarnings("rawtypes")
    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector)
    {
        this.stormConf = stormConf;
        this.context = context;
        this.collector = collector;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void execute(Tuple input)
    {
        // extract message from Tuple
        this.executingTuple = input;

// Path: src/main/java/acromusashi/stream/constants/FieldName.java
// public interface FieldName
// {
//     /** Message grouping key's field name. */
//     String MESSAGE_KEY   = "messageKey";
// 
//     /** Message value's field name. */
//     String MESSAGE_VALUE = "messageValue";
// }
// 
// Path: src/main/java/acromusashi/stream/entity/StreamMessage.java
// public class StreamMessage implements Serializable
// {
//     /** serialVersionUID */
//     private static final long   serialVersionUID = -7553630425199269093L;
// 
//     /** Message Header */
//     private StreamMessageHeader header;
// 
//     /** Message Body */
//     private Object              body;
// 
//     /**
//      * Constructs instance.
//      */
//     public StreamMessage()
//     {
//         this.header = new StreamMessageHeader();
//     }
// 
//     /**
//      * Constructs instance with Header and body.
//      *
//      * @param header Message Header
//      * @param body Message Body
//      */
//     public StreamMessage(StreamMessageHeader header, Object body)
//     {
//         this.header = header;
//         this.body = body;
//     }
// 
//     /**
//      * @return the header
//      */
//     public StreamMessageHeader getHeader()
//     {
//         return this.header;
//     }
// 
//     /**
//      * @param header the header to set
//      */
//     public void setHeader(StreamMessageHeader header)
//     {
//         this.header = header;
//     }
// 
//     /**
//      * @return the body
//      */
//     public Object getBody()
//     {
//         return this.body;
//     }
// 
//     /**
//      * @param body the body to set
//      */
//     public void setBody(Object body)
//     {
//         this.body = body;
//     }
// 
//     /**
//      * Add field to message body.
//      *
//      * @param key key
//      * @param value value
//      */
//     @SuppressWarnings({"unchecked", "rawtypes"})
//     public void addField(String key, Object value)
//     {
//         if (this.body == null || Map.class.isAssignableFrom(this.body.getClass()) == false)
//         {
//             this.body = Maps.newLinkedHashMap();
//         }
// 
//         ((Map) this.body).put(key, value);
//     }
// 
//     /**
//      * Get field from message body.
//      *
//      * @param key key
//      * @return field mapped key
//      */
//     @SuppressWarnings("rawtypes")
//     public Object getField(String key)
//     {
//         if (this.body == null || Map.class.isAssignableFrom(this.body.getClass()) == false)
//         {
//             return null;
//         }
// 
//         return ((Map) this.body).get(key);
//     }
// 
//     /**
//      * {@inheritDoc}
//      */
//     @Override
//     public String toString()
//     {
//         String result = this.header.toString() + "," + this.body.toString();
//         return result;
//     }
// }


// Path: src/main/java/acromusashi/stream/bolt/AmConfigurationBolt.java
import java.util.Map;
import acromusashi.stream.constants.FieldName;
import acromusashi.stream.entity.StreamMessage;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Tuple;

/**
* Copyright (c) Acroquest Technology Co, Ltd. All Rights Reserved.
* Please read the associated COPYRIGHTS file for more details.
*
* THE SOFTWARE IS PROVIDED BY Acroquest Technolog Co., Ltd.,
* WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING
* BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
* IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDER BE LIABLE FOR ANY
* CLAIM, DAMAGES SUFFERED BY LICENSEE AS A RESULT OF USING, MODIFYING
* OR DISTRIBUTING THIS SOFTWARE OR ITS DERIVATIVES.
*/
package acromusashi.stream.bolt;



/**
 * BaseBolt, this class has the following values ​​in the field, <br>
 * and auto extract {@link acromusashi.stream.entity.StreamMessage} from {@link backtype.storm.tuple.Tuple}.
 *
 * <ol>
 * <li>Storm configuration</li>
 * <li>Topology context</li>
 * <li>SpoutOutputCollector</li>
 * </ol>
 */
public abstract class AmConfigurationBolt extends BaseRichBolt
{
    /** serialVersionUID */
    private static final long serialVersionUID = 103046340453102132L;

    /** StormConfig */
    @SuppressWarnings("rawtypes")
    private Map               stormConf;

    /** TopologyContext */
    private TopologyContext   context;

    /** OutputCollector */
    private OutputCollector   collector;

    /** Executing Tuple */
    private Tuple             executingTuple;

    /**
     * {@inheritDoc}
     */
    @SuppressWarnings("rawtypes")
    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector)
    {
        this.stormConf = stormConf;
        this.context = context;
        this.collector = collector;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void execute(Tuple input)
    {
        // extract message from Tuple
        this.executingTuple = input;

        StreamMessage message = null;