/*
 *
 *  * 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.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;

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

    @Override
    public synchronized Observable<QueueMessage> getMessages(int limit, int transactionTimeout, int waitTime, Class klass) {
        List<QueueMessage> returnQueue = new ArrayList<>();
        queue.drainTo(returnQueue,1);
        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() {

    }
}

@@ -982,16 +982,48 @@
 eption;%0A
+import java.util.AbstractQueue;%0A
 import j
@@ -1038,24 +1038,24 @@
 .ArrayList;%0A
-
 import java.
@@ -1135,16 +1135,116 @@
 ngQueue;
+%0Aimport java.util.concurrent.LinkedBlockingQueue;%0Aimport java.util.concurrent.PriorityBlockingQueue;
 %0A%0A/**%0A *
@@ -1384,21 +1384,22 @@
  public 
-Array
+Linked
 Blocking
@@ -1430,21 +1430,22 @@
 e = new 
-Array
+Linked
 Blocking
@@ -1456,16 +1456,8 @@
 e%3C%3E(
-10000000
 );%0A%0A
@@ -1651,24 +1651,24 @@
 ayList%3C%3E();%0A
-
         queu
@@ -1690,16 +1690,17 @@
 nQueue,1
+0
 );%0A     

change queue impl