Package org.apache.nifi.processors.standard

Class WaitNotifyProtocol

java.lang.Object

org.apache.nifi.processors.standard.WaitNotifyProtocol

public class WaitNotifyProtocol
extends Object

This class provide a protocol for Wait and Notify processors to work together. Once AtomicDistributedMapCacheClient is passed to this protocol, components that wish to join the notification mechanism should only use methods provided by this protocol, instead of calling cache API directly.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static class	WaitNotifyProtocol.Signal

Field Summary

Fields

Modifier and Type	Field	Description
private final AtomicDistributedMapCacheClient	cache
static final String	CONSUMED_COUNT_NAME
static final String	DEFAULT_COUNT_NAME
private static final org.slf4j.Logger	logger
private static final int	MAX_REPLACE_RETRY_COUNT
private static final com.fasterxml.jackson.databind.ObjectMapper	objectMapper
private static final int	REPLACE_RETRY_WAIT_MILLIS
private final Deserializer<String>	stringDeserializer
private static final Serializer<String>	stringSerializer

Constructor Summary

Constructors

Constructor	Description
WaitNotifyProtocol(AtomicDistributedMapCacheClient cache)

Method Summary

Modifier and Type	Method	Description
void	complete(String signalId)	Finish protocol and remove the cache entry.
WaitNotifyProtocol.Signal	getSignal(String signalId)	Retrieve a stored Signal in the cache engine.
WaitNotifyProtocol.Signal	notify(String signalId, String counterName, int delta, Map<String,String> attributes)	Notify a signal to increase a counter.
WaitNotifyProtocol.Signal	notify(String signalId, Map<String,Integer> deltas, Map<String,String> attributes)	Notify a signal to increase a counter.
boolean	replace(WaitNotifyProtocol.Signal signal)

Methods inherited from class java.lang.Object

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

Field Details

logger

private static final org.slf4j.Logger logger

DEFAULT_COUNT_NAME

public static final String DEFAULT_COUNT_NAME

See Also:

• Constant Field Values

CONSUMED_COUNT_NAME

public static final String CONSUMED_COUNT_NAME

See Also:

• Constant Field Values

MAX_REPLACE_RETRY_COUNT

private static final int MAX_REPLACE_RETRY_COUNT

See Also:

• Constant Field Values

REPLACE_RETRY_WAIT_MILLIS

private static final int REPLACE_RETRY_WAIT_MILLIS

See Also:

• Constant Field Values

objectMapper

private static final com.fasterxml.jackson.databind.ObjectMapper objectMapper

stringSerializer

private static final Serializer<String> stringSerializer

stringDeserializer

private final Deserializer<String> stringDeserializer

cache

private final AtomicDistributedMapCacheClient cache

Constructor Details

WaitNotifyProtocol

public WaitNotifyProtocol(AtomicDistributedMapCacheClient cache)

Method Details

notify

public WaitNotifyProtocol.Signal notify(String signalId,
 Map<String,Integer> deltas,
 Map<String,String> attributes)
                                 throws IOException,
ConcurrentModificationException

Notify a signal to increase a counter.

Parameters:

signalId - a key in the underlying cache engine

deltas - a map containing counterName and delta entries, 0 has special meaning, clears the counter back to 0

attributes - attributes to save in the cache entry

Returns:

A Signal instance, merged with an existing signal if any

Throws:

IOException - thrown when it failed interacting with the cache engine

ConcurrentModificationException - thrown if other process is also updating the same signal and failed to update after few retry attempts

notify

public WaitNotifyProtocol.Signal notify(String signalId,
 String counterName,
 int delta,
 Map<String,String> attributes)
                                 throws IOException,
ConcurrentModificationException

Notify a signal to increase a counter.

Parameters:

signalId - a key in the underlying cache engine

counterName - specify count to update

delta - delta to update a counter, 0 has special meaning, clears the counter back to 0

attributes - attributes to save in the cache entry

Returns:

A Signal instance, merged with an existing signal if any

Throws:

IOException - thrown when it failed interacting with the cache engine

ConcurrentModificationException - thrown if other process is also updating the same signal and failed to update after few retry attempts

getSignal

public WaitNotifyProtocol.Signal getSignal(String signalId)
                                    throws IOException,
DeserializationException

Retrieve a stored Signal in the cache engine. If a caller gets satisfied with the returned Signal state and finish waiting, it should call complete(String) to complete the Wait Notify protocol.

Parameters:

signalId - a key in the underlying cache engine

Returns:

A Signal instance

Throws:

IOException - thrown when it failed interacting with the cache engine

DeserializationException - thrown if the cache found is not in expected serialized format

complete

public void complete(String signalId)
              throws IOException

Finish protocol and remove the cache entry.

Parameters:

signalId - a key in the underlying cache engine

Throws:

IOException - thrown when it failed interacting with the cache engine

replace

public boolean replace(WaitNotifyProtocol.Signal signal)
                throws IOException

Throws:

IOException