// Path: qbit/core/src/main/java/io/advantageous/qbit/queue/ReceiveQueue.java
// public interface ReceiveQueue<T> extends Input, Stoppable {
// 
//     /**
//      * Gets the next item. If the item is null,
//      * means the timeout has been reached.
//      *
//      * @return value from poll
//      */
//     T pollWait();
// 
// 
//     /**
//      * Gets the next item. If the item is null the queue currently has no items.
//      *
//      * @return value from poll
//      */
//     T poll();
// 
//     /**
//      * Wait for the next item.
//      *
//      * @return value from take
//      */
//     T take();
// 
//     /**
//      * Read in a batch of items.
//      *
//      * @param max max number you want from batch
//      * @return batch of values
//      */
//     Iterable<T> readBatch(int max);
// 
// 
//     /**
//      * Read in a batch of items.
//      *
//      * @return batch of values
//      */
//     Iterable<T> readBatch();
// }


import io.advantageous.qbit.concurrent.ExecutorContext;
import io.advantageous.qbit.queue.ReceiveQueue;
import io.advantageous.qbit.queue.ReceiveQueueListener;
import io.advantageous.qbit.queue.ReceiveQueueManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.atomic.AtomicBoolean;
import static io.advantageous.qbit.concurrent.ScheduledExecutorBuilder.scheduledExecutorBuilder;
/*
 * Copyright (c) 2015. Rick Hightower, Geoff Chandler
 *
 * 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.
 *
 * QBit - The Microservice lib for Java : JSON, WebSocket, REST. Be The Web!
 */

package io.advantageous.qbit.queue.impl;





/**
 * created by Richard on 9/8/14.
 *
 * @author rhightower
 */
public class BasicReceiveQueueManager<T> implements ReceiveQueueManager<T> {


    private final Logger logger = LoggerFactory.getLogger(BasicReceiveQueueManager.class);
    private final boolean debug = logger.isDebugEnabled();
    private final String name;
    private final AtomicBoolean stop = new AtomicBoolean();
    private ExecutorContext executorContext;
    private QueueInfo<T> queueInfo;

    public BasicReceiveQueueManager(final String name) {
        this.name = name;
    }

    @Override
    public void start() {

        this.executorContext = scheduledExecutorBuilder()
                //.setDaemon(true) TODO #444 https://github.com/advantageous/qbit/issues/444
                .setThreadName("QueueListener|" + name)
                .setInitialDelay(50)
                .setPeriod(50).setRunnable(this::manageQueue)
                .build();

        executorContext.start();
    }

    private void manageQueue() {

        if (queueInfo == null) {
            return;
        }

        final String name = queueInfo.name;
// Path: qbit/core/src/main/java/io/advantageous/qbit/queue/ReceiveQueue.java
// public interface ReceiveQueue<T> extends Input, Stoppable {
// 
//     /**
//      * Gets the next item. If the item is null,
//      * means the timeout has been reached.
//      *
//      * @return value from poll
//      */
//     T pollWait();
// 
// 
//     /**
//      * Gets the next item. If the item is null the queue currently has no items.
//      *
//      * @return value from poll
//      */
//     T poll();
// 
//     /**
//      * Wait for the next item.
//      *
//      * @return value from take
//      */
//     T take();
// 
//     /**
//      * Read in a batch of items.
//      *
//      * @param max max number you want from batch
//      * @return batch of values
//      */
//     Iterable<T> readBatch(int max);
// 
// 
//     /**
//      * Read in a batch of items.
//      *
//      * @return batch of values
//      */
//     Iterable<T> readBatch();
// }


// Path: qbit/core/src/main/java/io/advantageous/qbit/queue/impl/BasicReceiveQueueManager.java
import io.advantageous.qbit.concurrent.ExecutorContext;
import io.advantageous.qbit.queue.ReceiveQueue;
import io.advantageous.qbit.queue.ReceiveQueueListener;
import io.advantageous.qbit.queue.ReceiveQueueManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.atomic.AtomicBoolean;
import static io.advantageous.qbit.concurrent.ScheduledExecutorBuilder.scheduledExecutorBuilder;

/*
 * Copyright (c) 2015. Rick Hightower, Geoff Chandler
 *
 * 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.
 *
 * QBit - The Microservice lib for Java : JSON, WebSocket, REST. Be The Web!
 */

package io.advantageous.qbit.queue.impl;





/**
 * created by Richard on 9/8/14.
 *
 * @author rhightower
 */
public class BasicReceiveQueueManager<T> implements ReceiveQueueManager<T> {


    private final Logger logger = LoggerFactory.getLogger(BasicReceiveQueueManager.class);
    private final boolean debug = logger.isDebugEnabled();
    private final String name;
    private final AtomicBoolean stop = new AtomicBoolean();
    private ExecutorContext executorContext;
    private QueueInfo<T> queueInfo;

    public BasicReceiveQueueManager(final String name) {
        this.name = name;
    }

    @Override
    public void start() {

        this.executorContext = scheduledExecutorBuilder()
                //.setDaemon(true) TODO #444 https://github.com/advantageous/qbit/issues/444
                .setThreadName("QueueListener|" + name)
                .setInitialDelay(50)
                .setPeriod(50).setRunnable(this::manageQueue)
                .build();

        executorContext.start();
    }

    private void manageQueue() {

        if (queueInfo == null) {
            return;
        }

        final String name = queueInfo.name;
        final ReceiveQueue<T> inputQueue = queueInfo.inputQueue;