/*
 * Copyright (c) 2012-2015 The original author or authors
 * ------------------------------------------------------
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * and Apache License v2.0 which accompanies this distribution.
 *
 * The Eclipse Public License is available at
 * http://www.eclipse.org/legal/epl-v10.html
 *
 * The Apache License v2.0 is available at
 * http://www.opensource.org/licenses/apache2.0.php
 *
 * You may elect to redistribute this code under either of these licenses.
 */
package io.moquette.spi.persistence;

import io.moquette.spi.IMatchingCondition;
import io.moquette.spi.IMessagesStore;
import org.mapdb.DB;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;

/**
 * IMessagesStore implementation backed by MapDB.
 *
 * @author andrea
 */
class MapDBMessagesStore implements IMessagesStore {

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

    private DB m_db;

    //map clientID <-> set of currently in flight packet identifiers
    private Map<String, AtomicLong> m_inFlightIds;
    //maps clientID -> guid
    private ConcurrentMap<String, String> m_retainedStore;
    //maps guid to message, it's message store
    private ConcurrentMap<String, IMessagesStore.StoredMessage> m_persistentMessageStore;


    MapDBMessagesStore(DB db) {
        m_db = db;
    }

    @Override
    public void initStore() {
        m_retainedStore = m_db.getHashMap("retained");
        m_persistentMessageStore = m_db.getHashMap("persistedMessages");
        m_inFlightIds = m_db.getHashMap("inflightPacketIDs");
    }

    @Override
    public void storeRetained(String topic, String guid) {
        m_retainedStore.put(topic, guid);
    }

    @Override
    public Collection<StoredMessage> searchMatching(IMatchingCondition condition) {
        LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size());

        List<StoredMessage> results = new ArrayList<>();
        for (Map.Entry<String, String> entry : m_retainedStore.entrySet()) {
            final String guid = entry.getValue();
            StoredMessage storedMsg = m_persistentMessageStore.get(guid);
            if (condition.match(entry.getKey())) {
                results.add(storedMsg);
            }
        }

        return results;
    }

    @Override
    public String storePublishForFuture(StoredMessage evt) {
        LOG.debug("storePublishForFuture store evt {}", evt);
        if (evt.getClientID() == null) {
            LOG.error("persisting a message without a clientID, bad programming error msg: {}", evt);
            throw new IllegalArgumentException("\"persisting a message without a clientID, bad programming error");
        }
        String guid = UUID.randomUUID().toString();
        evt.setGuid(guid);
        m_persistentMessageStore.put(guid, evt);
        ConcurrentMap<Integer, String> messageIdToGuid = m_db.getHashMap(MapDBSessionsStore.messageId2GuidsMapName(evt.getClientID()));
        messageIdToGuid.put(evt.getMessageID(), guid);
        return guid;
    }

    @Override
    public List<StoredMessage> listMessagesInSession(Collection<String> guids) {
        List<StoredMessage> ret = new ArrayList<>();
        for (String guid : guids) {
            ret.add(m_persistentMessageStore.get(guid));
        }
        return ret;
    }

    @Override
    public void dropMessagesInSession(String clientID) {
        m_db.getHashMap(MapDBSessionsStore.messageId2GuidsMapName(clientID)).clear();
        m_persistentMessageStore.remove(clientID);
    }

    @Override
    public StoredMessage getMessageByGuid(String guid) {
        return m_persistentMessageStore.get(guid);
    }

    /**
     * Return the next valid packetIdentifier for the given client session.
     * */
    @Override
    public int nextPacketID(String clientID) {
        AtomicLong atomicLong = this.m_inFlightIds.get(clientID);
        if (atomicLong == null) {
            AtomicLong messageId = new AtomicLong(1);
            this.m_inFlightIds.put(clientID, messageId);
            return 1;
        }
        int incrementAndGet = (int) atomicLong.incrementAndGet();
        if (incrementAndGet == 0XFFFF) {
            atomicLong.set(1);
            return 1;
        }

        return incrementAndGet;
    }

    @Override
    public void cleanRetained(String topic) {
        m_retainedStore.remove(topic);
    }
}

@@ -856,55 +856,8 @@
 Map;
-%0Aimport java.util.concurrent.atomic.AtomicLong;
 %0A%0A/*
@@ -1192,26 +1192,28 @@
 String, 
-AtomicLong
+Set%3CInteger%3E
 %3E m_inFl
@@ -4023,29 +4023,38 @@
     
-AtomicLong atomicLong
+Set%3CInteger%3E inFlightForClient
  = t
@@ -4098,26 +4098,33 @@
     if (
-atomicLong
+inFlightForClient
  == null
@@ -4143,47 +4143,125 @@
     
-AtomicLong messageId = new AtomicLong(1
+int nextPacketId = 1;%0A            inFlightForClient = new HashSet%3C%3E();%0A            inFlightForClient.add(nextPacketId
 );%0A 
@@ -4304,25 +4304,33 @@
 ientID, 
-messageId
+inFlightForClient
 );%0A     
@@ -4335,35 +4335,47 @@
          return 
-1
+nextPacketId
 ;%0A
+%0A
         %7D%0A      
@@ -4360,32 +4360,33 @@
 tId;%0A%0A        %7D%0A
+%0A
         int incr
@@ -4385,102 +4385,134 @@
 int 
-incrementAndGet = (int) atomicLong.incrementAndGet();%0A        if (incrementAndGet == 0X
+maxId = inFlightForClient.isEmpty() ? 0 : Collections.max(inFlightForClient);%0A        int nextPacketId = (maxId + 1) %25 0x
 FFFF
-) %7B
+;
 %0A   
@@ -4520,94 +4520,72 @@
     
-    atomicLong.set(1);%0A            return 1;%0A        %7D%0A%0A        return incrementAndGet
+inFlightForClient.add(nextPacketId);%0A        return nextPacketId
 ;%0A  

Rollback issue #130 because the remove from the inflight is done by MapDBSessionsStore.inFlightAck