/*
    * Copyright 2016 The Lannister Project
    * 
    * Licensed under the Apache License, Version 2.0 (the "License");
    * you may not use this file except in compliance with the License.
    * You may obtain a copy of the License at
    * 
    *     http://www.apache.org/licenses/LICENSE-2.0
    * 
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package net.anyflow.lannister.message;
  
  import java.util.Set;
  import java.util.concurrent.locks.Lock;
  
  import com.google.common.collect.Sets;
  
  import net.anyflow.lannister.cluster.ClusterDataFactory;
  import net.anyflow.lannister.cluster.Map;
  import net.anyflow.lannister.cluster.SerializableIntegerSet;
  import net.anyflow.lannister.cluster.SerializableStringSet;
  
  public class InboundMessageStatuses {
  	private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(InboundMessageStatuses.class);
  
  	private final Map<String, InboundMessageStatus> data;
  	private final Map<Integer, SerializableStringSet> messageidIndex;
  	private final Map<String, SerializableIntegerSet> clientidIndex;
  
  	private final Lock modifyLock;
  
  	protected InboundMessageStatuses() {
  		this.data = ClusterDataFactory.INSTANCE.createMap("InboundMessageStatuses_data");
  		this.messageidIndex = ClusterDataFactory.INSTANCE.createMap("InboundMessageStatuses_messageidIndex");
  		this.clientidIndex = ClusterDataFactory.INSTANCE.createMap("InboundMessageStatuses_clientidIndex");
  
  		this.modifyLock = ClusterDataFactory.INSTANCE.createLock("InboundMessageStatuses_modifyLock");
  	}
  
  	public static String key(Integer messageId, String clientId) {
  		return clientId + "_" + Integer.toString(messageId);
  	}
  
  	public int size() {
  		return data.size();
  	}
  
  	public Set<String> keySet() {
  		return Sets.newHashSet(data.keySet());
  	}
  
  	public void put(InboundMessageStatus inboundMessageStatus) {
  		if (inboundMessageStatus == null) { return; }
  
  		modifyLock.lock();
  		try {
  			this.data.put(inboundMessageStatus.key(), inboundMessageStatus);
  
  			SerializableStringSet clientIds = this.messageidIndex.get(inboundMessageStatus.messageId());
  			if (clientIds == null) {
  				clientIds = new SerializableStringSet();
  			}
  			clientIds.add(inboundMessageStatus.clientId());
  			this.messageidIndex.put(inboundMessageStatus.messageId(), clientIds);
  
  			SerializableIntegerSet messageIds = this.clientidIndex.get(inboundMessageStatus.clientId());
  			if (messageIds == null) {
  				messageIds = new SerializableIntegerSet();
  			}
  			messageIds.add(inboundMessageStatus.messageId());
  			this.clientidIndex.put(inboundMessageStatus.clientId(), messageIds);
  
  			MessageReferenceCounts.INSTANCE.retain(inboundMessageStatus.messageKey());
  
  			if (logger.isDebugEnabled()) {
  				logger.debug("InboundMessageStatus removed [messageId={}, clientId={}, status=]",
  						inboundMessageStatus.messageId(), inboundMessageStatus.clientId(),
  						inboundMessageStatus.status());
  				logger.debug("InboundMessageStatuses Size [data={}, messageidIndex={}, clientidIndex={}]", data.size(),
  						messageidIndex.size(), clientidIndex.size());
  			}
  		}
  		finally {
  			modifyLock.unlock();
  		}
  	}
  
  	public InboundMessageStatus getBy(Integer messageId, String clientId) {
  		return get(key(messageId, clientId));
  	}
  
  	public InboundMessageStatus get(String key) {
  		return data.get(key);
  	}
 
 	public InboundMessageStatus removeByKey(Integer messageId, String clientId) {
 		return removeByKey(key(messageId, clientId));
 	}
 
 	private InboundMessageStatus removeByKey(String key) {
 		modifyLock.lock();
 
 		try {
 			InboundMessageStatus removed = this.data.remove(key);
 			if (removed == null) { return null; }
 
 			SerializableStringSet clientIds = messageidIndex.get(removed.messageId());
 			clientIds.remove(removed.clientId());
 			if (clientIds.size() <= 0) {
 				messageidIndex.remove(removed.messageId());
 			}
 			else {
 				messageidIndex.put(removed.messageId(), clientIds);
 			}
 
 			SerializableIntegerSet messageIds = clientidIndex.get(removed.clientId());
 			messageIds.remove(removed.messageId());
 			if (messageIds.size() <= 0) {
 				clientidIndex.remove(removed.clientId());
 			}
 			else {
 				clientidIndex.put(removed.clientId(), messageIds);
 			}
 
 			MessageReferenceCounts.INSTANCE.release(removed.messageKey());
 
 			if (logger.isDebugEnabled()) {
 				logger.debug("InboundMessageStatus removed [messageId={}, clientId={}, status=]", removed.messageId(),
 						removed.clientId(), removed.status());
 				logger.debug("InboundMessageStatuses Size [data={}, messageidIndex={}, clientidIndex={}]", data.size(),
 						messageidIndex.size(), clientidIndex.size());
 			}
 
 			return removed;
 		}
 		finally {
 			modifyLock.unlock();
 		}
 	}
 
 	public void update(Integer messageId, String clientId, InboundMessageStatus.Status targetStatus) {
 		String key = key(messageId, clientId);
 
 		InboundMessageStatus status = data.get(key);
 		if (status == null) { return; }
 
 		status.status(targetStatus);
 
 		data.put(key, status);
 
 		if (logger.isDebugEnabled()) {
 			logger.debug("InboundMessageStatus removed [messageId={}, clientId={}, status=]", status.messageId(),
 					status.clientId(), status.status());
 		}
 	}
 }