package org.onetwo.ext.ons.transaction;

import java.util.Collection;
import java.util.Collections;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

import org.onetwo.common.db.spi.BaseEntityManager;
import org.onetwo.common.utils.LangUtils;
import org.onetwo.common.utils.StringUtils;
import org.onetwo.dbm.id.SnowflakeIdGenerator;
import org.onetwo.ext.ons.ONSUtils;
import org.onetwo.ext.ons.producer.SendMessageContext;
import org.onetwo.ext.ons.transaction.SendMessageEntity.SendStates;
import org.slf4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionSynchronizationManager;

import com.aliyun.openservices.ons.api.Message;

/**
 * @author wayshall
 * <br/>
 */
@Transactional
public class DbmSendMessageRepository implements SendMessageRepository {
	
	private Logger log = ONSUtils.getONSLogger();
	
	private SnowflakeIdGenerator idGenerator = new SnowflakeIdGenerator(30);
	@Autowired
	private MessageBodyStoreSerializer messageBodyStoreSerializer;
	@Autowired
	private BaseEntityManager baseEntityManager;
	
//	private NamedThreadLocal<Set<SendMessageContext>> messageStorer = new NamedThreadLocal<>("rmq message thread storer");
	
	@Override
	public void save(SendMessageContext ctx){
		Message message = ctx.getMessage();
		SendMessageEntity send = new SendMessageEntity();
		String key = message.getKey();
		if(StringUtils.isBlank(key)){
			key = String.valueOf(idGenerator.nextId());
		}
		send.setKey(key);
		send.setState(SendStates.TO_SEND);
		send.setBody(messageBodyStoreSerializer.serialize(message));
		
		baseEntityManager.persist(send);

		ctx.setMessageEntity(send);
		storeInCurrentContext(ctx);
	}

	@Override
	public void remove(Collection<SendMessageContext> msgCtxs){
		boolean debug = msgCtxs.iterator().next().isDebug();
		List<String> keys = getSendMessageKeys(msgCtxs);
		baseEntityManager.removeByIds(SendMessageEntity.class, keys.toArray(new String[0]));
		if(debug && log.isInfoEnabled()){
			log.info("remove message data from database: {}", keys);
		}
	}
	
	@SuppressWarnings("unchecked")
	protected void storeInCurrentContext(SendMessageContext ctx){
		boolean debug = ctx.isDebug();
		Set<SendMessageContext> msgCtxs = (Set<SendMessageContext>)TransactionSynchronizationManager.unbindResourceIfPossible(this);
		if(debug && log.isInfoEnabled()){
			List<String> keys = getSendMessageKeys(msgCtxs);
			log.info("clear old SendMessageContext from transaction resources before store: {}", keys);
		}
		
		Set<SendMessageContext> contexts = findAllInCurrentContext();
		if(contexts==null){
			contexts = new LinkedHashSet<SendMessageContext>();
//			messageStorer.set(contexts);
			TransactionSynchronizationManager.bindResource(this, contexts);
		}
		contexts.add(ctx);
		if(debug && log.isInfoEnabled()){
			log.info("store in current context: {}", ctx.getMessage());
		}
	}
	
	@SuppressWarnings("unchecked")
	public Set<SendMessageContext> findAllInCurrentContext(){
//		return messageStorer.get();
		return (Set<SendMessageContext>)TransactionSynchronizationManager.getResource(this);
	}
	
	@SuppressWarnings("unchecked")
	public void clearInCurrentContext(){
		Set<SendMessageContext> msgCtxs = (Set<SendMessageContext>)TransactionSynchronizationManager.unbindResourceIfPossible(this);
		if(LangUtils.isEmpty(msgCtxs)){
			return ;
		}
		
		boolean debug = msgCtxs.iterator().next().isDebug();
		List<String> keys = getSendMessageKeys(msgCtxs);
		if(debug && log.isInfoEnabled()){
			log.info("clear SendMessageContext from transaction resources: {}", keys);
		}
		/*baseEntityManager.removeByIds(SendMessageEntity.class, keys.toArray(new String[0]));
		if(debug && log.isInfoEnabled()){
			log.info("clear SendMessageContext from database: {}", keys);
		}*/
	}
	
	public static List<String> getSendMessageKeys(Collection<SendMessageContext> msgCtxs){
		if(LangUtils.isEmpty(msgCtxs)){
			return Collections.emptyList();
		}
		return msgCtxs.stream().map(ctx->ctx.getMessageEntity().getKey()).collect(Collectors.toList());
	}

}

@@ -254,24 +254,77 @@
 ityManager;%0A
+import org.onetwo.common.exception.ServiceException;%0A
 import org.o
@@ -401,55 +401,8 @@
 ls;%0A
-import org.onetwo.dbm.id.SnowflakeIdGenerator;%0A
 impo
@@ -1017,16 +1017,18 @@
 er();%0A%09%0A
+//
 %09private
@@ -1562,16 +1562,18 @@
 (key))%7B%0A
+//
 %09%09%09key =
@@ -1607,24 +1607,119 @@
 .nextId());%0A
+%09%09%09//%E5%BC%BA%E5%88%B6%E5%BF%85%E5%A1%AB%EF%BC%8C%E5%8F%AF%E7%94%A8%E4%BA%8Eclient%E5%81%9Aidempotent%0A%09%09%09throw new ServiceException(%22message key can not be blank!%22);%0A
 %09%09%7D%0A%09%09send.s

修改为message key必填