com.hazelcast.ringbuffer.impl

Class RingbufferContainer<T,E>

java.lang.Object

com.hazelcast.ringbuffer.impl.RingbufferContainer<T,E>

Type Parameters:

T - the type of items in the ringbuffer container

E - the type of items in the ringbuffer

All Implemented Interfaces:

DataSerializable, IdentifiedDataSerializable, Versioned, Notifier

public class RingbufferContainer<T,E>
extends Object
implements IdentifiedDataSerializable, Notifier, Versioned

The RingbufferContainer is responsible for the functionalities supporting the underlying ringbuffer structure containing the data. This includes creating the actual ringbuffer, handling optional ringbuffer storage, keeping the wait/notify key for blocking operations, ringbuffer item expiration and other things not related to the ringbuffer data structure.

Constructor Summary

Constructors

Constructor and Description
RingbufferContainer() For purposes of IdentifiedDataSerializable instance creation.
RingbufferContainer(ObjectNamespace namespace, int partitionId) Constructs the ring buffer container with only the name and the key for blocking operations.
RingbufferContainer(ObjectNamespace namespace, RingbufferConfig config, NodeEngine nodeEngine, int partitionId) Constructs a fully initialized ringbuffer that can be used immediately.

Method Summary

Modifier and Type	Method and Description
long	add(T item) Adds one item to the ring buffer.
long	addAll(T[] items) Adds all items to the ringbuffer.
void	checkBlockableReadSequence(long readSequence) Check if the sequence is of an item that can be read immediately or is the sequence of the next item to be added into the ringbuffer.
void	cleanup()
void	clear() Clears the data in the ringbuffer.
long	getCapacity() Returns the capacity of this ringbuffer.
RingbufferConfig	getConfig()
int	getFactoryId() Returns DataSerializableFactory factory ID for this class.
int	getId() Returns type identifier for this class.
ObjectNamespace	getNamespace()
WaitNotifyKey	getNotifiedKey()
Ringbuffer<E>	getRingbuffer() Returns the ringbuffer containing the actual items.
RingbufferWaitNotifyKey	getRingEmptyWaitNotifyKey() Gets the wait/notify key for the blocking operations of reading from the ring buffer.
RingbufferStoreWrapper	getStore()
long	headSequence()
void	init(RingbufferConfig config, NodeEngine nodeEngine) Initializes the ring buffer with references to other services, the ringbuffer store and the config.
boolean	isEmpty()
boolean	isStaleSequence(long sequence)
boolean	isTooLargeSequence(long sequence)
Data	readAsData(long sequence) Reads one item from the ring buffer and returns the serialized format.
void	readData(ObjectDataInput in) Reads fields from the input stream
long	readMany(long beginSequence, ReadResultSetImpl result) Reads multiple items from the ring buffer and adds them to result in the stored format.
long	remainingCapacity() Returns the remaining capacity of the ringbuffer.
void	set(long sequenceId, T item) Sets the item at the given sequence ID and updates the expiration time if TTL is configured.
void	setHeadSequence(long sequence) Sets the head sequence.
void	setTailSequence(long sequence) Sets the tail sequence.
boolean	shouldNotify()
boolean	shouldWait(long sequence) Returns if the sequence is one after the sequence of the newest item in the ring buffer.
long	size()
long	tailSequence()
void	writeData(ObjectDataOutput out) Writes object fields to output stream

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

RingbufferContainer

public RingbufferContainer()

For purposes of IdentifiedDataSerializable instance creation. For any other purpose, use other constructors in this class.

RingbufferContainer

public RingbufferContainer(ObjectNamespace namespace,
                           int partitionId)

Constructs the ring buffer container with only the name and the key for blocking operations. This does not fully prepare the container for usage. The caller must invoke the init(RingbufferConfig, NodeEngine) method to complete the initialization before usage.

Parameters:

namespace - the namespace of the ringbuffer container

RingbufferContainer

public RingbufferContainer(ObjectNamespace namespace,
                           RingbufferConfig config,
                           NodeEngine nodeEngine,
                           int partitionId)

Constructs a fully initialized ringbuffer that can be used immediately. References to other services, the ringbuffer store, expiration policy and the config are all set.

Parameters:

namespace - the namespace of the ring buffer container

config - the configuration of the ring buffer

nodeEngine - the NodeEngine

Method Detail

init

public void init(RingbufferConfig config,
                 NodeEngine nodeEngine)

Initializes the ring buffer with references to other services, the ringbuffer store and the config. This is because on a replication operation the container is only partially constructed. The init method finishes the configuration of the ring buffer container for further usage.

Parameters:

config - the configuration of the ring buffer

nodeEngine - the NodeEngine

getStore

public RingbufferStoreWrapper getStore()

getRingEmptyWaitNotifyKey

public RingbufferWaitNotifyKey getRingEmptyWaitNotifyKey()

Gets the wait/notify key for the blocking operations of reading from the ring buffer.

Returns:

the wait/notify key

getConfig

public RingbufferConfig getConfig()

tailSequence

public long tailSequence()

headSequence

public long headSequence()

setHeadSequence

public void setHeadSequence(long sequence)

Sets the head sequence. The head sequence cannot be larger than tailSequence() + 1

Parameters:

sequence - the new head sequence

Throws:

IllegalArgumentException - if the target sequence is greater than tailSequence() + 1

See Also:

tailSequence()

setTailSequence

public void setTailSequence(long sequence)

Sets the tail sequence. The tail sequence cannot be less than headSequence() - 1.

Parameters:

sequence - the new tail sequence

Throws:

IllegalArgumentException - if the target sequence is less than headSequence() - 1

See Also:

headSequence()

getCapacity

public long getCapacity()

Returns the capacity of this ringbuffer.

Returns:

the capacity

size

public long size()

isEmpty

public boolean isEmpty()

shouldWait

public boolean shouldWait(long sequence)

Returns if the sequence is one after the sequence of the newest item in the ring buffer. In this case, the caller needs to wait for the item to be available. This method will throw an exception if the requested instance is larger than the sequence of the next item to be added (e.g. the second or third next item to added). If the ring buffer store is configured and enabled, this method will allow sequence values older than the head sequence.

Parameters:

sequence - the requested sequence

Returns:

should the caller wait for

Throws:

StaleSequenceException - if the requested sequence is 1. greater than the tail sequence + 1 or 2. smaller than the head sequence and the data store is not enabled

remainingCapacity

public long remainingCapacity()

Returns the remaining capacity of the ringbuffer. If TTL is enabled, then the returned capacity is equal to the total capacity of the ringbuffer minus the number of used slots in the ringbuffer which have not yet been marked as expired and cleaned up. Keep in mind that some slots could have expired items that have not yet been cleaned up and that the returned value could be stale as soon as it is returned.

If TTL is disabled, the remaining capacity is equal to the total ringbuffer capacity.

Returns:

the remaining capacity of the ring buffer

add

public long add(T item)

Adds one item to the ring buffer. Sets the expiration time if TTL is configured and also attempts to store the item in the data store if one is configured. The provided item can be Data or the deserialized object. The provided item will be transformed to the configured ringbuffer InMemoryFormat if necessary.

Parameters:

item - item to be stored in the ring buffer and data store, can be Data or an deserialized object

Returns:

the sequence ID of the item stored in the ring buffer

Throws:

HazelcastException - if there was any exception thrown by the data store

HazelcastSerializationException - if the ring buffer is configured to keep items in object format and the item could not be deserialized

addAll

public long addAll(T[] items)

Adds all items to the ringbuffer. Sets the expiration time if TTL is configured and also attempts to store the items in the data store if one is configured.

Parameters:

items - items to be stored in the ring buffer and data store

Returns:

the sequence ID of the last item stored in the ring buffer

Throws:

HazelcastException - if there was any exception thrown by the data store

HazelcastSerializationException - if the ring buffer is configured to keep items in object format and the item could not be deserialized

set

public void set(long sequenceId,
                T item)

Sets the item at the given sequence ID and updates the expiration time if TTL is configured. Unlike other methods for adding items into the ring buffer, does not attempt to store the item in the data store. This method expands the ring buffer tail and head sequence to accommodate for the sequence. This means that it will move the head or tail sequence to the target sequence if the target sequence is less than the head sequence or greater than the tail sequence.

Parameters:

sequenceId - the sequence ID under which the item is stored

item - item to be stored in the ring buffer and data store

Throws:

HazelcastSerializationException - if the ring buffer is configured to keep items in object format and the item could not be deserialized

readAsData

public Data readAsData(long sequence)

Reads one item from the ring buffer and returns the serialized format. If the item is not available, it will try and load it from the ringbuffer store. If the stored format is already serialized, there is no serialization.

Parameters:

sequence - The sequence of the item to be read

Returns:

The item read

Throws:

StaleSequenceException - if the sequence is : 1. larger than the tailSequence or 2. smaller than the headSequence and the data store is disabled

readMany

public long readMany(long beginSequence,
                     ReadResultSetImpl result)

Reads multiple items from the ring buffer and adds them to result in the stored format. If an item is not available, it will try and load it from the ringbuffer store.

Parameters:

beginSequence - the sequence of the first item to read.

result - the List where the result are stored in.

Returns:

returns the sequenceId of the next item to read. This is needed if not all required items are found.

Throws:

StaleSequenceException - if the sequence is : 1. larger than the tailSequence or 2. smaller than the headSequence and the data store is disabled

cleanup

public void cleanup()

isStaleSequence

public boolean isStaleSequence(long sequence)

isTooLargeSequence

public boolean isTooLargeSequence(long sequence)

checkBlockableReadSequence

public void checkBlockableReadSequence(long readSequence)

Check if the sequence is of an item that can be read immediately or is the sequence of the next item to be added into the ringbuffer. Since this method allows the sequence to be one larger than the tailSequence(), the caller can use this method to check the sequence before performing a possibly blocking read. Also, the requested sequence can be smaller than the head sequence if the data store is enabled.

Parameters:

readSequence - the sequence wanting to be read

Throws:

StaleSequenceException - if the requested sequence is smaller than the head sequence and the data store is not enabled

IllegalArgumentException - if the requested sequence is greater than the tail sequence + 1 or

writeData

public void writeData(ObjectDataOutput out)
               throws IOException

Description copied from interface: DataSerializable

Writes object fields to output stream

Specified by:

writeData in interface DataSerializable

Parameters:

out - output

Throws:

IOException - if an I/O error occurs. In particular, an IOException may be thrown if the output stream has been closed.

readData

public void readData(ObjectDataInput in)
              throws IOException

Description copied from interface: DataSerializable

Reads fields from the input stream

Specified by:

readData in interface DataSerializable

Parameters:

in - input

Throws:

IOException - if an I/O error occurs. In particular, an IOException may be thrown if the input stream has been closed.

getRingbuffer

public Ringbuffer<E> getRingbuffer()

Returns the ringbuffer containing the actual items.

getNamespace

public ObjectNamespace getNamespace()

getFactoryId

public int getFactoryId()

Description copied from interface: IdentifiedDataSerializable

Returns DataSerializableFactory factory ID for this class.

Specified by:

getFactoryId in interface IdentifiedDataSerializable

Returns:

factory ID

getId

public int getId()

Description copied from interface: IdentifiedDataSerializable

Returns type identifier for this class. It should be unique per DataSerializableFactory.

Specified by:

getId in interface IdentifiedDataSerializable

Returns:

type ID

shouldNotify

public boolean shouldNotify()

Specified by:

shouldNotify in interface Notifier

getNotifiedKey

public WaitNotifyKey getNotifiedKey()

Specified by:

getNotifiedKey in interface Notifier

clear

public void clear()

Clears the data in the ringbuffer.