ch.cern.dirq

Class QueueSimple

java.lang.Object

ch.cern.dirq.QueueSimple

All Implemented Interfaces:

Queue, Iterable<String>

public class QueueSimple
extends Object
implements Queue

QueueSimple - object oriented interface to a simple directory based queue.
A port of Perl module Directory::Queue::Simple http://search.cpan.org/dist/Directory-Queue/
The documentation from Directory::Queue::Simple module has been adapted for Java.

Usage

 
 // sample producer
 QueueSimple dirq = new QueueSimple("/tmp/test");
 for (int i=0; i < 100; i++) {
     String name = dirq.add("element " + i);
     System.out.println("# added element " + i + " as " + name);
 }

 // sample consumer
 dirq = QueueSimple("/tmp/test");
 for (String name: dirq) {
     if (! dirq.lock(name)) {
         continue;
     }
     System.out.println("# reading element " + name);
     String data = dirq.get(name);
     // one could use dirq.unlock(name) to only browse the queue...
     dirq.remove(name);
 }
 
 

Description

This module is very similar to the normal directory queue, but uses a different way to store data in the filesystem, using less directories. Its API is almost identical.
Compared to normal directory queue, this module:

• is simpler
• is faster
• uses less space on disk
• can be given existing files to store
• does not support schemas
• can only store and retrieve byte strings
• is not compatible (at filesystem level) with the normal directory queue

Directory Structure

The toplevel directory contains intermediate directories that contain the stored elements, each of them in a file.
The names of the intermediate directories are time based: the element insertion time is used to create a 8-digits long hexadecimal number. The granularity (see the constructor) is used to limit the number of new directories. For instance, with a granularity of 60 (the default), new directories will be created at most once per minute.
Since there is usually a filesystem limit in the number of directories a directory can hold, there is a trade-off to be made. If you want to support many added elements per second, you should use a low granularity to keep small directories. However, in this case, you will create many directories and this will limit the total number of elements you can store.
The elements themselves are stored in files (one per element) with a 14-digits long hexadecimal name SSSSSSSSMMMMMR where:

• SSSSSSSS represents the number of seconds since the Epoch
• MMMMM represents the microsecond part of the time since the Epoch
• R is a random hexadecimal digit used to reduce name collisions

A temporary element (being added to the queue) will have a .tmp suffix.
A locked element will have a hard link with the same name and the .lck suffix.
Please refer to Queue for general information about directory queues.

Author:

Lionel Cons <lionel.cons@cern.ch>, Massimo Paladin <massimo.paladin@gmail.com> Copyright (C) CERN 2012-2016

Field Summary

Fields

Modifier and Type	Field and Description
static Pattern	DIRECTORY_REGEXP
static Pattern	ELEMENT_REGEXP
static String	LOCKED_SUFFIX
static String	TEMPORARY_SUFFIX

Constructor Summary

Constructors

Constructor and Description
QueueSimple(String path) Constructor creating a simple directory queue from the given path.
QueueSimple(String path, int numask) Constructor creating a simple directory queue from the given path and umask.

Method Summary

Modifier and Type	Method and Description
String	add(byte[] data) Add byte array data to the queue.
String	add(String data) Add String data to the queue.
String	addPath(String path) Add the given file (identified by its path) to the queue and return the corresponding element name, the file must be on the same filesystem and will be moved to the queue.
int	count() Return the number of elements in the queue.
String	get(String name) Get the given locked element as String data.
byte[]	getAsByteArray(String name) Get the given locked element as byte array data.
int	getGranularity() Get the granularity.
String	getId() Return a unique identifier for the queue.
int	getMaxLock() Get the default maxLock for purge().
int	getMaxTemp() Get the default maxTemp for purge().
String	getPath(String name) Get the path of the given locked element.
String	getQueuePath() Return the path of the queue.
int	getRndHex() Get the random hexadecimal digit.
int	getUmask() Get the umask.
Iterator<String>	iterator() Iterator for the simple directory queue.
boolean	lock(String name) Lock an element in permissive mode.
boolean	lock(String name, boolean permissive) Lock an element.
void	purge() Purge the queue by removing unused intermediate directories, removing too old temporary elements and unlocking too old locked elements (aka staled locks); note: this can take a long time on queues with many elements.
void	purge(int maxLock) Purge the queue by removing unused intermediate directories, removing too old temporary elements and unlocking too old locked elements (aka staled locks); note: this can take a long time on queues with many elements.
void	purge(int maxLock, int maxTemp) Purge the queue by removing unused intermediate directories, removing too old temporary elements and unlocking too old locked elements (aka staled locks); note: this can take a long time on queues with many elements.
void	remove(String name) Remove a locked element from the queue.
QueueSimple	setGranularity(int value) Set the granularity.
QueueSimple	setMaxLock(int value) Set the default maxLock for purge().
QueueSimple	setMaxTemp(int value) Set the default maxTemp for purge().
QueueSimple	setRndHex(int value) Set the random hexadecimal digit.
QueueSimple	setUmask(int value) Set the umask.
boolean	unlock(String name) Unlock an element in non-permissive mode.
boolean	unlock(String name, boolean permissive) Unlock an element.

Methods inherited from class java.lang.Object

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

Methods inherited from interface java.lang.Iterable

forEach, spliterator

Field Detail

TEMPORARY_SUFFIX

public static final String TEMPORARY_SUFFIX

See Also:

Constant Field Values

LOCKED_SUFFIX

public static final String LOCKED_SUFFIX

See Also:

Constant Field Values

DIRECTORY_REGEXP

public static final Pattern DIRECTORY_REGEXP

ELEMENT_REGEXP

public static final Pattern ELEMENT_REGEXP

Constructor Detail

QueueSimple

public QueueSimple(String path)
            throws IOException

Constructor creating a simple directory queue from the given path.

Parameters:

path - path of the directory queue

Throws:

IOException - if any file operation fails

QueueSimple

public QueueSimple(String path,
                   int numask)
            throws IOException

Constructor creating a simple directory queue from the given path and umask.

Parameters:

path - path of the directory queue

numask - numerical umask of the directory queue

Throws:

IOException - if any file operation fails

Method Detail

getQueuePath

public String getQueuePath()

Description copied from interface: Queue

Return the path of the queue.

Specified by:

getQueuePath in interface Queue

Returns:

queue path

getId

public String getId()

Description copied from interface: Queue

Return a unique identifier for the queue.

Specified by:

getId in interface Queue

Returns:

unique queue identifier

add

public String add(String data)
           throws IOException

Description copied from interface: Queue

Add String data to the queue.

Specified by:

add in interface Queue

Parameters:

data - data to be added

Returns:

element name (as directory_name/file_name)

Throws:

IOException - if any file operation fails

add

public String add(byte[] data)
           throws IOException

Description copied from interface: Queue

Add byte array data to the queue.

Specified by:

add in interface Queue

Parameters:

data - data to be added

Returns:

element name (as directory_name/file_name)

Throws:

IOException - if any file operation fails

addPath

public String addPath(String path)
               throws IOException

Description copied from interface: Queue

Add the given file (identified by its path) to the queue and return the corresponding element name, the file must be on the same filesystem and will be moved to the queue.

Specified by:

addPath in interface Queue

Parameters:

path - path of the file to be added

Returns:

element name (as directory_name/file_name)

Throws:

IOException - if any file operation fails

get

public String get(String name)
           throws IOException

Description copied from interface: Queue

Get the given locked element as String data.

Specified by:

get in interface Queue

Parameters:

name - name of the element to be retrieved

Returns:

data associated with the given element

Throws:

IOException

getAsByteArray

public byte[] getAsByteArray(String name)
                      throws IOException

Description copied from interface: Queue

Get the given locked element as byte array data.

Specified by:

getAsByteArray in interface Queue

Parameters:

name - name of the element to be retrieved

Returns:

data associated with the given element

Throws:

IOException

getPath

public String getPath(String name)

Description copied from interface: Queue

Get the path of the given locked element.
This pathFile can be read but not removed, you must use the remove() method for this purpose.

Specified by:

getPath in interface Queue

Parameters:

name - name of the element

Returns:

path of the element

lock

public boolean lock(String name)
             throws IOException

Description copied from interface: Queue

Lock an element in permissive mode.

Specified by:

lock in interface Queue

Parameters:

name - name of the element to be locked

Returns:

true on success, false if the element could not be locked

Throws:

IOException - if any file operation fails

lock

public boolean lock(String name,
                    boolean permissive)
             throws IOException

Description copied from interface: Queue

Lock an element.

Specified by:

lock in interface Queue

Parameters:

name - name of the element to be locked

permissive - work in permissive mode

Returns:

true on success, false if the element could not be locked

Throws:

IOException - if any file operation fails

unlock

public boolean unlock(String name)
               throws IOException

Description copied from interface: Queue

Unlock an element in non-permissive mode.

Specified by:

unlock in interface Queue

Parameters:

name - name of the element to be unlocked

Returns:

true on success, false if the element could not be unlocked

Throws:

IOException - if any file operation fails

unlock

public boolean unlock(String name,
                      boolean permissive)
               throws IOException

Description copied from interface: Queue

Unlock an element.

Specified by:

unlock in interface Queue

Parameters:

name - name of the element to be unlocked

permissive - work in permissive mode

Returns:

true on success, false if the element could not be unlocked

Throws:

IOException - if any file operation fails

remove

public void remove(String name)
            throws IOException

Description copied from interface: Queue

Remove a locked element from the queue.

Specified by:

remove in interface Queue

Parameters:

name - name of the element to be removed

Throws:

IOException - if any file operation fails

count

public int count()

Description copied from interface: Queue

Return the number of elements in the queue.
Locked elements are counted but temporary elements are not.

Specified by:

count in interface Queue

Returns:

number of elements in the queue

purge

public void purge()
           throws IOException

Description copied from interface: Queue

Purge the queue by removing unused intermediate directories, removing too old temporary elements and unlocking too old locked elements (aka staled locks); note: this can take a long time on queues with many elements.
It uses default value for maxTemp and maxLock

Specified by:

purge in interface Queue

Throws:

IOException - if any file operation fails

purge

public void purge(int maxLock)
           throws IOException

Description copied from interface: Queue

Purge the queue by removing unused intermediate directories, removing too old temporary elements and unlocking too old locked elements (aka staled locks); note: this can take a long time on queues with many elements.

Specified by:

purge in interface Queue

Parameters:

maxLock - maximum time for a locked element (in seconds); if set to 0, locked elements will not be unlocked; if set to null, the object's default value will be used

Throws:

IOException - if any file operation fails

purge

public void purge(int maxLock,
                  int maxTemp)
           throws IOException

Description copied from interface: Queue

Purge the queue by removing unused intermediate directories, removing too old temporary elements and unlocking too old locked elements (aka staled locks); note: this can take a long time on queues with many elements.

Specified by:

purge in interface Queue

Parameters:

maxLock - maximum time for a locked element (in seconds); if set to 0, locked elements will not be unlocked; if set to null, the object's default value will be used

maxTemp - maximum time for a temporary element (in seconds); if set to 0, temporary elements will not be removed if set to null, the object's default value will be used

Throws:

IOException - if any file operation fails

getGranularity

public int getGranularity()

Get the granularity.

Returns:

granularity (in seconds)

setGranularity

public QueueSimple setGranularity(int value)

Set the granularity.

Parameters:

value - granularity to be set (in seconds)

Returns:

the object itself

getUmask

public int getUmask()

Get the umask.

Returns:

numerical umask

setUmask

public QueueSimple setUmask(int value)

Set the umask.

Parameters:

value - umask to be set (numerical)

Returns:

the object itself

getMaxLock

public int getMaxLock()

Get the default maxLock for purge().

Returns:

maximum lock time (in seconds)

setMaxLock

public QueueSimple setMaxLock(int value)

Set the default maxLock for purge().

Parameters:

value - maximum lock time (in seconds)

Returns:

the object itself

getMaxTemp

public int getMaxTemp()

Get the default maxTemp for purge().

Returns:

maximum temporary time (in seconds)

setMaxTemp

public QueueSimple setMaxTemp(int value)

Set the default maxTemp for purge().

Parameters:

value - maximum temporary time (in seconds)

Returns:

the object itself

getRndHex

public int getRndHex()

Get the random hexadecimal digit.

Returns:

numerical hexadecimal digit

setRndHex

public QueueSimple setRndHex(int value)

Set the random hexadecimal digit.

Parameters:

value - hexadecimal digit to be set (numerical)

Returns:

the object itself

iterator

public Iterator<String> iterator()

Iterator for the simple directory queue.

Specified by:

iterator in interface Iterable<String>