/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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 org.apache.usergrid.persistence.qakka.core.impl;

import com.datastax.driver.core.utils.UUIDs;
import com.google.inject.Inject;
import com.google.inject.Singleton;
import org.apache.usergrid.persistence.model.util.UUIDGenerator;
import org.apache.usergrid.persistence.qakka.QakkaFig;
import org.apache.usergrid.persistence.qakka.distributed.actors.QueueRefresher;
import org.apache.usergrid.persistence.qakka.serialization.queuemessages.DatabaseQueueMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;
import java.util.Queue;
import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedQueue;


@Singleton
public class InMemoryQueue {
    private static final Logger logger = LoggerFactory.getLogger( QueueRefresher.class );

    private final Map<String, Queue<DatabaseQueueMessage>> queuesByName;
    private final Map<String, UUID> newestByQueueName;


    @Inject
    InMemoryQueue(QakkaFig qakkaFig) {
        queuesByName = new HashMap<>( qakkaFig.getQueueInMemorySize() );
        newestByQueueName = new HashMap<>( qakkaFig.getQueueInMemorySize() );
    }

    private Queue<DatabaseQueueMessage> getQueue( String queueName ) {
        synchronized ( queuesByName ) {
            if ( !queuesByName.containsKey( queueName )) {
                queuesByName.put( queueName, new ConcurrentLinkedQueue<>() );
            }
            return queuesByName.get( queueName );
        }
    }

    public void add( String queueName, DatabaseQueueMessage databaseQueueMessage ) {
        UUID newest = newestByQueueName.get( queueName );
        if ( newest == null ) {
            newest = databaseQueueMessage.getQueueMessageId();
        } else {
            if ( databaseQueueMessage.getQueueMessageId().compareTo( newest ) > 0 ) {
                newest = databaseQueueMessage.getQueueMessageId();
            }
        }
        newestByQueueName.put( queueName, newest );
        getQueue( queueName ).add( databaseQueueMessage );
    }

    public UUID getNewest( String queueName ) {
        UUID newest = newestByQueueName.get( queueName );
//        if ( newest == null ) {
//            // Create oldest UUID from a UNIX timestamp via DataStax utility
//            // https://docs.datastax.com/en/drivers/java/2.0/com/datastax/driver/core/utils/UUIDs.html
//            newest = UUIDs.startOf( 0L );
//            newestByQueueName.put( queueName, newest );
//        }
        return newest;
    }

    public DatabaseQueueMessage poll( String queueName ) {
        return getQueue( queueName ).poll();
    }

    public DatabaseQueueMessage peek( String queueName ) {
        return getQueue( queueName ).peek();
    }

    public int size( String queueName ) {
        return getQueue( queueName ).size();
    }
}

@@ -865,53 +865,8 @@
 l;%0A%0A
-import com.datastax.driver.core.utils.UUIDs;%0A
 impo
@@ -973,167 +973,22 @@
 nce.
-model.util.UUIDGenerator;%0Aimport org.apache.usergrid.persistence.qakka.QakkaFig;%0Aimport org.apache.usergrid.persistence.qakka.distributed.actors.QueueRefresher
+qakka.QakkaFig
 ;%0Aim
@@ -1159,84 +1159,9 @@
 til.
-HashMap;%0Aimport java.util.Map;%0Aimport java.util.Queue;%0Aimport java.util.UUID
+*
 ;%0Aim
@@ -1321,32 +1321,67 @@
 er( 
-QueueRefresher.class );%0A
+InMemoryQueue.class );%0A%0A    /** In-memory queues by name */
 %0A   
@@ -1449,16 +1449,67 @@
 sByName;
+%0A%0A    /** Newest message seen by in-memory queue */
 %0A    pri
@@ -1626,24 +1626,29 @@
 ueuesByName 
+     
 = new HashMa
@@ -2182,24 +2182,25 @@
 Message ) %7B%0A
+%0A
         UUID
@@ -2332,32 +2332,33 @@
 eueMessageId();%0A
+%0A
         %7D else %7B
@@ -2420,31 +2420,40 @@
 d().
-compareTo( newest ) %3E 0
+timestamp() %3E newest.timestamp()
  ) %7B
@@ -2516,24 +2516,118 @@
 essageId();%0A
+                logger.debug(%22New newest for queue %7B%7D is %7B%7D%22, queueName, newest.timestamp());%0A
             
@@ -2630,32 +2630,33 @@
     %7D%0A        %7D%0A
+%0A
         newestBy
@@ -2817,94 +2817,49 @@
     
-UUID newest = newestByQueueName.get( queueName );%0A//        if ( newest == null
+if ( getQueue( queueName ).isEmpty()
  ) %7B%0A
-//
     
@@ -2870,232 +2870,65 @@
     
-// Create oldest UUID from a UNIX timestamp via DataStax utility%0A//            // https://docs.datastax.com/en/drivers/java/2.0/com/datastax/driver/core/utils/UUIDs.html%0A//            newest = UUIDs.startOf( 0L );%0A//
+newestByQueueName.remove( queueName );%0A        %7D%0A
         
     
@@ -2915,35 +2915,38 @@
       %7D%0A        
-   
+return
  newestByQueueNa
@@ -2940,34 +2940,34 @@
 westByQueueName.
-pu
+ge
 t( queueName, ne
@@ -2966,53 +2966,10 @@
 Name
-, newest );%0A//        %7D%0A        return newest
+ )
 ;%0A  

Fix logic to compare timestamps, not UUIDs and to zero-out the "since" time when queue is empty.