org.apache.ignite.cache.query

## Class ContinuousQuery<K,V>

- java.lang.Object
- - org.apache.ignite.cache.query.Query<javax.cache.Cache.Entry<K,V>>
  - - org.apache.ignite.cache.query.ContinuousQuery<K,V>

All Implemented Interfaces:
  Serializable

    public final class ContinuousQuery<K,V>
    extends Query<javax.cache.Cache.Entry<K,V>>

API for configuring continuous cache queries.

Continuous queries allow to register a remote filter and a local listener for cache updates. If an update event passes the filter, it will be sent to the node that executed the query and local listener will be notified.

Additionally, you can execute initial query to get currently existing data. Query can be of any type (SQL, TEXT or SCAN) and can be set via `setInitialQuery(Query)` method.

Query can be executed either on all nodes in topology using `IgniteCache.query(Query)` method, or only on the local node, if `Query.setLocal(boolean)` parameter is set to `true`. Note that in case query is distributed and a new node joins, it will get the remote filter for the query during discovery process before it actually joins topology, so no updates will be missed.

# Example

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.
     ContinuousQuery<Long, Person> qry = new ContinuousQuery<>();

     // Initial iteration query will return all persons with salary above 1000.
     qry.setInitialQuery(new ScanQuery<>((id, p) -> p.getSalary() > 1000));

     // Callback that is called locally when update notifications are received.
     // It simply prints out information about all created persons.
     qry.setLocalListener((evts) -> {
         for (CacheEntryEvent<? extends Long, ? extends Person> e : evts) {
             Person p = e.getValue();

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

     // Continuous listener will be notified for persons with salary above 1000.
     qry.setRemoteFilter(evt -> evt.getValue().getSalary() > 1000);

     // Execute query and get cursor that iterates through initial data.
     QueryCursor<Cache.Entry<Long, Person>> cur = cache.query(qry);

This will execute query on all nodes that have cache you are working with and listener will start to receive notifications for cache updates.

To stop receiving updates call `QueryCursor.close()` method:

     cur.close();

Note that this works even if you didn't provide initial query. Cursor will be empty in this case, but it will still unregister listeners when `QueryCursor.close()` is called.

See Also:
  Serialized Form

### Field Summary

Fields
`static boolean` `DFLT_AUTO_UNSUBSCRIBE` Default value for automatic unsubscription flag.
`static int` `DFLT_PAGE_SIZE` Default page size.
`static long` `DFLT_TIME_INTERVAL` Maximum default time interval after which buffer will be flushed (if buffering is enabled).

### Constructor Summary

Constructors
Constructor and Description
`ContinuousQuery()` Creates new continuous query.

### Method Summary

Methods
`Query<javax.cache.Cache.Entry<K,V>>` `getInitialQuery()` Gets initial query.
`javax.cache.event.CacheEntryUpdatedListener<K,V>` `getLocalListener()` Gets local listener.
`CacheEntryEventSerializableFilter<K,V>` `getRemoteFilter()` Gets remote filter.
`long` `getTimeInterval()` Gets time interval.
`boolean` `isAutoUnsubscribe()` Gets automatic unsubscription flag value.
`ContinuousQuery<K,V>` `setAutoUnsubscribe(boolean autoUnsubscribe)` Sets automatic unsubscribe flag.
`ContinuousQuery<K,V>` `setInitialQuery(Query<javax.cache.Cache.Entry<K,V>> initQry)` Sets initial query.
`ContinuousQuery<K,V>` `setLocal(boolean loc)` Sets whether this query should be executed on local node only.
`ContinuousQuery<K,V>` `setLocalListener(javax.cache.event.CacheEntryUpdatedListener<K,V> locLsnr)` Sets local callback.
`ContinuousQuery<K,V>` `setPageSize(int pageSize)` Sets optional page size, if `0`, then default is used.
`ContinuousQuery<K,V>` `setRemoteFilter(CacheEntryEventSerializableFilter<K,V> rmtFilter)` Sets optional key-value filter.
`ContinuousQuery<K,V>` `setTimeInterval(long timeInterval)` Sets time interval.

### Methods inherited from class org.apache.ignite.cache.query.Query

`getPageSize, isLocal, toString`

### Methods inherited from class java.lang.Object

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

### Field Detail

#### DFLT_PAGE_SIZE

    public static final int DFLT_PAGE_SIZE

Default page 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

    public 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

    public 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

### Constructor Detail

#### ContinuousQuery

    public ContinuousQuery()

Creates new continuous query.

### Method Detail

#### setInitialQuery

    public ContinuousQuery<K,V> setInitialQuery(Query<javax.cache.Cache.Entry<K,V>> initQry)

Sets initial query.

This query will be executed before continuous listener is registered which allows to iterate through entries which already existed at the time continuous query is executed.

Parameters:
  `initQry` - Initial query.
Returns:
  `this` for chaining.

#### getInitialQuery

    public Query<javax.cache.Cache.Entry<K,V>> getInitialQuery()

Gets initial query.

Returns:
  Initial query.

#### setLocalListener

    public ContinuousQuery<K,V> setLocalListener(javax.cache.event.CacheEntryUpdatedListener<K,V> locLsnr)

Sets 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:
  `locLsnr` - Local callback.
Returns:
  `this` for chaining.

#### getLocalListener

    public javax.cache.event.CacheEntryUpdatedListener<K,V> getLocalListener()

Gets local listener.

Returns:
  Local listener.

#### setRemoteFilter

    public ContinuousQuery<K,V> setRemoteFilter(CacheEntryEventSerializableFilter<K,V> rmtFilter)

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:
  `rmtFilter` - Key-value filter.
Returns:
  `this` for chaining.

#### getRemoteFilter

    public CacheEntryEventSerializableFilter<K,V> getRemoteFilter()

Gets remote filter.

Returns:
  Remote filter.

#### setTimeInterval

    public ContinuousQuery<K,V> setTimeInterval(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 `setPageSize(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.
Returns:
  `this` for chaining.

#### getTimeInterval

    public long getTimeInterval()

Gets time interval.

Returns:
  Time interval.

#### setAutoUnsubscribe

    public ContinuousQuery<K,V> setAutoUnsubscribe(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.
Returns:
  `this` for chaining.

#### setPageSize

    public ContinuousQuery<K,V> setPageSize(int pageSize)

Sets optional page size, if `0`, then default is used.

Overrides:
  `setPageSize` in class `Query<javax.cache.Cache.Entry<K,V>>`
Parameters:
  `pageSize` - Optional page size.
Returns:
  `this` for chaining.

#### setLocal

    public ContinuousQuery<K,V> setLocal(boolean loc)

Sets whether this query should be executed on local node only.

Overrides:
  `setLocal` in class `Query<javax.cache.Cache.Entry<K,V>>`
Parameters:
  `loc` - Local flag.
Returns:
  `this` for chaining.

#### isAutoUnsubscribe

    public boolean isAutoUnsubscribe()

Gets automatic unsubscription flag value.

Returns:
  Automatic unsubscription flag.

2015 Copyright © Apache Software Foundation

Follow @ApacheIgnite

Ignite Fabric | : ver. 1.3.0-incubating
Release Date | : July 10 2015