Package org.jumpmind.symmetric.service.impl

Class RouterService

java.lang.Object

org.jumpmind.symmetric.service.impl.AbstractService

org.jumpmind.symmetric.service.impl.RouterService

All Implemented Interfaces:

INodeCommunicationService.INodeCommunicationExecutor, IRouterService, IService

public class RouterService
extends AbstractService
implements IRouterService, INodeCommunicationService.INodeCommunicationExecutor

See Also:

• IRouterService

Field Summary

Fields

Modifier and Type	Field	Description
protected long	commonBatchesCacheTime
protected Map<String,Boolean>	commonBatchesLastKnownState
protected Map<String,Boolean>	defaultRouterOnlyLastKnownState
protected long	defaultRoutersCacheTime
protected ISymmetricEngine	engine
protected IExtensionService	extensionService
protected boolean	firstTimeCheck
protected DataGapDetector	gapDetector
protected Map<String,Boolean>	hasMaxDataRoutedByChannel
protected Map<String,CounterStat>	invalidRouterType
protected Map<String,Boolean>	isAllDataReadByChannel
protected boolean	isUsingTargetExternalId
protected Map<Integer,CounterStat>	missingColumns
protected Map<Integer,CounterStat>	missingTriggerRouter
protected ExecutorService	readThread
protected long	triggerRouterCacheTime
protected boolean	useChannelThreading

Fields inherited from class org.jumpmind.symmetric.service.impl.AbstractService

log, parameterService, platform, sqlTemplate, sqlTemplateDirty, symmetricDialect, tablePrefix

Constructor Summary

Constructors

Constructor	Description
RouterService(ISymmetricEngine engine)

Method Summary

Modifier and Type	Method	Description
protected Table	buildTableFromTriggerHistory(TriggerHistory triggerHistory)
protected void	completeBatches(ChannelRouterContext context, List<OutgoingBatch> batches, Set<IDataRouter> usedRouters)
protected void	completeBatchesAndCommit(ChannelRouterContext context)
protected boolean	doesColumnCountMatchValues(DataMetaData dataMetaData, Data data)
void	execute(NodeCommunication nodeCommunication, RemoteNodeStatus status)
protected Set<Node>	findAvailableNodes(TriggerRouter triggerRouter, ChannelRouterContext context)
void	flushCache()
List<String>	getAvailableBatchAlgorithms()	Get a list of available batch algorithms that can be used for the different channels
List<DataGap>	getDataGaps()
protected IDataRouter	getDataRouter(Router router, DataMetaData dataMetaData)
protected Set<String>	getReadyChannels()
Map<String,IDataRouter>	getRouters()
protected List<TriggerRouter>	getTriggerRoutersForData(Data data, ChannelRouterContext context)
long	getUnroutedDataCount()
protected boolean	hasMaxDataRouted()
protected int	insertDataEvents(ProcessInfo processInfo, ChannelRouterContext context, DataMetaData dataMetaData, Collection<String> nodeIds, TriggerRouter triggerRouter)
protected boolean	isAllDataRead()
protected boolean	onlyDefaultRoutersAssigned(Channel channel, String nodeGroupId, List<TriggerRouter> triggerRouters)
protected boolean	producesCommonBatches(Channel channel, String nodeGroupId, List<TriggerRouter> triggerRouters)	Deprecated.
protected String	qualifyUsingDataGaps(List<DataGap> dataGaps, int numberOfGapsToQualify, String sql)
long	routeData(boolean force)	This method will route data to specific nodes.
protected int	routeData(ProcessInfo processInfo, Data data, ChannelRouterContext context)
protected long	routeDataForChannel(ProcessInfo processInfo, NodeChannel nodeChannel, Node sourceNode, NodeCommunication nodeCommunication, ChannelRouterContext context)
protected long	routeDataForEachChannel()	We route data channel by channel for two reasons.
protected long	selectDataAndRoute(ProcessInfo processInfo, NodeCommunication nodeCommunication, ChannelRouterContext context)	Pre-read data and fill up a queue so we can peek ahead to see if we have crossed a database transaction boundary.
boolean	shouldDataBeRouted(SimpleRouterContext context, DataMetaData dataMetaData, Node node, boolean initialLoad, boolean initialLoadSelectUsed, TriggerRouter triggerRouter)	For use in data load events
protected IDataToRouteReader	startReading(ChannelRouterContext context)
void	stop()

Methods inherited from class org.jumpmind.symmetric.service.impl.AbstractService

assertNotNull, buildBatchOrderBy, buildBatchParams, buildBatchWhere, buildBatchWhereFromFilter, close, createSqlReplacementTokens, getJdbcTemplate, getParameterService, getSql, getSymmetricDialect, getTablePrefix, getTargetDialect, getTargetPlatform, getTargetPlatform, isCalledFromSymmetricAdminTool, isSet, isStreamClosedByClient, isSymmetricTable, logOnce, maxDate, readAcks, sendAck, setSqlMap, synchronize, toNodeIds, toNodeIds

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.jumpmind.symmetric.service.IService

getSql, synchronize

Field Details

missingTriggerRouter

protected Map<Integer,CounterStat> missingTriggerRouter

invalidRouterType

protected Map<String,CounterStat> invalidRouterType

missingColumns

protected Map<Integer,CounterStat> missingColumns

triggerRouterCacheTime

protected long triggerRouterCacheTime

commonBatchesLastKnownState

protected Map<String,Boolean> commonBatchesLastKnownState

commonBatchesCacheTime

protected long commonBatchesCacheTime

defaultRouterOnlyLastKnownState

protected Map<String,Boolean> defaultRouterOnlyLastKnownState

defaultRoutersCacheTime

protected long defaultRoutersCacheTime

isAllDataReadByChannel

protected Map<String,Boolean> isAllDataReadByChannel

hasMaxDataRoutedByChannel

protected Map<String,Boolean> hasMaxDataRoutedByChannel

readThread

protected transient ExecutorService readThread

engine

protected ISymmetricEngine engine

extensionService

protected IExtensionService extensionService

gapDetector

protected DataGapDetector gapDetector

firstTimeCheck

protected boolean firstTimeCheck

isUsingTargetExternalId

protected boolean isUsingTargetExternalId

useChannelThreading

protected boolean useChannelThreading

Constructor Details

RouterService

public RouterService(ISymmetricEngine engine)

Method Details

shouldDataBeRouted

public boolean shouldDataBeRouted(SimpleRouterContext context,
 DataMetaData dataMetaData,
 Node node,
 boolean initialLoad,
 boolean initialLoadSelectUsed,
 TriggerRouter triggerRouter)

For use in data load events

Specified by:

shouldDataBeRouted in interface IRouterService

stop

public void stop()

Specified by:

stop in interface IRouterService

flushCache

public void flushCache()

Specified by:

flushCache in interface IRouterService

routeData

public long routeData(boolean force)

This method will route data to specific nodes.

Specified by:

routeData in interface IRouterService

routeDataForEachChannel

protected long routeDataForEachChannel()

We route data channel by channel for two reasons. One is that if/when we decide to multi-thread the routing it is a simple matter of inserting a thread pool here and waiting for all channels to be processed. The other reason is to reduce the number of connections we are required to have.

getReadyChannels

protected Set<String> getReadyChannels()

qualifyUsingDataGaps

protected String qualifyUsingDataGaps(List<DataGap> dataGaps,
 int numberOfGapsToQualify,
 String sql)

producesCommonBatches

@Deprecated
protected boolean producesCommonBatches(Channel channel,
 String nodeGroupId,
 List<TriggerRouter> triggerRouters)

Deprecated.

onlyDefaultRoutersAssigned

protected boolean onlyDefaultRoutersAssigned(Channel channel,
 String nodeGroupId,
 List<TriggerRouter> triggerRouters)

execute

public void execute(NodeCommunication nodeCommunication,
 RemoteNodeStatus status)

Specified by:

execute in interface INodeCommunicationService.INodeCommunicationExecutor

routeDataForChannel

protected long routeDataForChannel(ProcessInfo processInfo,
 NodeChannel nodeChannel,
 Node sourceNode,
 NodeCommunication nodeCommunication,
 ChannelRouterContext context)

completeBatchesAndCommit

protected void completeBatchesAndCommit(ChannelRouterContext context)

completeBatches

protected void completeBatches(ChannelRouterContext context,
 List<OutgoingBatch> batches,
 Set<IDataRouter> usedRouters)

findAvailableNodes

protected Set<Node> findAvailableNodes(TriggerRouter triggerRouter,
 ChannelRouterContext context)

startReading

protected IDataToRouteReader startReading(ChannelRouterContext context)

selectDataAndRoute

protected long selectDataAndRoute(ProcessInfo processInfo,
 NodeCommunication nodeCommunication,
 ChannelRouterContext context)
                           throws InterruptedException

Pre-read data and fill up a queue so we can peek ahead to see if we have crossed a database transaction boundary. Then route each Data while continuing to keep the queue filled until the result set is entirely read.

Parameters:

conn - The connection to use for selecting the data.

context - The current context of the routing process

Throws:

InterruptedException

routeData

protected int routeData(ProcessInfo processInfo,
 Data data,
 ChannelRouterContext context)

insertDataEvents

protected int insertDataEvents(ProcessInfo processInfo,
 ChannelRouterContext context,
 DataMetaData dataMetaData,
 Collection<String> nodeIds,
 TriggerRouter triggerRouter)

getDataRouter

protected IDataRouter getDataRouter(Router router,
 DataMetaData dataMetaData)

getTriggerRoutersForData

protected List<TriggerRouter> getTriggerRoutersForData(Data data,
 ChannelRouterContext context)

getUnroutedDataCount

public long getUnroutedDataCount()

Specified by:

getUnroutedDataCount in interface IRouterService

getAvailableBatchAlgorithms

public List<String> getAvailableBatchAlgorithms()

Description copied from interface: IRouterService

Get a list of available batch algorithms that can be used for the different channels

Specified by:

getAvailableBatchAlgorithms in interface IRouterService

getRouters

public Map<String,IDataRouter> getRouters()

Specified by:

getRouters in interface IRouterService

getDataGaps

public List<DataGap> getDataGaps()

Specified by:

getDataGaps in interface IRouterService

buildTableFromTriggerHistory

protected Table buildTableFromTriggerHistory(TriggerHistory triggerHistory)

doesColumnCountMatchValues

protected boolean doesColumnCountMatchValues(DataMetaData dataMetaData,
 Data data)

hasMaxDataRouted

protected boolean hasMaxDataRouted()

isAllDataRead

protected boolean isAllDataRead()