PendingMessageQueue (mongodb-async-driver 2.0.1 API)

java.lang.Object
- com.allanbank.mongodb.client.message.PendingMessageQueue

```
public final class PendingMessageQueue
extends Object
```
PendingMessageQueue provides an optimized queue for pending messages inspired by the Disruptor project.
To reduce thread contention the queue uses a set of integer values to track the position of the ready messages (the last message that is ready to be read), reserve (the first message that can be reserved to be written to), and the take (the next (first) message to be read). For an infinite queue the following invariant holds:
```
 
 take < readyBefore <= reserve
 
 
```
To make handling a limited size queue easier the size of the queue is forced to power of 2 less than 1048576. The roll over can then be handled with a simple mask operation.

Rather than allocate a pending message per request we use an array of pre-allocated PendingMessages and copy the data into and out of the objects. this has a net positive effect on object allocation and garbage collection time at the cost of a longer initialization.

Lastly, This queue assumes there is a single consumer of messages. This is true for the driver's use case but don't copy the code and expect it to work with multiple consumers. The consumer should use the following basic structure:
```
 
 PendingMessage pm = new PendingMessage();
 
 queue.take(pm); // Blocks.
 // Handle the message.
 
 // or
 
 if( queue.poll(pm) ) { // Non-blocking.
    // Handle The Message.
 }
 
 
```
Warning: This class has been carefully tuned for the driver's use case. Changes should be carefully bench marked and tested. Comments have been embedded in the source indicating attempted changes and reverts. Due to its position in the driver subtle changes in this class can cause large changes in the performance of the driver.
See Also:
Disruptor Project
API Warning:

This class is NOT part of the drivers API. This class may be mutated in incompatible ways between any two releases of the driver.

Copyright ©:

2012-2014, Allanbank Consulting, Inc., All Rights Reserved

Field Summary

Fields
Modifier and Type	Field and Description
`static long`	`MAX_MESSAGE_ID_MASK` The mask for constraining the size the message id.
`static int`	`MAX_SIZE` The maximum size of the queue.
`static long`	`SPIN_TIME_NS` Amount of time to spin before yielding.
`static long`	`YIELD_TIME_NS` Amount of time to spin/yield before waiting.

Constructor Summary

Constructors
Constructor and Description

PendingMessageQueue(int size, LockType lockType)
Creates a new PendingMessageQueue.

Constructors
Constructor and Description
`PendingMessageQueue(int size, LockType lockType)` Creates a new PendingMessageQueue.

Method Summary

Methods
Modifier and Type	Method and Description
`int`	`capacity()` Returns the size of the queue.
`void`	`drainTo(List<PendingMessage> pending)` Drains the list of pending messages into the provided list.
`protected int`	`increment(int index)` Increments the index handling roll-over.
`boolean`	`isEmpty()` Returns true if the queue is empty.
`protected void`	`markReady(int index)` Marks the position as ready by incrementing the ready position to the provided position.
`protected void`	`markReady2(int index)` Marks the position and the next position as ready by incrementing the ready position to the provided position + 1.
`protected void`	`notifyWaiters(boolean all)` Notifies the waiting threads that the state of the queue has changed.
`protected int`	`offer()` Checks if there is remove for another message.
`boolean`	`offer(Message message, ReplyCallback replyCallback)` Puts a message onto the queue.
`boolean`	`offer(PendingMessage pendingMessage)` Puts a message onto the queue.
`protected int`	`offer2()` Checks if there is remove for another two message.
`boolean`	`poll(PendingMessage copyOut)` Returns the next message from the queue without blocking.
`void`	`put(Message message, ReplyCallback replyCallback)` Puts a message onto the queue.
`void`	`put(Message message, ReplyCallback replyCallback, Message message2, ReplyCallback replyCallback2)` Puts two messages onto the queue.
`void`	`put(PendingMessage pendingMessage)` Puts a message onto the queue.
`int`	`size()` Returns the number of messages in the queue.
`void`	`take(PendingMessage copyOut)` Returns the next message from the queue and will block waiting for a message.

Methods inherited from class java.lang.Object
clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

- Field Detail
  - MAX_MESSAGE_ID_MASK
```
public static final long MAX_MESSAGE_ID_MASK
```
    The mask for constraining the size the message id.
    
    See Also:
    Constant Field Values
  - MAX_SIZE
```
public static final int MAX_SIZE
```
    The maximum size of the queue. This it currently 2^20 but must be at most 2^30 to ensure masking works.
    
    See Also:
    Constant Field Values
  - SPIN_TIME_NS
```
public static final long SPIN_TIME_NS
```
    Amount of time to spin before yielding. Set to 1/100 of a millisecond.
  - YIELD_TIME_NS
```
public static final long YIELD_TIME_NS
```
    Amount of time to spin/yield before waiting. Set to 1/2 millisecond.
- Constructor Detail
  - PendingMessageQueue
```
public PendingMessageQueue(int size,
                   LockType lockType)
```
    Creates a new PendingMessageQueue.
    
    Parameters:
    size - The size of the queue to create.
    lockType - The lock type to use with the queue.
- Method Detail
  - capacity
```
public int capacity()
```
    Returns the size of the queue.
    
    Returns:
    The size of the queue.
  - drainTo
```
public void drainTo(List<PendingMessage> pending)
```
    Drains the list of pending messages into the provided list.
    
    Parameters:
    pending - The list to add all of the pending messages to.
  - isEmpty
```
public boolean isEmpty()
```
    Returns true if the queue is empty. e.g., the next take position is the read before position.
    
    Returns:
    If the queue is empty.
  - offer
```
public boolean offer(Message message,
            ReplyCallback replyCallback)
```
    Puts a message onto the queue. This method will not block waiting for a space to add the message.
    
    Parameters:
    message - The message to add.
    replyCallback - The callback for the message to add.
    
    Returns:
    True if the message was added, false otherwise.
  - offer
```
public boolean offer(PendingMessage pendingMessage)
```
    Puts a message onto the queue. This method will not block waiting for a space to add the message.
    
    Parameters:
    pendingMessage - The message to add.
    
    Returns:
    True if the message was added, false otherwise.
  - poll
```
public boolean poll(PendingMessage copyOut)
```
    Returns the next message from the queue without blocking.
    PendingMessage pm = new PendingMessage(); if( queue.poll(pm) } { // Handle the message copied into pm. }
    Parameters:
    copyOut - The PendingMessage to copy the pending message into.
    
    Returns:
    True if the pending message was updated.
  - put
```
public void put(Message message,
       ReplyCallback replyCallback)
         throws InterruptedException
```
    Puts a message onto the queue. This method will block waiting for a space to add the message.
    
    Parameters:
    message - The message to add.
    replyCallback - The callback for the message to add.
    
    Throws:
    
    InterruptedException - If the thread is interrupted while waiting for the message. If thrown the message will not have been enqueued.
  - put
```
public void put(Message message,
       ReplyCallback replyCallback,
       Message message2,
       ReplyCallback replyCallback2)
         throws InterruptedException
```
    Puts two messages onto the queue. This method will block waiting for a space to add the messages but ensures the messages are in sequence in the queue.
    
    Parameters:
    message - The first message to add.
    replyCallback - The callback for the first message to add.
    message2 - The second message to add.
    replyCallback2 - The callback for the second message to add.
    
    Throws:
    
    InterruptedException - If the thread is interrupted while waiting for the message. If thrown neither message will have been enqueued.
  - put
```
public void put(PendingMessage pendingMessage)
         throws InterruptedException
```
    Puts a message onto the queue. This method will block waiting for a space to add the message.
    
    Parameters:
    pendingMessage - The message to add.
    
    Throws:
    
    InterruptedException - If the thread is interrupted while waiting for the message. If thrown the message will not have been enqueued.
  - size
```
public int size()
```
    Returns the number of messages in the queue.
    
    Returns:
    The number of messages in the queue.
  - take
```
public void take(PendingMessage copyOut)
          throws InterruptedException
```
    Returns the next message from the queue and will block waiting for a message.
    
    Parameters:
    copyOut - The PendingMessage to copy the pending message into.
    
    Throws:
    
    InterruptedException - If the thread is interrupted while waiting for the message.
  - increment
```
protected int increment(int index)
```
    Increments the index handling roll-over.
    
    Parameters:
    index - The value to increment.
    
    Returns:
    The incremented value.
  - markReady
```
protected void markReady(int index)
```
    Marks the position as ready by incrementing the ready position to the provided position. This method uses a spin lock assuming any other threads will increment the ready position quickly to the position just before index.
    
    Parameters:
    index - The index of the ready message.
  - markReady2
```
protected void markReady2(int index)
```
    Marks the position and the next position as ready by incrementing the ready position to the provided position + 1. This method uses a spin lock assuming any other threads will increment the ready position quickly to the position just before index.
    
    Parameters:
    index - The index of the ready message.
  - notifyWaiters
```
protected void notifyWaiters(boolean all)
```
    Notifies the waiting threads that the state of the queue has changed.
    
    Parameters:
    all - If true then all threads will be woken. Otherwise only a single thread is woken.
  - offer
```
protected int offer()
```
    Checks if there is remove for another message. If so returns the index of the message to update. If not return a value less then zero.
    
    Returns:
    The position of the message that can be updated or a value of less than zero if the queue is full.
  - offer2
```
protected int offer2()
```
    Checks if there is remove for another two message. If so returns the index of the first message to update. If not return a value less then zero.
    
    Returns:
    The position of the first message that can be updated or a value of less than zero if the queue is full.

Class PendingMessageQueue

Field Summary

Constructor Summary

Method Summary

Methods inherited from class java.lang.Object

Field Detail

MAX_MESSAGE_ID_MASK

MAX_SIZE

SPIN_TIME_NS

YIELD_TIME_NS

Constructor Detail

PendingMessageQueue

Method Detail

capacity

drainTo

isEmpty

offer

offer

poll

put

put

put

size

take

increment

markReady

markReady2

notifyWaiters

offer

offer2