/*
 *
 *  * Licensed to the Apache Software Foundation (ASF) under one or more
 *  *  contributor license agreements.  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.  For additional information regarding
 *  * copyright in this work, please see the NOTICE file in the top level
 *  * directory of this distribution.
 *
 */

package org.apache.usergrid.persistence.queue;

import rx.Observable;

import java.io.IOException;
import java.util.AbstractQueue;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.PriorityBlockingQueue;

/**
 * Default queue manager implementation, uses in memory linked queue
 */
public class DefaultQueueManager implements QueueManager {
    public LinkedBlockingQueue<QueueMessage> queue = new LinkedBlockingQueue<>();

    @Override
    public  Observable<QueueMessage> getMessages(int limit, int transactionTimeout, int waitTime, Class klass) {
        List<QueueMessage> returnQueue = new ArrayList<>();
        queue.drainTo(returnQueue,1000);
        return Observable.from( returnQueue);
    }

    @Override
    public long getQueueDepth() {
        return queue.size();
    }

    @Override
    public void commitMessage(QueueMessage queueMessage) {
    }

    @Override
    public void commitMessages(List<QueueMessage> queueMessages) {
    }

    @Override
    public synchronized void sendMessages(List bodies) throws IOException {
        for(Object body : bodies){
            String uuid = UUID.randomUUID().toString();
            queue.add(new QueueMessage(uuid,"handle_"+uuid,body,"putappriate type here"));
        }
    }

    @Override
    public synchronized void sendMessage(Object body) throws IOException {
        String uuid = UUID.randomUUID().toString();
        queue.add(new QueueMessage(uuid,"handle_"+uuid,body,"put type here"));
    }

    @Override
    public void deleteQueue() {

    }
}

@@ -1235,16 +1235,54 @@
 ngQueue;
+%0Aimport java.util.concurrent.TimeUnit;
 %0A%0A/**%0A *
@@ -1422,22 +1422,21 @@
  public 
-Linked
+Array
 Blocking
@@ -1467,22 +1467,21 @@
 e = new 
-Linked
+Array
 Blocking
@@ -1488,16 +1488,20 @@
 Queue%3C%3E(
+1000
 );%0A%0A    
@@ -1522,16 +1522,18 @@
 public  
+  
 Observab
@@ -1697,40 +1697,410 @@
     
-queue.drainTo(returnQueue,1000);
+try %7B%0A            QueueMessage message=null;%0A            int count = 10;%0A            do %7B%0A                message = queue.poll(100, TimeUnit.MILLISECONDS);%0A                if (message != null) %7B%0A                    returnQueue.add(message);%0A                %7D%0A            %7Dwhile(message!=null && count--%3E0);%0A        %7Dcatch (InterruptedException ie)%7B%0A            throw new RuntimeException(ie);%0A        %7D
 %0A   
@@ -2422,36 +2422,24 @@
 %0A    public 
-synchronized
  void sendMe
@@ -2562,32 +2562,54 @@
 D().toString();%0A
+            try %7B%0A    
             queu
@@ -2602,35 +2602,35 @@
           queue.
-add
+put
 (new QueueMessag
@@ -2628,32 +2628,33 @@
 eueMessage(uuid,
+ 
 %22handle_%22+uuid,b
@@ -2650,31 +2650,27 @@
 le_%22
-+
+ + 
 uuid,
+ 
 body,
+ 
 %22put
-appriate
  typ
@@ -2676,24 +2676,132 @@
 pe here%22));%0A
+            %7Dcatch (InterruptedException ie)%7B%0A                throw new RuntimeException(ie);%0A            %7D%0A
         %7D%0A  
@@ -2834,20 +2834,8 @@
 lic 
-synchronized
  voi
@@ -2926,32 +2926,50 @@
 D().toString();%0A
+        try %7B%0A    
         queue.ad
@@ -2966,19 +2966,19 @@
   queue.
-add
+put
 (new Que
@@ -2992,16 +2992,17 @@
 ge(uuid,
+ 
 %22handle_
@@ -3006,19 +3006,23 @@
 le_%22
-+
+ + 
 uuid,
+ 
 body,
+ 
 %22put
@@ -3032,24 +3032,120 @@
 pe here%22));%0A
+        %7Dcatch (InterruptedException ie)%7B%0A            throw new RuntimeException(ie);%0A        %7D%0A
     %7D%0A%0A    @

change queue impl