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

import java.text.MessageFormat;
import java.util.Map;

import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import acromusashi.stream.bolt.BaseConfigurationBolt;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Tuple;

/**
 * ElasticSearchに対してクエリを投入するBolt
 * 
 * @author kimura
 */
public class ElasticSearchBolt extends BaseConfigurationBolt
{
    /** serialVersionUID */
    private static final long   serialVersionUID  = 4987555107871741041L;

    /** logger */
    private static final Logger logger            = LoggerFactory.getLogger(ElasticSearchBolt.class);

    /** デフォルトポート値 */
    private static final int    DEFAULT_PORT      = 9300;

    /** クラスタ名称 */
    protected String            clusterName;

    /** ElasticSearchの投入先。「host1:port1;host2:port2;host3:port3...」という形式で定義 */
    protected String            servers;

    /** TupleをIndex Requestに変換するコンバータ */
    protected EsTupleConverter  converter;

    /** ElasticSearchClient */
    protected transient Client  client;

    /**
     * Converterを指定してインスタンスを生成する。
     * 
     * @param converter TupleをIndex Requestに変換するコンバータ
     */
    public ElasticSearchBolt(EsTupleConverter converter)
    {
        this.converter = converter;
    }

    /**
     * {@inheritDoc}
     */
    @SuppressWarnings("rawtypes")
    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector)
    {
        super.prepare(stormConf, context, collector);
        // ElasticSearchClientを初期化
        // Serversは「host1:port1;host2:port2;host3:port3...」形式のため分割して生成
        Settings settings = ImmutableSettings.settingsBuilder().put("cluster.name",
                this.clusterName).build();
        TransportClient transportClient = new TransportClient(settings);
        for (String server : this.servers.split(";"))
        {
            String[] components = server.trim().split(":");
            String host = components[0];
            int port = DEFAULT_PORT;
            if (components.length > 1)
            {
                port = Integer.parseInt(components[1]);
            }

            transportClient = transportClient.addTransportAddress(new InetSocketTransportAddress(
                    host, port));
        }

        this.client = transportClient;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void execute(Tuple input)
    {
        String documentId = null;
        String indexName = null;
        String typeName = null;
        String document = null;

        try
        {
            documentId = this.converter.convertToId(input);
            indexName = this.converter.convertToIndex(input);
            typeName = this.converter.convertToType(input);
            document = this.converter.convertToDocument(input);

            IndexResponse response = this.client.prepareIndex(indexName, typeName, documentId).setSource(
                    document).execute().actionGet();

            if (logger.isDebugEnabled() == true)
            {
                String logFormat = "Document Indexed. Id={0}, Type={1}, Index={2}, Version={3}";
                logger.debug(MessageFormat.format(logFormat, response.getId(), typeName, indexName,
                        response.getVersion()));
            }
        }
        catch (Exception ex)
        {
            String logFormat = "Document Index failed. Dispose Tuple. Id={0}, Type={1}, Index={2}";
            logger.warn(MessageFormat.format(logFormat, documentId, typeName, indexName), ex);
        }

        getCollector().ack(input);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer)
    {
        // 下流に送信を行わないため、未設定
    }

    /**
     * @param clusterName the clusterName to set
     */
    public void setClusterName(String clusterName)
    {
        this.clusterName = clusterName;
    }

    /**
     * @param servers the servers to set
     */
    public void setServers(String servers)
    {
        this.servers = servers;
    }
}

@@ -1333,17 +1333,16 @@
 %E3%82%8BBolt%0A *
- 
 %0A * @aut
@@ -2112,17 +2112,16 @@
 %E3%80%82%0A     *
- 
 %0A     * 
@@ -2340,16 +2340,17 @@
 arnings(
+%7B
 %22rawtype
@@ -2351,16 +2351,29 @@
 awtypes%22
+, %22resource%22%7D
 )%0A    @O
@@ -2471,22 +2471,111 @@
 lector)%0A
-
     %7B%0A
+        // cleanup%E3%83%A1%E3%82%BD%E3%83%83%E3%83%89%E3%81%AF%E5%91%BC%E3%81%B0%E3%82%8C%E3%81%AA%E3%81%84%E3%81%9F%E3%82%81%E3%80%81%E3%83%AA%E3%82%BD%E3%83%BC%E3%82%B9%E3%82%AF%E3%83%AD%E3%83%BC%E3%82%BA%E3%81%AF%E3%81%A7%E3%81%8D%E3%81%AA%E3%81%84%E3%80%82%0A        // %E3%81%A0%E3%81%8C%E3%80%811%E3%82%A4%E3%83%B3%E3%82%B9%E3%82%BF%E3%83%B3%E3%82%B9%E3%81%AE%E3%81%BF%E3%81%97%E3%81%8B%E7%94%9F%E6%88%90%E3%81%95%E3%82%8C%E3%81%AA%E3%81%84%E3%81%9F%E3%82%81%E3%80%81%E3%83%AA%E3%83%BC%E3%82%AF%E3%81%AF%E3%81%97%E3%81%AA%E3%81%84%E3%80%82%0A
         

Add resource comment