/**
 * TODO
 * 
 * @author smartloli.
 *
 *         Created by Mar 9, 2018
 */
public class KafkaSqlFactory {

	private static final Logger LOG = LoggerFactory.getLogger(KafkaSqlFactory.class);

	private static Ignite ignite = null;

	private static void getInstance() {
		if (ignite == null) {
			ignite = Ignition.start();
		}
	}

	private static IgniteCache<Long, TopicX> processor(List<TopicX> collectors) {
		getInstance();
		CacheConfiguration<Long, TopicX> topicDataCacheCfg = new CacheConfiguration<Long, TopicX>();
		topicDataCacheCfg.setName(TopicCache.NAME);
		topicDataCacheCfg.setCacheMode(CacheMode.PARTITIONED);
		topicDataCacheCfg.setIndexedTypes(Long.class, TopicX.class);
		IgniteCache<Long, TopicX> topicDataCache = ignite.getOrCreateCache(topicDataCacheCfg);
		for (TopicX topic : collectors) {
			topicDataCache.put(topic.getOffsets(), topic);
		}
		return topicDataCache;
	}

	public static String sql(String sql, List<TopicX> collectors) {
		try {
			IgniteCache<Long, TopicX> topicDataCache = processor(collectors);
			SqlFieldsQuery qry = new SqlFieldsQuery(sql);
			QueryCursor<List<?>> cursor = topicDataCache.query(qry);
			for (List<?> row : cursor) {
				System.out.println(row.toString());
			}
		} catch (Exception ex) {
			LOG.error("Query kafka topic has error, msg is " + ex.getMessage());
		} finally {
			close();
		}
		return "";
	}

	private static void close() {
		try {
			if (ignite != null) {
				ignite.close();
			}
		} catch (Exception ex) {
			LOG.error("Close Ignite has error, msg is " + ex.getMessage());
		} finally {
			if (ignite != null) {
				ignite.close();
			}
		}
	}

}

public static void ignite(){
		List<TopicX> collectors = new ArrayList<>();
		int count = 0;
		for (int i = 0; i < 10; i++) {
			TopicX td = new TopicX();
			if (count > 3) {
				count = 0;
			}
			td.setPartitionId(count);
			td.setOffsets(i);
			td.setMessage("hello_" + i);
			td.setTopicName("test");
			collectors.add(td);
			count++;
		}

		String sql = "select offsets,message from TopicX where offsets>6 and partitionId in (0,1) limit 1";
		long stime = System.currentTimeMillis();
		KafkaSqlFactory.sql(sql, collectors);
		System.out.println("Cost time [" + (System.currentTimeMillis() - stime) / 1000.0 + "]ms");
	}