Package io.debezium.pipeline.signal

Class SignalProcessor<P extends Partition,O extends OffsetContext>

java.lang.Object

io.debezium.pipeline.signal.SignalProcessor<P,O>

public class SignalProcessor<P extends Partition,O extends OffsetContext>
extends Object

This class permits to process signals coming from the different channels.

Author:

Mario Fiore Vitale

Field Summary

Fields

Modifier and Type	Field	Description
private final CommonConnectorConfig	connectorConfig
static final String	DATA_COLLECTIONS_FIELD_NAME
private final DocumentReader	documentReader
private final List<SignalChannelReader>	enabledChannelReaders
private static final org.slf4j.Logger	LOGGER
private final Map<P,O>	partitionOffsets
static final String	POINT_REGEX
private final Semaphore	semaphore
static final int	SEMAPHORE_WAIT_TIME
private final Map<String,SignalAction<P>>	signalActions
private final List<SignalChannelReader>	signalChannelReaders
private final ScheduledExecutorService	signalProcessorExecutor

Constructor Summary

Constructors

Constructor	Description
SignalProcessor(Class<? extends org.apache.kafka.connect.source.SourceConnector> connector, CommonConnectorConfig config, Map<String,SignalAction<P>> signalActions, List<SignalChannelReader> signalChannelReaders, DocumentReader documentReader, Offsets<P,O> previousOffsets)

Method Summary

Modifier and Type	Method	Description
private void	executeSignal(SignalAction<P> action, SignalRecord signalRecord, Document jsonData, P partition, O offset)
private void	executeWithSemaphore(Runnable operation)
private Optional<P>	extractPartitionFromData(Document jsonData)
private List<SignalChannelReader>	getEnabledChannelReaders()
private Offsets<P,O>	getOffsets(P knownPartition, Document jsonData)
<T extends SignalChannelReader> T	getSignalChannel(Class<T> channel)	The method permits to get specified SignalChannelReader instance from the available SPI implementations
private Predicate<SignalChannelReader>	isEnabled()
private static <T extends SignalChannelReader> Predicate<SignalChannelReader>	isSignal(Class<T> channelClass)
private boolean	matchesDatabase(P partition, String databaseName)
void	process()
private void	processSignal(SignalRecord signalRecord, P knownPartition)
void	processSourceSignal(P partition)
void	registerSignalAction(String id, SignalAction<P> signal)
void	setContext(Offsets<P,O> offsets)
void	start()
void	stop()

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

SEMAPHORE_WAIT_TIME

public static final int SEMAPHORE_WAIT_TIME

See Also:

• Constant Field Values

DATA_COLLECTIONS_FIELD_NAME

public static final String DATA_COLLECTIONS_FIELD_NAME

See Also:

• Constant Field Values

POINT_REGEX

public static final String POINT_REGEX

See Also:

• Constant Field Values

signalActions

private final Map<String,SignalAction<P extends Partition>> signalActions

connectorConfig

private final CommonConnectorConfig connectorConfig

enabledChannelReaders

private final List<SignalChannelReader> enabledChannelReaders

signalChannelReaders

private final List<SignalChannelReader> signalChannelReaders

signalProcessorExecutor

private final ScheduledExecutorService signalProcessorExecutor

documentReader

private final DocumentReader documentReader

partitionOffsets

private final Map<P extends Partition,O extends OffsetContext> partitionOffsets

semaphore

private final Semaphore semaphore

Constructor Details

SignalProcessor

public SignalProcessor(Class<? extends org.apache.kafka.connect.source.SourceConnector> connector,
 CommonConnectorConfig config,
 Map<String,SignalAction<P>> signalActions,
 List<SignalChannelReader> signalChannelReaders,
 DocumentReader documentReader,
 Offsets<P,O> previousOffsets)

Method Details

isEnabled

private Predicate<SignalChannelReader> isEnabled()

getEnabledChannelReaders

private List<SignalChannelReader> getEnabledChannelReaders()

setContext

public void setContext(Offsets<P,O> offsets)

start

public void start()

stop

public void stop()
          throws InterruptedException

Throws:

InterruptedException

registerSignalAction

public void registerSignalAction(String id,
 SignalAction<P> signal)

process

public void process()

processSourceSignal

public void processSourceSignal(P partition)

getSignalChannel

public <T extends SignalChannelReader> T getSignalChannel(Class<T> channel)

The method permits to get specified SignalChannelReader instance from the available SPI implementations

Parameters:

channel - the class of the channel to get

Returns:

the specified instance from the available SPI implementations

executeWithSemaphore

private void executeWithSemaphore(Runnable operation)

processSignal

private void processSignal(SignalRecord signalRecord,
 P knownPartition)

getOffsets

private Offsets<P,O> getOffsets(P knownPartition,
 Document jsonData)

extractPartitionFromData

private Optional<P> extractPartitionFromData(Document jsonData)

matchesDatabase

private boolean matchesDatabase(P partition,
 String databaseName)

executeSignal

private void executeSignal(SignalAction<P> action,
 SignalRecord signalRecord,
 Document jsonData,
 P partition,
 O offset)
                    throws InterruptedException

Throws:

InterruptedException

isSignal

private static <T extends SignalChannelReader>
Predicate<SignalChannelReader> isSignal(Class<T> channelClass)