org.gridgain.grid.cache.query

Interface GridCacheContinuousQuery<K,V>

All Superinterfaces:

AutoCloseable

All Known Implementing Classes:

GridCacheContinuousQueryAdapter

public interface GridCacheContinuousQuery<K,V>
extends AutoCloseable

API for configuring and executing continuous cache queries.

Continuous queries are executed as follows:

1. Query is sent to requested grid nodes. Note that for LOCAL and REPLICATED caches query will be always executed locally.
2. Each node iterates through existing cache data and registers listeners that will notify about further updates.
3. Each key-value pair is passed through optional filter and if this filter returns true, key-value pair is sent to the master node (the one that executed query). If filter is not provided, all pairs are sent.
4. When master node receives key-value pairs, it notifies the local callback.

NOTE

Under some concurrent circumstances callback may get several notifications for one cache update. This should be taken into account when implementing callback.

Query usage

As an example, suppose we have cache with 'Person' objects and we need to query all persons with salary above 1000.

Here is the Person class:

 public class Person {
     // Name.
     private String name;

     // Salary.
     private double salary;

     ...
 }
 

You can create and execute continuous query like so:

 // Create new continuous query.
 qry = cache.createContinuousQuery();

 // Callback that is called locally when update notifications are received.
 // It simply prints out information about all created persons.
 qry.callback(new GridPredicate2<UUID, Collection<Map.Entry<UUID, Person>>>() {
     @Override public boolean apply(UUID uuid, Collection<Map.Entry<UUID, Person>> entries) {
         for (Map.Entry<UUID, Person> e : entries) {
             Person p = e.getValue();

             X.println(">>>");
             X.println(">>> " + p.getFirstName() + " " + p.getLastName() +
                 "'s salary is " + p.getSalary());
             X.println(">>>");
         }

         return true;
     }
 });

 // This query will return persons with salary above 1000.
 qry.filter(new GridPredicate2<UUID, Person>() {
     @Override public boolean apply(UUID uuid, Person person) {
         return person.getSalary() > 1000;
     }
 });

 // Execute query.
 qry.execute();
 

This will execute query on all nodes that have cache you are working with and notify callback with both data that already exists in cache and further updates.

To stop receiving updates call close() method:

 qry.cancel();
 

Note that one query instance can be executed only once. After it's cancelled, it's non-operational. If you need to repeat execution, use GridCacheQueries.createContinuousQuery() method to create new query.

Field Summary

Fields

Modifier and Type	Field and Description
static boolean	DFLT_AUTO_UNSUBSCRIBE Default value for automatic unsubscription flag.
static int	DFLT_BUF_SIZE Default buffer size.
static long	DFLT_TIME_INTERVAL Maximum default time interval after which buffer will be flushed (if buffering is enabled).

Method Summary

Methods

Modifier and Type	Method and Description
void	autoUnsubscribe(boolean autoUnsubscribe) Sets automatic unsubscribe flag.
int	bufferSize() Gets buffer size.
void	bufferSize(int bufSize) Sets buffer size.
GridBiPredicate<UUID,Collection<Map.Entry<K,V>>>	callback() Gets local callback.
void	callback(GridBiPredicate<UUID,Collection<Map.Entry<K,V>>> cb) Sets mandatory local callback.
void	close() Stops continuous query execution.
void	execute() Starts continuous query execution on the whole grid.
void	execute(GridProjection prj) Starts continuous query execution on provided set of nodes.
GridBiPredicate<K,V>	filter() Gets key-value filter.
void	filter(GridBiPredicate<K,V> filter) Sets optional key-value filter.
boolean	isAutoUnsubscribe() Gets automatic unsubscribe flag.
long	timeInterval() Gets time interval.
void	timeInterval(long timeInterval) Sets time interval.

Field Detail

DFLT_BUF_SIZE

static final int DFLT_BUF_SIZE

Default buffer size. Size of 1 means that all entries will be sent to master node immediately (buffering is disabled).

See Also:

Constant Field Values

DFLT_TIME_INTERVAL

static final long DFLT_TIME_INTERVAL

Maximum default time interval after which buffer will be flushed (if buffering is enabled).

See Also:

Constant Field Values

DFLT_AUTO_UNSUBSCRIBE

static final boolean DFLT_AUTO_UNSUBSCRIBE

Default value for automatic unsubscription flag. Remote filters will be unregistered by default if master node leaves topology.

See Also:

Constant Field Values

Method Detail

callback

void callback(GridBiPredicate<UUID,Collection<Map.Entry<K,V>>> cb)

Sets mandatory local callback. This callback is called only in local node when new updates are received.

The callback predicate accepts ID of the node from where updates are received and collection of received entries. Note that for removed entries value will be null.

If the predicate returns false, query execution will be cancelled.

WARNING: all operations that involve any kind of JVM-local or distributed locking (e.g., synchronization or transactional cache operations), should be executed asynchronously without blocking the thread that called the callback. Otherwise, you can get deadlocks.

Parameters:

cb - Local callback.

callback

GridBiPredicate<UUID,Collection<Map.Entry<K,V>>> callback()

Gets local callback. See callback(GridBiPredicate) for more information.

Returns:

Local callback.

filter

void filter(@Nullable
          GridBiPredicate<K,V> filter)

Sets optional key-value filter. This filter is called before entry is sent to the master node.

WARNING: all operations that involve any kind of JVM-local or distributed locking (e.g., synchronization or transactional cache operations), should be executed asynchronously without blocking the thread that called the filter. Otherwise, you can get deadlocks.

Parameters:

filter - Key-value filter.

filter

@Nullable
GridBiPredicate<K,V> filter()

Gets key-value filter. See filter(GridBiPredicate) for more information.

Returns:

Key-value filter.

bufferSize

void bufferSize(int bufSize)

Sets buffer size.

When a cache update happens, entry is first put into a buffer. Entries from buffer will be sent to the master node only if the buffer is full or time provided via timeInterval(long) method is exceeded.

Default buffer size is 1 which means that entries will be sent immediately (buffering is disabled).

Parameters:

bufSize - Buffer size.

bufferSize

int bufferSize()

Gets buffer size. See bufferSize(int) for more information.

Returns:

Buffer size.

timeInterval

void timeInterval(long timeInterval)

Sets time interval.

When a cache update happens, entry is first put into a buffer. Entries from buffer will be sent to the master node only if the buffer is full (its size can be provided via bufferSize(int) method) or time provided via this method is exceeded.

Default time interval is 0 which means that time check is disabled and entries will be sent only when buffer is full.

Parameters:

timeInterval - Time interval.

timeInterval

long timeInterval()

Gets time interval. See timeInterval(long) for more information.

Returns:

Gets time interval.

autoUnsubscribe

void autoUnsubscribe(boolean autoUnsubscribe)

Sets automatic unsubscribe flag.

This flag indicates that query filters on remote nodes should be automatically unregistered if master node (node that initiated the query) leaves topology. If this flag is false, filters will be unregistered only when the query is cancelled from master node, and won't ever be unregistered if master node leaves grid.

Default value for this flag is true.

Parameters:

autoUnsubscribe - Automatic unsubscription flag.

isAutoUnsubscribe

boolean isAutoUnsubscribe()

Gets automatic unsubscribe flag. See autoUnsubscribe(boolean) for more information.

Returns:

Automatic unsubscribe flag.

execute

void execute()
             throws GridException

Starts continuous query execution on the whole grid.

Note that if grid contains nodes without appropriate cache, these nodes will be filtered out.

Also note that for LOCAL and REPLICATED caches query will be always executed locally.

Throws:

GridException - In case of error.

execute

void execute(@Nullable
           GridProjection prj)
             throws GridException

Starts continuous query execution on provided set of nodes.

Note that if provided projection contains nodes without appropriate cache, these nodes will be filtered out.

Also note that for LOCAL and REPLICATED caches query will be always executed locally.

Parameters:

prj - Grid projection.

Throws:

GridException - In case of error.

close

void close()
           throws GridException

Stops continuous query execution.

Note that one query instance can be executed only once. After it's cancelled, it's non-operational. If you need to repeat execution, use GridCacheQueries.createContinuousQuery() method to create new query.

Specified by:

close in interface AutoCloseable

Throws:

GridException - In case of error.