package com.ctrip.hermes.metaservice.service;

import java.util.Date;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.unidal.dal.jdbc.DalException;
import org.unidal.dal.jdbc.DalNotFoundException;
import org.unidal.lookup.annotation.Inject;
import org.unidal.lookup.annotation.Named;

import com.alibaba.fastjson.JSON;
import com.ctrip.hermes.meta.entity.Meta;
import com.ctrip.hermes.metaservice.model.MetaDao;
import com.ctrip.hermes.metaservice.model.MetaEntity;

/**
 * @author Leo Liang(jhliang@ctrip.com)
 *
 */
@Named(type = MetaService.class)
public class DefaultMetaService implements MetaService {
	protected static final Logger m_logger = LoggerFactory.getLogger(DefaultMetaService.class);

	@Inject
	protected MetaDao m_metaDao;

	@Inject
	protected ZookeeperService m_zookeeperService;

	@Override
	public Meta findLatestMeta() throws DalException {
		com.ctrip.hermes.metaservice.model.Meta dalMeta = null;
		try {
			dalMeta = m_metaDao.findLatest(MetaEntity.READSET_FULL);
		} catch (DalNotFoundException e) {
			// ignore if no record
		}
		if (dalMeta == null) {
			dalMeta = new com.ctrip.hermes.metaservice.model.Meta();
			dalMeta.setValue(JSON.toJSONString(new Meta()));
		}
		return JSON.parseObject(dalMeta.getValue(), Meta.class);
	}

	@Override
	public synchronized boolean updateMeta(Meta meta) throws DalException {
		Meta latest = findLatestMeta();
		if (!latest.getVersion().equals(meta.getVersion())) {
			String e = String.format("Outdated Version. Latest: %s, Offered: %s", latest.getVersion(), meta.getVersion());
			throw new RuntimeException(e);
		}

		com.ctrip.hermes.metaservice.model.Meta dalMeta = new com.ctrip.hermes.metaservice.model.Meta();
		try {
			meta.setVersion(meta.getVersion() + 1);
			dalMeta.setValue(JSON.toJSONString(meta));
			dalMeta.setDataChangeLastTime(new Date(System.currentTimeMillis()));
			m_metaDao.insert(dalMeta);
			m_zookeeperService.updateZkBaseMetaVersion(meta.getVersion());
		} catch (Exception e) {
			m_logger.warn("Update meta failed", e);
			throw new RuntimeException("Update meta failed.", e);
		}
		return true;
	}

}

@@ -375,16 +375,61 @@
 y.Meta;%0A
+import com.ctrip.hermes.meta.entity.Storage;%0A
 import c
@@ -928,85 +928,41 @@
 %7B%0A%09%09
-com.ctrip.hermes.metaservice.model.Meta dalMeta = null;%0A%09%09try %7B%0A%09%09%09dalMeta = 
+try %7B%0A%09%09%09return JSON.parseObject(
 m_me
@@ -1002,16 +1002,40 @@
 ET_FULL)
+.getValue(), Meta.class)
 ;%0A%09%09%7D ca
@@ -1072,234 +1072,123 @@
 %0A%09%09%09
-// ignore if no record%0A%09%09%7D%0A%09%09if (dalMeta == null) %7B%0A%09%09%09dalMeta = new com.ctrip.hermes.metaservice.model.Meta();%0A%09%09%09dalMeta.setValue(JSON.toJSONString(new Meta()));%0A%09%09%7D%0A%09%09return JSON.parseObject(dalMeta.getValue(), Meta.class);
+return new Meta().addStorage(new Storage(Storage.MYSQL)).addStorage(new Storage(Storage.KAFKA)).setVersion(0L);%0A%09%09%7D
 %0A%09%7D%0A

return an empty meta with empty storage when catch dal-not-found